diff --git a/mindspore/core/mindrt/src/thread/actor_threadpool.cc b/mindspore/core/mindrt/src/thread/actor_threadpool.cc index 2427a84da48..6a6aedb4a98 100644 --- a/mindspore/core/mindrt/src/thread/actor_threadpool.cc +++ b/mindspore/core/mindrt/src/thread/actor_threadpool.cc @@ -13,7 +13,8 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - +#include +#include #include "thread/actor_threadpool.h" #include "thread/core_affinity.h" @@ -26,6 +27,7 @@ void ActorWorker::CreateThread(ActorThreadPool *pool) { } void ActorWorker::RunWithSpin() { + SetAffinity(); #if !defined(__APPLE__) && !defined(SUPPORT_MSVC) static std::atomic_int index = {0}; pthread_setname_np(pthread_self(), ("ActorThread_" + std::to_string(index++)).c_str()); @@ -116,7 +118,7 @@ void ActorThreadPool::PushActorToQueue(ActorBase *actor) { actor_queue_.push(actor); #endif } - THREAD_INFO("actor[%s] enqueue success", actor->GetAID().Name().c_str()); + THREAD_DEBUG("actor[%s] enqueue success", actor->GetAID().Name().c_str()); // active one idle actor thread if exist for (size_t i = 0; i < actor_thread_num_; ++i) { auto worker = reinterpret_cast(workers_[i]); @@ -126,11 +128,13 @@ void ActorThreadPool::PushActorToQueue(ActorBase *actor) { } } -int ActorThreadPool::CreateThreads(size_t actor_thread_num, size_t all_thread_num) { +int ActorThreadPool::CreateThreads(size_t actor_thread_num, size_t all_thread_num, const std::vector &core_list) { #ifdef USE_HQUEUE actor_queue_.Init(MAX_READY_ACTOR_NR); #endif - +#ifdef BIND_CORE + affinity_->SetCoreId(core_list); +#endif size_t core_num = std::thread::hardware_concurrency(); THREAD_INFO("ThreadInfo, Actor: [%zu], All: [%zu], CoreNum: [%zu]", actor_thread_num, all_thread_num, core_num); actor_thread_num_ = actor_thread_num < core_num ? actor_thread_num : core_num; @@ -142,27 +146,56 @@ int ActorThreadPool::CreateThreads(size_t actor_thread_num, size_t all_thread_nu std::lock_guard _l(pool_mutex_); auto worker = new (std::nothrow) ActorWorker(); THREAD_ERROR_IF_NULL(worker); +#ifdef BIND_CORE + cpu_set_t mask; + CPU_ZERO(&mask); + if (core_list.size() > 0) { + CPU_SET(core_list[workers_.size() % core_list.size()], &mask); + } + worker->set_mask(mask); +#endif worker->CreateThread(this); workers_.push_back(worker); THREAD_INFO("create actor thread[%zu]", i); } size_t kernel_thread_num = all_thread_num - actor_thread_num_; if (kernel_thread_num > 0) { - return ThreadPool::CreateThreads(kernel_thread_num); + return ThreadPool::CreateThreads(kernel_thread_num, core_list); } return THREAD_OK; } -ActorThreadPool *ActorThreadPool::CreateThreadPool(size_t actor_thread_num, size_t all_thread_num) { +ActorThreadPool *ActorThreadPool::CreateThreadPool(size_t actor_thread_num, size_t all_thread_num, BindMode bind_mode) { ActorThreadPool *pool = new (std::nothrow) ActorThreadPool(); if (pool == nullptr) { return nullptr; } - int ret = pool->CreateThreads(actor_thread_num, all_thread_num); + int ret; + std::vector core_list; +#ifdef BIND_CORE + ret = pool->InitAffinityInfo(); if (ret != THREAD_OK) { delete pool; return nullptr; } + core_list = pool->affinity_->GetCoreId(all_thread_num, bind_mode); +#endif // BIND_CORE + ret = pool->CreateThreads(actor_thread_num, all_thread_num, core_list); + if (ret != THREAD_OK) { + delete pool; + return nullptr; + } + + return pool; +} + +ActorThreadPool *ActorThreadPool::CreateThreadPool(size_t actor_thread_num, size_t all_thread_num, + const std::vector &core_list) { + ActorThreadPool *pool = new (std::nothrow) ActorThreadPool(); + if (pool == nullptr) { + return nullptr; + } + int ret; #ifdef BIND_CORE ret = pool->InitAffinityInfo(); if (ret != THREAD_OK) { @@ -170,6 +203,12 @@ ActorThreadPool *ActorThreadPool::CreateThreadPool(size_t actor_thread_num, size return nullptr; } #endif // BIND_CORE + ret = pool->CreateThreads(actor_thread_num, all_thread_num, core_list); + if (ret != THREAD_OK) { + delete pool; + return nullptr; + } + return pool; } @@ -178,7 +217,7 @@ ActorThreadPool *ActorThreadPool::CreateThreadPool(size_t thread_num) { if (pool == nullptr) { return nullptr; } - int ret = pool->CreateThreads(thread_num, thread_num); + int ret = pool->CreateThreads(thread_num, thread_num, {}); if (ret != THREAD_OK) { delete pool; return nullptr; diff --git a/mindspore/core/mindrt/src/thread/actor_threadpool.h b/mindspore/core/mindrt/src/thread/actor_threadpool.h index b588844388c..bb4bc4f57ba 100644 --- a/mindspore/core/mindrt/src/thread/actor_threadpool.h +++ b/mindspore/core/mindrt/src/thread/actor_threadpool.h @@ -18,6 +18,7 @@ #define MINDSPORE_CORE_MINDRT_RUNTIME_ACTOR_THREADPOOL_H_ #include +#include #include #include #include @@ -43,7 +44,10 @@ class ActorWorker : public Worker { class ActorThreadPool : public ThreadPool { public: // create ThreadPool that contains actor thread and kernel thread - static ActorThreadPool *CreateThreadPool(size_t actor_thread_num, size_t all_thread_num); + static ActorThreadPool *CreateThreadPool(size_t actor_thread_num, size_t all_thread_num, BindMode bind_mode); + + static ActorThreadPool *CreateThreadPool(size_t actor_thread_num, size_t all_thread_num, + const std::vector &core_list); // create ThreadPool that contains only actor thread static ActorThreadPool *CreateThreadPool(size_t thread_num); ~ActorThreadPool() override; @@ -53,7 +57,7 @@ class ActorThreadPool : public ThreadPool { private: ActorThreadPool() {} - int CreateThreads(size_t actor_thread_num, size_t all_thread_num); + int CreateThreads(size_t actor_thread_num, size_t all_thread_num, const std::vector &core_list); size_t actor_thread_num_{0}; std::mutex actor_mutex_; diff --git a/mindspore/core/mindrt/src/thread/core_affinity.cc b/mindspore/core/mindrt/src/thread/core_affinity.cc index 72417f018c7..f24f0d613cd 100644 --- a/mindspore/core/mindrt/src/thread/core_affinity.cc +++ b/mindspore/core/mindrt/src/thread/core_affinity.cc @@ -248,21 +248,31 @@ int CoreAffinity::InitHardwareCoreInfo() { return THREAD_OK; } -int CoreAffinity::InitBindCoreId(size_t thread_num, BindMode bind_mode) { +std::vector CoreAffinity::GetCoreId(size_t thread_num, BindMode bind_mode) { + std::vector bind_id; if (core_num_ != sorted_id_.size()) { THREAD_ERROR("init sorted core id failed"); - return THREAD_ERROR; + return bind_id; } - bind_id_.clear(); if (bind_mode == Power_Higher || bind_mode == Power_NoBind) { for (size_t i = 0; i < thread_num; ++i) { - bind_id_.push_back(sorted_id_[i % core_num_]); + bind_id.push_back(sorted_id_[i % core_num_]); } } else if (bind_mode == Power_Middle) { for (size_t i = 0; i < thread_num; ++i) { - bind_id_.push_back(sorted_id_[(i + higher_num_) % core_num_]); + bind_id.push_back(sorted_id_[(i + higher_num_) % core_num_]); } } else { + return bind_id; + } + return bind_id; +} +void CoreAffinity::SetCoreId(const std::vector &core_list) { bind_id_ = core_list; } + +int CoreAffinity::InitBindCoreId(size_t thread_num, BindMode bind_mode) { + bind_id_.clear(); + bind_id_ = GetCoreId(thread_num, bind_mode); + if (bind_id_.empty()) { return THREAD_ERROR; } return THREAD_OK; diff --git a/mindspore/core/mindrt/src/thread/core_affinity.h b/mindspore/core/mindrt/src/thread/core_affinity.h index 6dc3aae44ae..7138e41d131 100644 --- a/mindspore/core/mindrt/src/thread/core_affinity.h +++ b/mindspore/core/mindrt/src/thread/core_affinity.h @@ -43,6 +43,8 @@ class CoreAffinity { int BindThreads(const std::vector &workers, const std::vector &core_list); int BindThreads(const std::vector &workers, BindMode bind_mode); int BindProcess(BindMode bind_mode) const; + std::vector GetCoreId(size_t thread_num, BindMode bind_mode); + void SetCoreId(const std::vector &core_list); private: #ifdef BIND_CORE diff --git a/mindspore/core/mindrt/src/thread/threadlog.h b/mindspore/core/mindrt/src/thread/threadlog.h index 5318fa9d899..8594d852daa 100644 --- a/mindspore/core/mindrt/src/thread/threadlog.h +++ b/mindspore/core/mindrt/src/thread/threadlog.h @@ -20,14 +20,23 @@ namespace mindspore { #ifdef THREAD_POOL_DEBUG #include +#define THREAD_DEBUG(content, args...) \ + { printf("[DEBUG] %s|%d: " #content "\r\n", __func__, __LINE__, ##args); } #define THREAD_INFO(content, args...) \ { printf("[INFO] %s|%d: " #content "\r\n", __func__, __LINE__, ##args); } #define THREAD_ERROR(content, args...) \ { printf("[ERROR] %s|%d: " #content "\r\n", __func__, __LINE__, ##args); } #else +#define THREAD_DEBUG(content, ...) #define THREAD_INFO(content, ...) +#if defined(__ANDROID__) +#include +#define THREAD_ERROR(content, args...) \ + { __android_log_print(ANDROID_LOG_ERROR, "MS_LITE", "%s|%d: " #content "\r\n", __func__, __LINE__, ##args); } +#else #define THREAD_ERROR(content, ...) #endif +#endif #define THREAD_ERROR_IF_NULL(ptr) \ do { \ diff --git a/mindspore/core/mindrt/src/thread/threadpool.cc b/mindspore/core/mindrt/src/thread/threadpool.cc index 1690e91f291..1e7293f23bb 100644 --- a/mindspore/core/mindrt/src/thread/threadpool.cc +++ b/mindspore/core/mindrt/src/thread/threadpool.cc @@ -13,7 +13,8 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - +#include +#include #include "thread/threadpool.h" #include "thread/core_affinity.h" @@ -31,7 +32,28 @@ Worker::~Worker() { void Worker::CreateThread() { thread_ = std::thread(&Worker::Run, this); } +void Worker::SetAffinity() { +#ifdef BIND_CORE +#ifdef __ANDROID__ + int ret = sched_setaffinity(gettid(), sizeof(cpu_set_t), &mask_); + if (ret != THREAD_OK) { + THREAD_ERROR("bind thread %d to cpu failed. ERROR %d", gettid(), errno); + } + return; +#else +#if !defined(__APPLE__) && !defined(SUPPORT_MSVC) + int ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &mask_); + if (ret != THREAD_OK) { + THREAD_ERROR("bind thread %lu to cpu failed. ERROR %d", pthread_self(), errno); + } + return; +#endif +#endif +#endif +} + void Worker::Run() { + SetAffinity(); #if !defined(__APPLE__) && !defined(SUPPORT_MSVC) static std::atomic_int index = {0}; pthread_setname_np(pthread_self(), ("KernelThread_" + std::to_string(index++)).c_str()); @@ -105,7 +127,7 @@ ThreadPool::~ThreadPool() { THREAD_INFO("destruct success"); } -int ThreadPool::CreateThreads(size_t thread_num) { +int ThreadPool::CreateThreads(size_t thread_num, const std::vector &core_list) { size_t core_num = std::thread::hardware_concurrency(); thread_num = thread_num < core_num ? thread_num : core_num; THREAD_INFO("ThreadInfo, Num: [%zu], CoreNum: [%zu]", thread_num, core_num); @@ -117,6 +139,14 @@ int ThreadPool::CreateThreads(size_t thread_num) { for (size_t i = 0; i < thread_num; ++i) { auto worker = new (std::nothrow) Worker(); THREAD_ERROR_IF_NULL(worker); +#ifdef BIND_CORE + cpu_set_t mask; + CPU_ZERO(&mask); + if (core_list.size() > 0) { + CPU_SET(core_list[workers_.size() % core_list.size()], &mask); + } + worker->set_mask(mask); +#endif worker->CreateThread(); workers_.push_back(worker); THREAD_INFO("create kernel thread[%zu]", i); @@ -127,7 +157,7 @@ int ThreadPool::CreateThreads(size_t thread_num) { int ThreadPool::ParallelLaunch(const Func &func, Content content, int task_num) const { // distribute task to the KernelThread and the idle ActorThread, // if the task num is greater than the KernelThread num - THREAD_INFO("launch: %d", task_num); + THREAD_DEBUG("launch: %d", task_num); Task task = {func, content}; DistributeTask(&task, task_num); @@ -266,12 +296,12 @@ int ThreadPool::SetProcessAffinity(BindMode bind_mode) const { #endif // BIND_CORE } -ThreadPool *ThreadPool::CreateThreadPool(size_t thread_num) { +ThreadPool *ThreadPool::CreateThreadPool(size_t thread_num, const std::vector &core_list) { ThreadPool *pool = new (std::nothrow) ThreadPool(); if (pool == nullptr) { return nullptr; } - int ret = pool->CreateThreads(thread_num); + int ret = pool->CreateThreads(thread_num, core_list); if (ret != THREAD_OK) { delete pool; return nullptr; diff --git a/mindspore/core/mindrt/src/thread/threadpool.h b/mindspore/core/mindrt/src/thread/threadpool.h index f6b478391ac..4db2c8e4aea 100644 --- a/mindspore/core/mindrt/src/thread/threadpool.h +++ b/mindspore/core/mindrt/src/thread/threadpool.h @@ -73,16 +73,21 @@ class Worker { std::thread::id thread_id() const { return thread_.get_id(); } #ifdef BIND_CORE + void set_mask(const cpu_set_t &mask) { mask_ = mask; } pthread_t handle() { return thread_.native_handle(); } #endif protected: + void SetAffinity(); void Run(); void YieldAndDeactive(); void WaitUntilActive(); bool alive_{true}; std::thread thread_; +#ifdef BIND_CORE + cpu_set_t mask_; +#endif std::atomic_int status_{kThreadBusy}; std::mutex mutex_; @@ -98,7 +103,7 @@ class Worker { class ThreadPool { public: - static ThreadPool *CreateThreadPool(size_t thread_num); + static ThreadPool *CreateThreadPool(size_t thread_num, const std::vector &core_list = {}); virtual ~ThreadPool(); size_t thread_num() const { return workers_.size(); } @@ -112,7 +117,7 @@ class ThreadPool { protected: ThreadPool() = default; - int CreateThreads(size_t thread_num); + int CreateThreads(size_t thread_num, const std::vector &core_list); int InitAffinityInfo(); diff --git a/mindspore/lite/src/inner_context.cc b/mindspore/lite/src/inner_context.cc index bc85d724f70..b225d6b2970 100644 --- a/mindspore/lite/src/inner_context.cc +++ b/mindspore/lite/src/inner_context.cc @@ -72,16 +72,21 @@ int InnerContext::Init() { } if (this->thread_pool_ == nullptr && this->IsCpuEnabled()) { int actor_parallel_thread = this->enable_parallel_ ? kDefaultParallelNum : 1; - thread_pool_ = ActorThreadPool::CreateThreadPool(actor_parallel_thread, this->thread_num_); - if (thread_pool_ == nullptr) { - MS_LOG(ERROR) << "Create ThreadPool failed"; - return RET_NULL_PTR; - } + if (this->affinity_core_list_.empty()) { - thread_pool_->SetCpuAffinity( - static_cast(this->device_list_.front().device_info_.cpu_device_info_.cpu_bind_mode_)); + auto bind_mode = static_cast(this->device_list_.front().device_info_.cpu_device_info_.cpu_bind_mode_); + thread_pool_ = ActorThreadPool::CreateThreadPool(actor_parallel_thread, this->thread_num_, bind_mode); + if (thread_pool_ == nullptr) { + MS_LOG(ERROR) << "Create ThreadPool failed"; + return RET_NULL_PTR; + } } else { - thread_pool_->SetCpuAffinity(this->affinity_core_list_); + thread_pool_ = + ActorThreadPool::CreateThreadPool(actor_parallel_thread, this->thread_num_, this->affinity_core_list_); + if (thread_pool_ == nullptr) { + MS_LOG(ERROR) << "Create ThreadPool failed"; + return RET_NULL_PTR; + } } } if (this->allocator == nullptr) { @@ -115,7 +120,6 @@ int InnerContext::Init() { InnerContext::~InnerContext() { if (this->thread_pool_ != nullptr) { - thread_pool_->SetCpuAffinity(static_cast(NO_BIND)); delete thread_pool_; this->thread_pool_ = nullptr; }