!15457 mindrt thread init : singleton -> append

From: @ling_qiao_min
Reviewed-by: @zhang_xue_tong,@zhanghaibo5
Signed-off-by: @zhang_xue_tong

mindspore/core/mindrt/src/actor/actorthread.cc

@@ -15,32 +15,58 @@
  */
 
 #include "actor/actorthread.h"
+#ifdef __WIN32__
+#include <windows.h>
+#else
+#include <unistd.h>
+#endif
 #include <atomic>
 #include <utility>
 #include <memory>
 
 namespace mindspore {
 constexpr int MAXTHREADNAMELEN = 12;
+
+size_t GetMaxThreadCount() {
+  size_t max_num;
+#ifdef __WIN32__
+  SYSTEM_INFO sys_info;
+  GetSystemInfo(&sys_info);
+  max_num = sys_info.dwNumberOfProcessors;
+#else
+  max_num = sysconf(_SC_NPROCESSORS_ONLN);
+#endif
+  return max_num;
+}
+
 ActorThread::ActorThread() : readyActors(), workers() {
   readyActors.clear();
   workers.clear();
 
-  char *envThreadName = getenv("LITEBUS_THREAD_NAME");
+  char *envThreadName = getenv("MINDRT_THREAD_NAME");
   if (envThreadName != nullptr) {
     threadName = envThreadName;
     if (threadName.size() > MAXTHREADNAMELEN) {
       threadName.resize(MAXTHREADNAMELEN);
     }
   } else {
-    threadName = "HARES_LB_ACT";
+    threadName = "HARES_MINDRT_ACT";
   }
+
+  maxThreads_ = GetMaxThreadCount();
 }
 
 ActorThread::~ActorThread() {}
 void ActorThread::AddThread(int threadCount) {
+  std::unique_lock<std::mutex> lock(initLock_);
   for (int i = 0; i < threadCount; ++i) {
+    if (workers.size() >= maxThreads_) {
+      MS_LOG(DEBUG) << "threads number in mindrt reach upper limit. maxThreads:" << maxThreads_;
+      break;
+    }
     std::unique_ptr<std::thread> worker(new (std::nothrow) std::thread(&ActorThread::Run, this));
     BUS_OOM_EXIT(worker);
+
     workers.push_back(std::move(worker));
   }
 }

mindspore/core/mindrt/src/actor/actorthread.h

@@ -45,6 +45,9 @@ class ActorThread {
 
   std::list<std::unique_ptr<std::thread>> workers;
   std::string threadName;
+
+  size_t maxThreads_;
+  std::mutex initLock_;
 };
 
 };  // end of namespace mindspore

mindspore/core/mindrt/src/litebus.cc

@@ -83,15 +83,8 @@ int InitializeImp(const std::string &tcpUrl, const std::string &tcpUrlAdv, const
 
 int Initialize(const std::string &tcpUrl, const std::string &tcpUrlAdv, const std::string &udpUrl,
                const std::string &udpUrlAdv, int threadCount) {
-  static std::atomic_bool initLitebusStatus(false);
-  bool inite = false;
-  if (initLitebusStatus.compare_exchange_strong(inite, true) == false) {
-    ICTSBASE_LOG0(ICTSBASE_LOG_COMMON_CODE, HLOG_LEVEL_INFO, PID_LITEBUS_LOG, "litebus has been initialized");
-    return BUS_OK;
-  }
-
-  int result = BUS_OK;
-  result = InitializeImp(tcpUrl, tcpUrlAdv, udpUrl, udpUrlAdv, threadCount);
+  /* support repeat initialize  */
+  int result = InitializeImp(tcpUrl, tcpUrlAdv, udpUrl, udpUrlAdv, threadCount);
   static LiteBusExit busExit;
 
   return result;

mindspore/lite/src/lite_mindrt.cc

@@ -65,7 +65,7 @@ void LiteOpActor::SetOutputData(OpContext<Tensor> *context) {
   }
 }
 
-int MindrtInit() { return mindspore::Initialize("tcp://127.0.0.1:8080", "", "", "", 2); }
+int MindrtInit() { return mindspore::Initialize("tcp://127.0.0.1:8080", "", "", "", 1); }
 
 void MindrtTerminate(std::vector<std::shared_ptr<LiteOpActor>> actor_list) {
   for (auto actor : actor_list) {