| 36 | } |
| 37 | |
| 38 | void ThreadMgr::AddThreads(const HashSet<int64_t>& thread_ids) { |
| 39 | const int64_t this_rank = GlobalProcessCtx::Rank(); |
| 40 | for (int64_t thrd_id : thread_ids) { |
| 41 | const auto& it = threads_.find(thrd_id); |
| 42 | if (it != threads_.end()) { |
| 43 | // NOTE(chengcheng): check thread is not null. |
| 44 | CHECK(it->second) << " RuntimeError! Thread: " << thrd_id << " in manager must be NOT null."; |
| 45 | VLOG(1) << " Actor thread: " << thrd_id << " reused."; |
| 46 | continue; |
| 47 | } |
| 48 | StreamId stream_id = DecodeStreamIdFromInt64(thrd_id); |
| 49 | if (stream_id.rank() != this_rank) { continue; } |
| 50 | Thread* thread = new Thread(stream_id); |
| 51 | CHECK_NOTNULL(thread); |
| 52 | threads_[thrd_id].reset(thread); |
| 53 | VLOG(1) << " Actor thread: " << thrd_id << " created."; |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | void ThreadMgr::DeleteThreads(const HashSet<int64_t>& thread_ids) { |
| 58 | std::unique_lock<std::mutex> lock(mutex4del_threads_); |