71 lines · cpp
1//===- ThreadPoolTaskDispatcher.cpp ---------------------------------------===//2//3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.4// See https://llvm.org/LICENSE.txt for license information.5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception6//7//===----------------------------------------------------------------------===//8//9// Contains the implementation of APIs in the orc-rt/ThreadPoolTaskDispatcher.h10// header.11//12//===----------------------------------------------------------------------===//13 14#include "orc-rt/ThreadPoolTaskDispatcher.h"15 16#include <cassert>17 18namespace orc_rt {19 20ThreadPoolTaskDispatcher::~ThreadPoolTaskDispatcher() {21 assert(!AcceptingTasks && "shutdown was not run");22}23 24ThreadPoolTaskDispatcher::ThreadPoolTaskDispatcher(size_t NumThreads) {25 Threads.reserve(NumThreads);26 for (size_t I = 0; I < NumThreads; ++I)27 Threads.emplace_back([this]() { taskLoop(); });28}29 30void ThreadPoolTaskDispatcher::dispatch(std::unique_ptr<Task> T) {31 {32 std::scoped_lock<std::mutex> Lock(M);33 if (!AcceptingTasks)34 return;35 PendingTasks.push_back(std::move(T));36 }37 CV.notify_one();38}39 40void ThreadPoolTaskDispatcher::shutdown() {41 {42 std::scoped_lock<std::mutex> Lock(M);43 assert(AcceptingTasks && "ThreadPoolTaskDispatcher already shut down?");44 AcceptingTasks = false;45 }46 CV.notify_all();47 for (auto &Thread : Threads)48 Thread.join();49}50 51void ThreadPoolTaskDispatcher::taskLoop() {52 while (true) {53 std::unique_ptr<Task> T;54 {55 std::unique_lock<std::mutex> Lock(M);56 CV.wait(Lock,57 [this]() { return !PendingTasks.empty() || !AcceptingTasks; });58 59 if (!AcceptingTasks && PendingTasks.empty())60 return;61 62 T = std::move(PendingTasks.back());63 PendingTasks.pop_back();64 }65 66 T->run();67 }68}69 70} // namespace orc_rt71