text: /**
   * Copyright 2021 Huawei Technologies Co., Ltd
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  #ifndef _MSC_VER
  #include <sched.h>
  #include <unistd.h>
  #endif
  #include "thread/actor_threadpool.h"
  #include "thread/core_affinity.h"
  
  namespace mindspore {
  constexpr size_t MAX_READY_ACTOR_NR = 4096;
  void ActorWorker::CreateThread(ActorThreadPool *pool) {
    THREAD_RETURN_IF_NULL(pool);
    pool_ = pool;
    thread_ = std::thread(&ActorWorker::RunWithSpin, this);
  }
  
  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());
  #endif
    while (alive_) {
      // only run either local KernelTask or PoolQueue ActorTask
      if (RunLocalKernelTask() || RunQueueActorTask()) {
        spin_count_ = 0;
      } else {
        YieldAndDeactive();
      }
      if (spin_count_ >= kDefaultSpinCount) {
        WaitUntilActive();
        spin_count_ = 0;
      }
    }
  }
  
  bool ActorWorker::RunQueueActorTask() {
    THREAD_ERROR_IF_NULL(pool_);
    auto actor = pool_->PopActorFromQueue();
    if (actor == nullptr) {
      return false;
    }
    actor->Run();
    return true;
  }
  
  bool ActorWorker::Active() {
    if (status_ != kThreadIdle) {
      return false;
    }
    status_ = kThreadBusy;
    cond_var_.notify_one();
    return true;
  }
  
  ActorThreadPool::~ActorThreadPool() {
    // wait until actor queue is empty
    bool terminate = false;
    do {
      {
  #ifdef USE_HQUEUE
        terminate = actor_queue_.Empty();
  #else
        std::lock_guard<std::mutex> _l(actor_mutex_);
        terminate = actor_queue_.empty();
  #endif
      }
      if (!terminate) {
        std::this_thread::yield();
      }
    } while (!terminate);
    for (auto &worker : workers_) {
      delete worker;
      worker = nullptr;
    }
    workers_.clear();
  #ifdef USE_HQUEUE
    actor_queue_.Clean();
  #endif
  }
  
  ActorBase *ActorThreadPool::PopActorFromQueue() {
  #ifdef USE_HQUEUE
    return actor_queue_.Dequeue();
  #else
    std::lock_guard<std::mutex> _l(actor_mutex_);
    if (actor_queue_.empty()) {
      return nullptr;
    }
    auto actor = actor_queue_.front();
    actor_queue_.pop();
    return actor;
  #endif
  }
  
  void ActorThreadPool::PushActorToQueue(ActorBase *actor) {
    if (!actor) {
      return;
    }
    {
  #ifdef USE_HQUEUE
      while (!actor_queue_.Enqueue(actor)) {
      }
  #else
      std::lock_guard<std::mutex> _l(actor_mutex_);
      actor_queue_.push(actor);
  #endif
    }
    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<ActorWorker *>(workers_[i]);
      if (worker->Active()) {
        break;
      }
    }
  }
  
  int ActorThreadPool::CreateThreads(size_t actor_thread_num, size_t all_thread_num, const std::vector<int> &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;
    if (actor_thread_num > all_thread_num) {
      THREAD_ERROR("thread num is invalid");
      return THREAD_ERROR;
    }
    for (size_t i = 0; i < actor_thread_num_; ++i) {
      std::lock_guard<std::mutex> _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, core_list);
    }
    return THREAD_OK;
  }
  
  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;
    std::vector<int> 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<int> &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) {
      delete pool;
      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;
  }
  
  ActorThreadPool *ActorThreadPool::CreateThreadPool(size_t thread_num) {
    ActorThreadPool *pool = new (std::nothrow) ActorThreadPool();
    if (pool == nullptr) {
      return nullptr;
    }
    int ret = pool->CreateThreads(thread_num, thread_num, {});
    if (ret != THREAD_OK) {
      delete pool;
      return nullptr;
    }
    return pool;
  }
  }  // namespace mindspore
  