/* 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 #include #include #include #include #include #include #include #include #include #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; using DataFunction = std::function; /// On error the ssize_t argument is a non zero libuv error code using EndFunction = std::function; 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 this->UVPipe_.get(); } //! uv_pipe() casted to libuv stream uv_stream_t* uv_stream() const { return static_cast(this->UVPipe_); } //! uv_pipe() casted to libuv handle uv_handle_t* uv_handle() { return static_cast(this->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 Buffer_; DataFunction DataFunction_; EndFunction EndFunction_; }; void cmUVPipeBuffer::reset() { if (this->UVPipe_.get() != nullptr) { this->EndFunction_ = nullptr; this->DataFunction_ = nullptr; this->Buffer_.clear(); this->Buffer_.shrink_to_fit(); this->UVPipe_.reset(); } } bool cmUVPipeBuffer::init(uv_loop_t* uv_loop) { this->reset(); if (uv_loop == nullptr) { return false; } int ret = this->UVPipe_.init(*uv_loop, 0, this); return (ret == 0); } bool cmUVPipeBuffer::startRead(DataFunction dataFunction, EndFunction endFunction) { if (this->UVPipe_.get() == nullptr) { return false; } if (!dataFunction || !endFunction) { return false; } this->DataFunction_ = std::move(dataFunction); this->EndFunction_ = std::move(endFunction); int ret = uv_read_start(this->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(handle->data); pipe.Buffer_.resize(suggestedSize); buf->base = pipe.Buffer_.data(); buf->len = static_cast(pipe.Buffer_.size()); } void cmUVPipeBuffer::UVData(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) { auto& pipe = *reinterpret_cast(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 Command; cmWorkerPool::ProcessResultT* Result = nullptr; bool MergedOutput = false; }; public: // -- Const accessors SetupT const& Setup() const { return this->Setup_; } cmWorkerPool::ProcessResultT* Result() const { return this->Setup_.Result; } bool IsStarted() const { return this->IsStarted_; } bool IsFinished() const { return this->IsFinished_; } // -- Runtime void setup(cmWorkerPool::ProcessResultT* result, bool mergedOutput, std::vector const& command, std::string const& workingDirectory = std::string()); bool start(uv_loop_t* uv_loop, std::function 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 FinishedCallback_; std::vector CommandPtr_; std::array 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 const& command, std::string const& workingDirectory) { this->Setup_.WorkingDirectory = workingDirectory; this->Setup_.Command = command; this->Setup_.Result = result; this->Setup_.MergedOutput = mergedOutput; } bool cmUVReadOnlyProcess::start(uv_loop_t* uv_loop, std::function finishedCallback) { if (this->IsStarted() || (this->Result() == nullptr)) { return false; } // Reset result before the start this->Result()->reset(); // Fill command string pointers if (!this->Setup().Command.empty()) { this->CommandPtr_.reserve(this->Setup().Command.size() + 1); for (std::string const& arg : this->Setup().Command) { this->CommandPtr_.push_back(arg.c_str()); } this->CommandPtr_.push_back(nullptr); } else { this->Result()->ErrorMessage = "Empty command"; } if (!this->Result()->error()) { if (!this->UVPipeOut_.init(uv_loop)) { this->Result()->ErrorMessage = "libuv stdout pipe initialization failed"; } } if (!this->Result()->error()) { if (!this->UVPipeErr_.init(uv_loop)) { this->Result()->ErrorMessage = "libuv stderr pipe initialization failed"; } } if (!this->Result()->error()) { // -- Setup process stdio options // stdin this->UVOptionsStdIO_[0].flags = UV_IGNORE; this->UVOptionsStdIO_[0].data.stream = nullptr; // stdout this->UVOptionsStdIO_[1].flags = static_cast(UV_CREATE_PIPE | UV_WRITABLE_PIPE); this->UVOptionsStdIO_[1].data.stream = this->UVPipeOut_.uv_stream(); // stderr this->UVOptionsStdIO_[2].flags = static_cast(UV_CREATE_PIPE | UV_WRITABLE_PIPE); this->UVOptionsStdIO_[2].data.stream = this->UVPipeErr_.uv_stream(); // -- Setup process options std::fill_n(reinterpret_cast(&this->UVOptions_), sizeof(this->UVOptions_), 0); this->UVOptions_.exit_cb = &cmUVReadOnlyProcess::UVExit; this->UVOptions_.file = this->CommandPtr_[0]; this->UVOptions_.args = const_cast(this->CommandPtr_.data()); this->UVOptions_.cwd = this->Setup_.WorkingDirectory.c_str(); this->UVOptions_.flags = UV_PROCESS_WINDOWS_HIDE; this->UVOptions_.stdio_count = static_cast(this->UVOptionsStdIO_.size()); this->UVOptions_.stdio = this->UVOptionsStdIO_.data(); // -- Spawn process int uvErrorCode = this->UVProcess_.spawn(*uv_loop, this->UVOptions_, this); if (uvErrorCode != 0) { this->Result()->ErrorMessage = "libuv process spawn failed"; if (const char* uvErr = uv_strerror(uvErrorCode)) { this->Result()->ErrorMessage += ": "; this->Result()->ErrorMessage += uvErr; } } } // -- Start reading from stdio streams if (!this->Result()->error()) { if (!this->UVPipeOut_.startRead( [this](cmUVPipeBuffer::DataRange range) { this->UVPipeOutData(range); }, [this](ssize_t error) { this->UVPipeOutEnd(error); })) { this->Result()->ErrorMessage = "libuv start reading from stdout pipe failed"; } } if (!this->Result()->error()) { if (!this->UVPipeErr_.startRead( [this](cmUVPipeBuffer::DataRange range) { this->UVPipeErrData(range); }, [this](ssize_t error) { this->UVPipeErrEnd(error); })) { this->Result()->ErrorMessage = "libuv start reading from stderr pipe failed"; } } if (!this->Result()->error()) { this->IsStarted_ = true; this->FinishedCallback_ = std::move(finishedCallback); } else { // Clear libuv handles and finish this->UVProcess_.reset(); this->UVPipeOut_.reset(); this->UVPipeErr_.reset(); this->CommandPtr_.clear(); } return this->IsStarted(); } void cmUVReadOnlyProcess::UVExit(uv_process_t* handle, int64_t exitStatus, int termSignal) { auto& proc = *reinterpret_cast(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) { this->Result()->StdOut.append(data.begin(), data.end()); } void cmUVReadOnlyProcess::UVPipeOutEnd(ssize_t error) { // Process pipe error if ((error != 0) && !this->Result()->error()) { this->Result()->ErrorMessage = cmStrCat( "Reading from stdout pipe failed with libuv error code ", error); } // Try finish this->UVTryFinish(); } void cmUVReadOnlyProcess::UVPipeErrData(cmUVPipeBuffer::DataRange data) { std::string* str = this->Setup_.MergedOutput ? &this->Result()->StdOut : &this->Result()->StdErr; str->append(data.begin(), data.end()); } void cmUVReadOnlyProcess::UVPipeErrEnd(ssize_t error) { // Process pipe error if ((error != 0) && !this->Result()->error()) { this->Result()->ErrorMessage = cmStrCat( "Reading from stderr pipe failed with libuv error code ", error); } // Try finish this->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 ((this->UVProcess_.get() != nullptr) || (this->UVPipeOut_.uv_pipe() != nullptr) || (this->UVPipeErr_.uv_pipe() != nullptr)) { return; } this->IsFinished_ = true; this->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) { this->Thread_ = std::move(aThread); } /** * Run an external process */ bool RunProcess(cmWorkerPool::ProcessResultT& result, std::vector 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 ROP; } Proc_; // -- System thread std::thread Thread_; }; cmWorkerPoolWorker::cmWorkerPoolWorker(uv_loop_t& uvLoop) { this->Proc_.Request.init(uvLoop, &cmWorkerPoolWorker::UVProcessStart, this); } cmWorkerPoolWorker::~cmWorkerPoolWorker() { if (this->Thread_.joinable()) { this->Thread_.join(); } } bool cmWorkerPoolWorker::RunProcess(cmWorkerPool::ProcessResultT& result, std::vector const& command, std::string const& workingDirectory) { if (command.empty()) { return false; } // Create process instance { std::lock_guard lock(this->Proc_.Mutex); this->Proc_.ROP = cm::make_unique(); this->Proc_.ROP->setup(&result, true, command, workingDirectory); } // Send asynchronous process start request to libuv loop this->Proc_.Request.send(); // Wait until the process has been finished and destroyed { std::unique_lock ulock(this->Proc_.Mutex); while (this->Proc_.ROP) { this->Proc_.Condition.wait(ulock); } } return !result.error(); } void cmWorkerPoolWorker::UVProcessStart(uv_async_t* handle) { auto* wrk = reinterpret_cast(handle->data); bool startFailed = false; { auto& Proc = wrk->Proc_; std::lock_guard 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 lock(this->Proc_.Mutex); if (this->Proc_.ROP && (this->Proc_.ROP->IsFinished() || !this->Proc_.ROP->IsStarted())) { this->Proc_.ROP.reset(); } // Notify idling thread this->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 UVHackRAII; #endif std::unique_ptr 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 Queue; std::condition_variable Condition; std::condition_variable ConditionFence; std::vector> Workers; // -- References cmWorkerPool* Pool = nullptr; }; void cmWorkerPool::ProcessResultT::reset() { this->ExitStatus = 0; this->TermSignal = 0; if (!this->StdOut.empty()) { this->StdOut.clear(); this->StdOut.shrink_to_fit(); } if (!this->StdErr.empty()) { this->StdErr.clear(); this->StdErr.shrink_to_fit(); } if (!this->ErrorMessage.empty()) { this->ErrorMessage.clear(); this->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(); #endif this->UVLoop = cm::make_unique(); uv_loop_init(this->UVLoop.get()); } cmWorkerPoolInternal::~cmWorkerPoolInternal() { uv_loop_close(this->UVLoop.get()); } bool cmWorkerPoolInternal::Process() { // Reset state flags this->Processing = true; this->Aborting = false; // Initialize libuv asynchronous request this->UVRequestBegin.init(*this->UVLoop, &cmWorkerPoolInternal::UVSlotBegin, this); this->UVRequestEnd.init(*this->UVLoop, &cmWorkerPoolInternal::UVSlotEnd, this); // Send begin request this->UVRequestBegin.send(); // Run libuv loop bool success = (uv_run(this->UVLoop.get(), UV_RUN_DEFAULT) == 0); // Update state flags this->Processing = false; this->Aborting = false; return success; } void cmWorkerPoolInternal::Abort() { // Clear all jobs and set abort flag std::lock_guard guard(this->Mutex); if (!this->Aborting) { // Register abort and clear queue this->Aborting = true; this->Queue.clear(); this->Condition.notify_all(); } } inline bool cmWorkerPoolInternal::PushJob(cmWorkerPool::JobHandleT&& jobHandle) { std::lock_guard guard(this->Mutex); if (this->Aborting) { return false; } // Append the job to the queue this->Queue.emplace_back(std::move(jobHandle)); // Notify an idle worker if there's one if (this->WorkersIdle != 0) { this->Condition.notify_one(); } // Return success return true; } void cmWorkerPoolInternal::UVSlotBegin(uv_async_t* handle) { auto& gint = *reinterpret_cast(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(*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(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 uLock(this->Mutex); // Increment running workers count ++this->WorkersRunning; // Enter worker main loop while (true) { // Abort on request if (this->Aborting) { break; } // Wait for new jobs on the main CV if (this->Queue.empty()) { ++this->WorkersIdle; this->Condition.wait(uLock); --this->WorkersIdle; continue; } // If there is a fence currently active or waiting, // sleep on the main CV and try again. if (this->FenceProcessing) { this->Condition.wait(uLock); continue; } // Pop next job from queue jobHandle = std::move(this->Queue.front()); this->Queue.pop_front(); // Check for fence jobs bool raisedFence = false; if (jobHandle->IsFence()) { this->FenceProcessing = true; raisedFence = true; // Wait on the Fence CV until all pending jobs are done. while (this->JobsProcessing != 0 && !this->Aborting) { this->ConditionFence.wait(uLock); } // When aborting, explicitly kick all threads alive once more. if (this->Aborting) { this->FenceProcessing = false; this->Condition.notify_all(); break; } } // Unlocked scope for job processing ++this->JobsProcessing; { uLock.unlock(); jobHandle->Work(this->Pool, workerIndex); // Process job jobHandle.reset(); // Destroy job uLock.lock(); } --this->JobsProcessing; // If this was the thread that entered fence processing // originally, notify all idling workers that the fence // is done. if (raisedFence) { this->FenceProcessing = false; this->Condition.notify_all(); } // If fence processing is still not done, notify the // the fencing worker when all active jobs are done. if (this->FenceProcessing && this->JobsProcessing == 0) { this->ConditionFence.notify_all(); } } // Decrement running workers count if (--this->WorkersRunning == 0) { // Last worker thread about to finish. Send libuv event. this->UVRequestEnd.send(); } } cmWorkerPool::JobT::~JobT() = default; bool cmWorkerPool::JobT::RunProcess(ProcessResultT& result, std::vector const& command, std::string const& workingDirectory) { // Get worker by index auto* wrk = this->Pool_->Int_->Workers.at(this->WorkerIndex_).get(); return wrk->RunProcess(result, command, workingDirectory); } cmWorkerPool::cmWorkerPool() : Int_(cm::make_unique(this)) { } cmWorkerPool::~cmWorkerPool() = default; void cmWorkerPool::SetThreadCount(unsigned int threadCount) { if (!this->Int_->Processing) { this->ThreadCount_ = (threadCount > 0) ? threadCount : 1u; } } bool cmWorkerPool::Process(void* userData) { // Setup user data this->UserData_ = userData; // Run libuv loop bool success = this->Int_->Process(); // Clear user data this->UserData_ = nullptr; // Return return success; } bool cmWorkerPool::PushJob(JobHandleT&& jobHandle) { return this->Int_->PushJob(std::move(jobHandle)); } void cmWorkerPool::Abort() { this->Int_->Abort(); }