Revision f4e02c46fc2b07836f6e35b92681f043f26f2361 authored by Ben Boeckel on 03 February 2020, 20:31:15 UTC, committed by Ben Boeckel on 03 February 2020, 20:31:26 UTC
1 parent 5f04dfe
cmWorkerPool.cxx
/* Distributed under the OSI-approved BSD 3-Clause License. See accompanying
file Copyright.txt or https://cmake.org/licensing for details. */
#include "cmWorkerPool.h"
#include <algorithm>
#include <array>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <functional>
#include <mutex>
#include <thread>
#include <cm/memory>
#include "cm_uv.h"
#include "cmRange.h"
#include "cmStringAlgorithms.h"
#include "cmUVHandlePtr.h"
#include "cmUVSignalHackRAII.h" // IWYU pragma: keep
/**
* @brief libuv pipe buffer class
*/
class cmUVPipeBuffer
{
public:
using DataRange = cmRange<const char*>;
using DataFunction = std::function<void(DataRange)>;
/// On error the ssize_t argument is a non zero libuv error code
using EndFunction = std::function<void(ssize_t)>;
public:
/**
* Reset to construction state
*/
void reset();
/**
* Initializes uv_pipe(), uv_stream() and uv_handle()
* @return true on success
*/
bool init(uv_loop_t* uv_loop);
/**
* Start reading
* @return true on success
*/
bool startRead(DataFunction dataFunction, EndFunction endFunction);
//! libuv pipe
uv_pipe_t* uv_pipe() const { return UVPipe_.get(); }
//! uv_pipe() casted to libuv stream
uv_stream_t* uv_stream() const { return static_cast<uv_stream_t*>(UVPipe_); }
//! uv_pipe() casted to libuv handle
uv_handle_t* uv_handle() { return static_cast<uv_handle_t*>(UVPipe_); }
private:
// -- Libuv callbacks
static void UVAlloc(uv_handle_t* handle, size_t suggestedSize,
uv_buf_t* buf);
static void UVData(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf);
private:
cm::uv_pipe_ptr UVPipe_;
std::vector<char> Buffer_;
DataFunction DataFunction_;
EndFunction EndFunction_;
};
void cmUVPipeBuffer::reset()
{
if (UVPipe_.get() != nullptr) {
EndFunction_ = nullptr;
DataFunction_ = nullptr;
Buffer_.clear();
Buffer_.shrink_to_fit();
UVPipe_.reset();
}
}
bool cmUVPipeBuffer::init(uv_loop_t* uv_loop)
{
reset();
if (uv_loop == nullptr) {
return false;
}
int ret = UVPipe_.init(*uv_loop, 0, this);
return (ret == 0);
}
bool cmUVPipeBuffer::startRead(DataFunction dataFunction,
EndFunction endFunction)
{
if (UVPipe_.get() == nullptr) {
return false;
}
if (!dataFunction || !endFunction) {
return false;
}
DataFunction_ = std::move(dataFunction);
EndFunction_ = std::move(endFunction);
int ret = uv_read_start(uv_stream(), &cmUVPipeBuffer::UVAlloc,
&cmUVPipeBuffer::UVData);
return (ret == 0);
}
void cmUVPipeBuffer::UVAlloc(uv_handle_t* handle, size_t suggestedSize,
uv_buf_t* buf)
{
auto& pipe = *reinterpret_cast<cmUVPipeBuffer*>(handle->data);
pipe.Buffer_.resize(suggestedSize);
buf->base = pipe.Buffer_.data();
buf->len = static_cast<unsigned long>(pipe.Buffer_.size());
}
void cmUVPipeBuffer::UVData(uv_stream_t* stream, ssize_t nread,
const uv_buf_t* buf)
{
auto& pipe = *reinterpret_cast<cmUVPipeBuffer*>(stream->data);
if (nread > 0) {
if (buf->base != nullptr) {
// Call data function
pipe.DataFunction_(DataRange(buf->base, buf->base + nread));
}
} else if (nread < 0) {
// Save the end function on the stack before resetting the pipe
EndFunction efunc;
efunc.swap(pipe.EndFunction_);
// Reset pipe before calling the end function
pipe.reset();
// Call end function
efunc((nread == UV_EOF) ? 0 : nread);
}
}
/**
* @brief External process management class
*/
class cmUVReadOnlyProcess
{
public:
// -- Types
//! @brief Process settings
struct SetupT
{
std::string WorkingDirectory;
std::vector<std::string> Command;
cmWorkerPool::ProcessResultT* Result = nullptr;
bool MergedOutput = false;
};
public:
// -- Const accessors
SetupT const& Setup() const { return Setup_; }
cmWorkerPool::ProcessResultT* Result() const { return Setup_.Result; }
bool IsStarted() const { return IsStarted_; }
bool IsFinished() const { return IsFinished_; }
// -- Runtime
void setup(cmWorkerPool::ProcessResultT* result, bool mergedOutput,
std::vector<std::string> const& command,
std::string const& workingDirectory = std::string());
bool start(uv_loop_t* uv_loop, std::function<void()> finishedCallback);
private:
// -- Libuv callbacks
static void UVExit(uv_process_t* handle, int64_t exitStatus, int termSignal);
void UVPipeOutData(cmUVPipeBuffer::DataRange data);
void UVPipeOutEnd(ssize_t error);
void UVPipeErrData(cmUVPipeBuffer::DataRange data);
void UVPipeErrEnd(ssize_t error);
void UVTryFinish();
private:
// -- Setup
SetupT Setup_;
// -- Runtime
bool IsStarted_ = false;
bool IsFinished_ = false;
std::function<void()> FinishedCallback_;
std::vector<const char*> CommandPtr_;
std::array<uv_stdio_container_t, 3> UVOptionsStdIO_;
uv_process_options_t UVOptions_;
cm::uv_process_ptr UVProcess_;
cmUVPipeBuffer UVPipeOut_;
cmUVPipeBuffer UVPipeErr_;
};
void cmUVReadOnlyProcess::setup(cmWorkerPool::ProcessResultT* result,
bool mergedOutput,
std::vector<std::string> const& command,
std::string const& workingDirectory)
{
Setup_.WorkingDirectory = workingDirectory;
Setup_.Command = command;
Setup_.Result = result;
Setup_.MergedOutput = mergedOutput;
}
bool cmUVReadOnlyProcess::start(uv_loop_t* uv_loop,
std::function<void()> finishedCallback)
{
if (IsStarted() || (Result() == nullptr)) {
return false;
}
// Reset result before the start
Result()->reset();
// Fill command string pointers
if (!Setup().Command.empty()) {
CommandPtr_.reserve(Setup().Command.size() + 1);
for (std::string const& arg : Setup().Command) {
CommandPtr_.push_back(arg.c_str());
}
CommandPtr_.push_back(nullptr);
} else {
Result()->ErrorMessage = "Empty command";
}
if (!Result()->error()) {
if (!UVPipeOut_.init(uv_loop)) {
Result()->ErrorMessage = "libuv stdout pipe initialization failed";
}
}
if (!Result()->error()) {
if (!UVPipeErr_.init(uv_loop)) {
Result()->ErrorMessage = "libuv stderr pipe initialization failed";
}
}
if (!Result()->error()) {
// -- Setup process stdio options
// stdin
UVOptionsStdIO_[0].flags = UV_IGNORE;
UVOptionsStdIO_[0].data.stream = nullptr;
// stdout
UVOptionsStdIO_[1].flags =
static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
UVOptionsStdIO_[1].data.stream = UVPipeOut_.uv_stream();
// stderr
UVOptionsStdIO_[2].flags =
static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
UVOptionsStdIO_[2].data.stream = UVPipeErr_.uv_stream();
// -- Setup process options
std::fill_n(reinterpret_cast<char*>(&UVOptions_), sizeof(UVOptions_), 0);
UVOptions_.exit_cb = &cmUVReadOnlyProcess::UVExit;
UVOptions_.file = CommandPtr_[0];
UVOptions_.args = const_cast<char**>(CommandPtr_.data());
UVOptions_.cwd = Setup_.WorkingDirectory.c_str();
UVOptions_.flags = UV_PROCESS_WINDOWS_HIDE;
UVOptions_.stdio_count = static_cast<int>(UVOptionsStdIO_.size());
UVOptions_.stdio = UVOptionsStdIO_.data();
// -- Spawn process
int uvErrorCode = UVProcess_.spawn(*uv_loop, UVOptions_, this);
if (uvErrorCode != 0) {
Result()->ErrorMessage = "libuv process spawn failed";
if (const char* uvErr = uv_strerror(uvErrorCode)) {
Result()->ErrorMessage += ": ";
Result()->ErrorMessage += uvErr;
}
}
}
// -- Start reading from stdio streams
if (!Result()->error()) {
if (!UVPipeOut_.startRead(
[this](cmUVPipeBuffer::DataRange range) {
this->UVPipeOutData(range);
},
[this](ssize_t error) { this->UVPipeOutEnd(error); })) {
Result()->ErrorMessage = "libuv start reading from stdout pipe failed";
}
}
if (!Result()->error()) {
if (!UVPipeErr_.startRead(
[this](cmUVPipeBuffer::DataRange range) {
this->UVPipeErrData(range);
},
[this](ssize_t error) { this->UVPipeErrEnd(error); })) {
Result()->ErrorMessage = "libuv start reading from stderr pipe failed";
}
}
if (!Result()->error()) {
IsStarted_ = true;
FinishedCallback_ = std::move(finishedCallback);
} else {
// Clear libuv handles and finish
UVProcess_.reset();
UVPipeOut_.reset();
UVPipeErr_.reset();
CommandPtr_.clear();
}
return IsStarted();
}
void cmUVReadOnlyProcess::UVExit(uv_process_t* handle, int64_t exitStatus,
int termSignal)
{
auto& proc = *reinterpret_cast<cmUVReadOnlyProcess*>(handle->data);
if (proc.IsStarted() && !proc.IsFinished()) {
// Set error message on demand
proc.Result()->ExitStatus = exitStatus;
proc.Result()->TermSignal = termSignal;
if (!proc.Result()->error()) {
if (termSignal != 0) {
proc.Result()->ErrorMessage = cmStrCat(
"Process was terminated by signal ", proc.Result()->TermSignal);
} else if (exitStatus != 0) {
proc.Result()->ErrorMessage = cmStrCat(
"Process failed with return value ", proc.Result()->ExitStatus);
}
}
// Reset process handle
proc.UVProcess_.reset();
// Try finish
proc.UVTryFinish();
}
}
void cmUVReadOnlyProcess::UVPipeOutData(cmUVPipeBuffer::DataRange data)
{
Result()->StdOut.append(data.begin(), data.end());
}
void cmUVReadOnlyProcess::UVPipeOutEnd(ssize_t error)
{
// Process pipe error
if ((error != 0) && !Result()->error()) {
Result()->ErrorMessage = cmStrCat(
"Reading from stdout pipe failed with libuv error code ", error);
}
// Try finish
UVTryFinish();
}
void cmUVReadOnlyProcess::UVPipeErrData(cmUVPipeBuffer::DataRange data)
{
std::string* str =
Setup_.MergedOutput ? &Result()->StdOut : &Result()->StdErr;
str->append(data.begin(), data.end());
}
void cmUVReadOnlyProcess::UVPipeErrEnd(ssize_t error)
{
// Process pipe error
if ((error != 0) && !Result()->error()) {
Result()->ErrorMessage = cmStrCat(
"Reading from stderr pipe failed with libuv error code ", error);
}
// Try finish
UVTryFinish();
}
void cmUVReadOnlyProcess::UVTryFinish()
{
// There still might be data in the pipes after the process has finished.
// Therefore check if the process is finished AND all pipes are closed
// before signaling the worker thread to continue.
if ((UVProcess_.get() != nullptr) || (UVPipeOut_.uv_pipe() != nullptr) ||
(UVPipeErr_.uv_pipe() != nullptr)) {
return;
}
IsFinished_ = true;
FinishedCallback_();
}
/**
* @brief Worker pool worker thread
*/
class cmWorkerPoolWorker
{
public:
cmWorkerPoolWorker(uv_loop_t& uvLoop);
~cmWorkerPoolWorker();
cmWorkerPoolWorker(cmWorkerPoolWorker const&) = delete;
cmWorkerPoolWorker& operator=(cmWorkerPoolWorker const&) = delete;
/**
* Set the internal thread
*/
void SetThread(std::thread&& aThread) { Thread_ = std::move(aThread); }
/**
* Run an external process
*/
bool RunProcess(cmWorkerPool::ProcessResultT& result,
std::vector<std::string> const& command,
std::string const& workingDirectory);
private:
// -- Libuv callbacks
static void UVProcessStart(uv_async_t* handle);
void UVProcessFinished();
private:
// -- Process management
struct
{
std::mutex Mutex;
cm::uv_async_ptr Request;
std::condition_variable Condition;
std::unique_ptr<cmUVReadOnlyProcess> ROP;
} Proc_;
// -- System thread
std::thread Thread_;
};
cmWorkerPoolWorker::cmWorkerPoolWorker(uv_loop_t& uvLoop)
{
Proc_.Request.init(uvLoop, &cmWorkerPoolWorker::UVProcessStart, this);
}
cmWorkerPoolWorker::~cmWorkerPoolWorker()
{
if (Thread_.joinable()) {
Thread_.join();
}
}
bool cmWorkerPoolWorker::RunProcess(cmWorkerPool::ProcessResultT& result,
std::vector<std::string> const& command,
std::string const& workingDirectory)
{
if (command.empty()) {
return false;
}
// Create process instance
{
std::lock_guard<std::mutex> lock(Proc_.Mutex);
Proc_.ROP = cm::make_unique<cmUVReadOnlyProcess>();
Proc_.ROP->setup(&result, true, command, workingDirectory);
}
// Send asynchronous process start request to libuv loop
Proc_.Request.send();
// Wait until the process has been finished and destroyed
{
std::unique_lock<std::mutex> ulock(Proc_.Mutex);
while (Proc_.ROP) {
Proc_.Condition.wait(ulock);
}
}
return !result.error();
}
void cmWorkerPoolWorker::UVProcessStart(uv_async_t* handle)
{
auto* wrk = reinterpret_cast<cmWorkerPoolWorker*>(handle->data);
bool startFailed = false;
{
auto& Proc = wrk->Proc_;
std::lock_guard<std::mutex> lock(Proc.Mutex);
if (Proc.ROP && !Proc.ROP->IsStarted()) {
startFailed =
!Proc.ROP->start(handle->loop, [wrk] { wrk->UVProcessFinished(); });
}
}
// Clean up if starting of the process failed
if (startFailed) {
wrk->UVProcessFinished();
}
}
void cmWorkerPoolWorker::UVProcessFinished()
{
{
std::lock_guard<std::mutex> lock(Proc_.Mutex);
if (Proc_.ROP && (Proc_.ROP->IsFinished() || !Proc_.ROP->IsStarted())) {
Proc_.ROP.reset();
}
}
// Notify idling thread
Proc_.Condition.notify_one();
}
/**
* @brief Private worker pool internals
*/
class cmWorkerPoolInternal
{
public:
// -- Constructors
cmWorkerPoolInternal(cmWorkerPool* pool);
~cmWorkerPoolInternal();
/**
* Runs the libuv loop.
*/
bool Process();
/**
* Clear queue and abort threads.
*/
void Abort();
/**
* Push a job to the queue and notify a worker.
*/
bool PushJob(cmWorkerPool::JobHandleT&& jobHandle);
/**
* Worker thread main loop method.
*/
void Work(unsigned int workerIndex);
// -- Request slots
static void UVSlotBegin(uv_async_t* handle);
static void UVSlotEnd(uv_async_t* handle);
public:
// -- UV loop
#ifdef CMAKE_UV_SIGNAL_HACK
std::unique_ptr<cmUVSignalHackRAII> UVHackRAII;
#endif
std::unique_ptr<uv_loop_t> UVLoop;
cm::uv_async_ptr UVRequestBegin;
cm::uv_async_ptr UVRequestEnd;
// -- Thread pool and job queue
std::mutex Mutex;
bool Processing = false;
bool Aborting = false;
bool FenceProcessing = false;
unsigned int WorkersRunning = 0;
unsigned int WorkersIdle = 0;
unsigned int JobsProcessing = 0;
std::deque<cmWorkerPool::JobHandleT> Queue;
std::condition_variable Condition;
std::vector<std::unique_ptr<cmWorkerPoolWorker>> Workers;
// -- References
cmWorkerPool* Pool = nullptr;
};
void cmWorkerPool::ProcessResultT::reset()
{
ExitStatus = 0;
TermSignal = 0;
if (!StdOut.empty()) {
StdOut.clear();
StdOut.shrink_to_fit();
}
if (!StdErr.empty()) {
StdErr.clear();
StdErr.shrink_to_fit();
}
if (!ErrorMessage.empty()) {
ErrorMessage.clear();
ErrorMessage.shrink_to_fit();
}
}
cmWorkerPoolInternal::cmWorkerPoolInternal(cmWorkerPool* pool)
: Pool(pool)
{
// Initialize libuv loop
uv_disable_stdio_inheritance();
#ifdef CMAKE_UV_SIGNAL_HACK
UVHackRAII = cm::make_unique<cmUVSignalHackRAII>();
#endif
UVLoop = cm::make_unique<uv_loop_t>();
uv_loop_init(UVLoop.get());
}
cmWorkerPoolInternal::~cmWorkerPoolInternal()
{
uv_loop_close(UVLoop.get());
}
bool cmWorkerPoolInternal::Process()
{
// Reset state flags
Processing = true;
Aborting = false;
// Initialize libuv asynchronous request
UVRequestBegin.init(*UVLoop, &cmWorkerPoolInternal::UVSlotBegin, this);
UVRequestEnd.init(*UVLoop, &cmWorkerPoolInternal::UVSlotEnd, this);
// Send begin request
UVRequestBegin.send();
// Run libuv loop
bool success = (uv_run(UVLoop.get(), UV_RUN_DEFAULT) == 0);
// Update state flags
Processing = false;
Aborting = false;
return success;
}
void cmWorkerPoolInternal::Abort()
{
bool notifyThreads = false;
// Clear all jobs and set abort flag
{
std::lock_guard<std::mutex> guard(Mutex);
if (Processing && !Aborting) {
// Register abort and clear queue
Aborting = true;
Queue.clear();
notifyThreads = true;
}
}
if (notifyThreads) {
// Wake threads
Condition.notify_all();
}
}
inline bool cmWorkerPoolInternal::PushJob(cmWorkerPool::JobHandleT&& jobHandle)
{
std::lock_guard<std::mutex> guard(Mutex);
if (Aborting) {
return false;
}
// Append the job to the queue
Queue.emplace_back(std::move(jobHandle));
// Notify an idle worker if there's one
if (WorkersIdle != 0) {
Condition.notify_one();
}
// Return success
return true;
}
void cmWorkerPoolInternal::UVSlotBegin(uv_async_t* handle)
{
auto& gint = *reinterpret_cast<cmWorkerPoolInternal*>(handle->data);
// Create worker threads
{
unsigned int const num = gint.Pool->ThreadCount();
// Create workers
gint.Workers.reserve(num);
for (unsigned int ii = 0; ii != num; ++ii) {
gint.Workers.emplace_back(
cm::make_unique<cmWorkerPoolWorker>(*gint.UVLoop));
}
// Start worker threads
for (unsigned int ii = 0; ii != num; ++ii) {
gint.Workers[ii]->SetThread(
std::thread(&cmWorkerPoolInternal::Work, &gint, ii));
}
}
// Destroy begin request
gint.UVRequestBegin.reset();
}
void cmWorkerPoolInternal::UVSlotEnd(uv_async_t* handle)
{
auto& gint = *reinterpret_cast<cmWorkerPoolInternal*>(handle->data);
// Join and destroy worker threads
gint.Workers.clear();
// Destroy end request
gint.UVRequestEnd.reset();
}
void cmWorkerPoolInternal::Work(unsigned int workerIndex)
{
cmWorkerPool::JobHandleT jobHandle;
std::unique_lock<std::mutex> uLock(Mutex);
// Increment running workers count
++WorkersRunning;
// Enter worker main loop
while (true) {
// Abort on request
if (Aborting) {
break;
}
// Wait for new jobs
if (Queue.empty()) {
++WorkersIdle;
Condition.wait(uLock);
--WorkersIdle;
continue;
}
// Check for fence jobs
if (FenceProcessing || Queue.front()->IsFence()) {
if (JobsProcessing != 0) {
Condition.wait(uLock);
continue;
}
// No jobs get processed. Set the fence job processing flag.
FenceProcessing = true;
}
// Pop next job from queue
jobHandle = std::move(Queue.front());
Queue.pop_front();
// Unlocked scope for job processing
++JobsProcessing;
{
uLock.unlock();
jobHandle->Work(Pool, workerIndex); // Process job
jobHandle.reset(); // Destroy job
uLock.lock();
}
--JobsProcessing;
// Was this a fence job?
if (FenceProcessing) {
FenceProcessing = false;
Condition.notify_all();
}
}
// Decrement running workers count
if (--WorkersRunning == 0) {
// Last worker thread about to finish. Send libuv event.
UVRequestEnd.send();
}
}
cmWorkerPool::JobT::~JobT() = default;
bool cmWorkerPool::JobT::RunProcess(ProcessResultT& result,
std::vector<std::string> const& command,
std::string const& workingDirectory)
{
// Get worker by index
auto* wrk = Pool_->Int_->Workers.at(WorkerIndex_).get();
return wrk->RunProcess(result, command, workingDirectory);
}
cmWorkerPool::cmWorkerPool()
: Int_(cm::make_unique<cmWorkerPoolInternal>(this))
{
}
cmWorkerPool::~cmWorkerPool() = default;
void cmWorkerPool::SetThreadCount(unsigned int threadCount)
{
if (!Int_->Processing) {
ThreadCount_ = (threadCount > 0) ? threadCount : 1u;
}
}
bool cmWorkerPool::Process(void* userData)
{
// Setup user data
UserData_ = userData;
// Run libuv loop
bool success = Int_->Process();
// Clear user data
UserData_ = nullptr;
// Return
return success;
}
bool cmWorkerPool::PushJob(JobHandleT&& jobHandle)
{
return Int_->PushJob(std::move(jobHandle));
}
void cmWorkerPool::Abort()
{
Int_->Abort();
}
![swh spinner](/static/img/swh-spinner.gif)
Computing file changes ...