Lang Hames
9b36e4fbc7
TaskDispatcher was only used to run wrapper-function calls that
originated from the controller. Replace it with a callable type:
Session::RunWrapperCall = move_only_function<void(
orc_rt_SessionRef, uint64_t, orc_rt_WrapperFunctionReturn,
orc_rt_WrapperFunction, WrapperFunctionBuffer)>
Each call carries an outstanding ManagedCodeTaskGroup token; the runner
must eventually invoke Fn (which calls Return) or call Return directly
to bail out, otherwise Session shutdown blocks indefinitely.
Clients can supply any callable that satisfies the contract above. The
new QueueingRunner and ThreadPoolRunner classes (replacing
QueueingTaskDispatcher and ThreadPoolTaskDispatcher, respectively) are
provided as off-the-shelf options.
70 lines
1.9 KiB
C++
70 lines
1.9 KiB
C++
//===- ThreadPoolRunner.cpp -----------------------------------------------===//
|
|
//
|
|
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
|
|
// See https://llvm.org/LICENSE.txt for license information.
|
|
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
|
|
//
|
|
//===----------------------------------------------------------------------===//
|
|
//
|
|
// Contains the implementation of APIs in the orc-rt/ThreadPoolRunner.h
|
|
// header.
|
|
//
|
|
//===----------------------------------------------------------------------===//
|
|
|
|
#include "orc-rt/ThreadPoolRunner.h"
|
|
|
|
#include <cassert>
|
|
|
|
namespace orc_rt {
|
|
|
|
ThreadPoolRunner::ThreadPoolRunner(size_t NumThreads) {
|
|
Workers.reserve(NumThreads);
|
|
for (size_t I = 0; I < NumThreads; ++I)
|
|
Workers.emplace_back([this]() { workerLoop(); });
|
|
}
|
|
|
|
ThreadPoolRunner::~ThreadPoolRunner() {
|
|
{
|
|
std::scoped_lock<std::mutex> Lock(M);
|
|
Stop = true;
|
|
}
|
|
CV.notify_all();
|
|
for (auto &Worker : Workers)
|
|
Worker.join();
|
|
}
|
|
|
|
void ThreadPoolRunner::operator()(orc_rt_SessionRef S, uint64_t CallId,
|
|
orc_rt_WrapperFunctionReturn Return,
|
|
orc_rt_WrapperFunction Fn,
|
|
WrapperFunctionBuffer ArgBytes) {
|
|
{
|
|
std::scoped_lock<std::mutex> Lock(M);
|
|
assert(!Stop &&
|
|
"operator() called on ThreadPoolRunner after destruction begun");
|
|
Pending.push_back([=, ArgBytes = std::move(ArgBytes)]() mutable {
|
|
Fn(S, CallId, Return, ArgBytes.release());
|
|
});
|
|
}
|
|
CV.notify_one();
|
|
}
|
|
|
|
void ThreadPoolRunner::workerLoop() {
|
|
while (true) {
|
|
move_only_function<void()> Call;
|
|
{
|
|
std::unique_lock<std::mutex> Lock(M);
|
|
CV.wait(Lock, [this]() { return !Pending.empty() || Stop; });
|
|
|
|
if (Pending.empty() && Stop)
|
|
return;
|
|
|
|
Call = std::move(Pending.back());
|
|
Pending.pop_back();
|
|
}
|
|
|
|
Call();
|
|
}
|
|
}
|
|
|
|
} // namespace orc_rt
|