diff --git a/Common/Thread/ThreadManager.cpp b/Common/Thread/ThreadManager.cpp index a98f78ced5..acfcab7040 100644 --- a/Common/Thread/ThreadManager.cpp +++ b/Common/Thread/ThreadManager.cpp @@ -55,6 +55,13 @@ ThreadManager::ThreadManager() : global_(new GlobalThreadContext()) { } ThreadManager::~ThreadManager() { + // Make sure the worker threads are stopped and joined before we free global_ - + // otherwise, if Teardown() was never called (e.g. an early return between Init() + // and the caller's normal shutdown path), the still-running threads would go on + // touching global_'s mutexes/queues/condition variables after they're freed here. + if (IsInitialized()) { + Teardown(); + } delete global_; } @@ -65,68 +72,48 @@ void ThreadManager::Teardown() { threadCtx->cond.notify_one(); } - // Purge any cancellable tasks while the threads shut down. - if (global_->compute_queue_size > 0 || global_->io_queue_size > 0) { - auto drainQueue = [this](std::deque queue[TASK_PRIORITY_COUNT], std::atomic &size) { - for (size_t i = 0; i < TASK_PRIORITY_COUNT; ++i) { - for (auto it = queue[i].begin(); it != queue[i].end(); ++it) { - if (TeardownTask(*it, false)) { - queue[i].erase(it); - size--; - return false; - } - } - } - return true; - }; - + // Once cancelled is set, no worker thread will pick up any more tasks from either + // queue, so anything still sitting in the global queues at this point will never + // run. Drain them here so they're properly cancelled/released instead of leaking + // when global_ is destroyed. + { std::unique_lock lock(global_->mutex); - while (!drainQueue(global_->compute_queue, global_->compute_queue_size)) - continue; - while (!drainQueue(global_->io_queue, global_->io_queue_size)) - continue; + for (size_t i = 0; i < TASK_PRIORITY_COUNT; ++i) { + for (Task *task : global_->compute_queue[i]) + TeardownTask(task); + global_->compute_queue[i].clear(); + for (Task *task : global_->io_queue[i]) + TeardownTask(task); + global_->io_queue[i].clear(); + } + global_->compute_queue_size = 0; + global_->io_queue_size = 0; } for (TaskThreadContext *&threadCtx : global_->threads_) { threadCtx->thread.join(); - // TODO: Is it better to just delete these? + // Same reasoning as above - anything left in a thread's private queue at this + // point will never run, since the thread has already exited. for (size_t i = 0; i < TASK_PRIORITY_COUNT; ++i) { for (Task *task : threadCtx->private_queue[i]) { - TeardownTask(task, true); + TeardownTask(task); } } delete threadCtx; } global_->threads_.clear(); - - if (global_->compute_queue_size > 0 || global_->io_queue_size > 0) { - WARN_LOG(Log::System, "ThreadManager::Teardown() with tasks still enqueued"); - } } -bool ThreadManager::TeardownTask(Task *task, bool enqueue) { +void ThreadManager::TeardownTask(Task *task) { if (!task) - return true; + return; if (task->Cancellable()) { task->Cancel(); - task->Release(); - return true; + } else { + WARN_LOG(Log::System, "ThreadManager::Teardown() dropping a non-cancellable task that will never run"); } - - if (enqueue) { - size_t queueIndex = (size_t)task->Priority(); - if (task->Type() == TaskType::CPU_COMPUTE) { - global_->compute_queue[queueIndex].push_back(task); - global_->compute_queue_size++; - } else if (task->Type() == TaskType::IO_BLOCKING) { - global_->io_queue[queueIndex].push_back(task); - global_->io_queue_size++; - } else { - _assert_(false); - } - } - return false; + task->Release(); } static void WorkerThreadFunc(GlobalThreadContext *global, TaskThreadContext *thread) { @@ -155,8 +142,8 @@ static void WorkerThreadFunc(GlobalThreadContext *global, TaskThreadContext *thr if (global_queue_size() > 0) { // Grab one from the global queue if there is any. std::unique_lock lock(global->mutex); - auto queue = isCompute ? global->compute_queue : global->io_queue; - auto &queue_size = isCompute ? global->compute_queue_size : global->io_queue_size; + std::deque *queue = isCompute ? global->compute_queue : global->io_queue; + std::atomic &queue_size = isCompute ? global->compute_queue_size : global->io_queue_size; for (size_t p = 0; p < TASK_PRIORITY_COUNT; ++p) { if (!queue[p].empty()) { @@ -169,6 +156,7 @@ static void WorkerThreadFunc(GlobalThreadContext *global, TaskThreadContext *thr break; } else if (thread->queue_size != 0) { // Check the thread, as we prefer a HIGH thread task to a global NORMAL task. + // NOTE: lock order is global->mutex before thread->mutex - never lock them in the opposite order. std::unique_lock lock(thread->mutex); if (!thread->private_queue[p].empty()) { task = thread->private_queue[p].front(); @@ -262,6 +250,8 @@ void ThreadManager::EnqueueTask(Task *task) { maxThread = numThreads_; } + _assert_msg_(maxThread > minThread, "ThreadManager: no threads available for task type %d (was Init() called with 0 compute threads?)", (int)task->Type()); + // Find a thread with no outstanding work. _assert_(maxThread <= (int)global_->threads_.size()); for (int threadNum = minThread; threadNum < maxThread; threadNum++) { @@ -318,10 +308,6 @@ int ThreadManager::GetNumLooperThreads() const { return numComputeThreads_; } -void ThreadManager::TryCancelTask(uint64_t taskID) { - // Do nothing for now, just let it finish. -} - bool ThreadManager::IsInitialized() const { return !global_->threads_.empty(); } diff --git a/Common/Thread/ThreadManager.h b/Common/Thread/ThreadManager.h index 6afa564e85..e41b83a446 100644 --- a/Common/Thread/ThreadManager.h +++ b/Common/Thread/ThreadManager.h @@ -28,9 +28,7 @@ public: virtual void Run() = 0; virtual bool Cancellable() const { return false; } virtual void Cancel() {} - virtual uint64_t id() { return 0; } virtual void Release() { delete this; } - virtual const char *Kind() const { return nullptr; } // Useful for selecting task by some qualifier, like, waiting for them all. }; class Waitable { @@ -58,23 +56,20 @@ public: // just ignore it and let the OS handle it. void Init(int numCores, int numLogicalCoresPerCpu); void EnqueueTask(Task *task); - // Use enforceSequence if this must run after all previously queued tasks. + // Directly assigns a task to a specific worker thread's private queue, bypassing + // the global queue and load balancing. Used to pin related tasks (e.g. tiles of a + // parallel loop) to distinct threads by index. void EnqueueTaskOnThread(int threadNum, Task *task); void Teardown(); bool IsInitialized() const; - // Currently does nothing. It will always be best-effort - maybe it cancels, - // maybe it doesn't. Note that the id is the id() returned by the task. You need to make that - // something meaningful yourself. - void TryCancelTask(uint64_t id); - // Parallel loops (assumed compute-limited) get one thread per logical core. We have a few extra threads too // for I/O bounds tasks, that can be run concurrently with those. int GetNumLooperThreads() const; private: - bool TeardownTask(Task *task, bool enqueue); + void TeardownTask(Task *task); // This is always pointing to a context, initialized in the constructor. GlobalThreadContext *global_;