cmWorkerPool.cxx 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776
  1. /* Distributed under the OSI-approved BSD 3-Clause License. See accompanying
  2. file Copyright.txt or https://cmake.org/licensing for details. */
  3. #include "cmWorkerPool.h"
  4. #include <algorithm>
  5. #include <array>
  6. #include <condition_variable>
  7. #include <cstddef>
  8. #include <deque>
  9. #include <functional>
  10. #include <mutex>
  11. #include <thread>
  12. #include <cm/memory>
  13. #include <cm3p/uv.h>
  14. #include "cmRange.h"
  15. #include "cmStringAlgorithms.h"
  16. #include "cmUVHandlePtr.h"
  17. /**
  18. * @brief libuv pipe buffer class
  19. */
  20. class cmUVPipeBuffer
  21. {
  22. public:
  23. using DataRange = cmRange<const char*>;
  24. using DataFunction = std::function<void(DataRange)>;
  25. /// On error the ssize_t argument is a non zero libuv error code
  26. using EndFunction = std::function<void(ssize_t)>;
  27. /**
  28. * Reset to construction state
  29. */
  30. void reset();
  31. /**
  32. * Initializes uv_pipe(), uv_stream() and uv_handle()
  33. * @return true on success
  34. */
  35. bool init(uv_loop_t* uv_loop);
  36. /**
  37. * Start reading
  38. * @return true on success
  39. */
  40. bool startRead(DataFunction dataFunction, EndFunction endFunction);
  41. //! libuv pipe
  42. uv_pipe_t* uv_pipe() const { return this->UVPipe_.get(); }
  43. //! uv_pipe() casted to libuv stream
  44. uv_stream_t* uv_stream() const
  45. {
  46. return static_cast<uv_stream_t*>(this->UVPipe_);
  47. }
  48. //! uv_pipe() casted to libuv handle
  49. uv_handle_t* uv_handle() { return static_cast<uv_handle_t*>(this->UVPipe_); }
  50. private:
  51. // -- Libuv callbacks
  52. static void UVAlloc(uv_handle_t* handle, size_t suggestedSize,
  53. uv_buf_t* buf);
  54. static void UVData(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf);
  55. cm::uv_pipe_ptr UVPipe_;
  56. std::vector<char> Buffer_;
  57. DataFunction DataFunction_;
  58. EndFunction EndFunction_;
  59. };
  60. void cmUVPipeBuffer::reset()
  61. {
  62. if (this->UVPipe_.get()) {
  63. this->EndFunction_ = nullptr;
  64. this->DataFunction_ = nullptr;
  65. this->Buffer_.clear();
  66. this->Buffer_.shrink_to_fit();
  67. this->UVPipe_.reset();
  68. }
  69. }
  70. bool cmUVPipeBuffer::init(uv_loop_t* uv_loop)
  71. {
  72. this->reset();
  73. if (!uv_loop) {
  74. return false;
  75. }
  76. int ret = this->UVPipe_.init(*uv_loop, 0, this);
  77. return (ret == 0);
  78. }
  79. bool cmUVPipeBuffer::startRead(DataFunction dataFunction,
  80. EndFunction endFunction)
  81. {
  82. if (!this->UVPipe_.get()) {
  83. return false;
  84. }
  85. if (!dataFunction || !endFunction) {
  86. return false;
  87. }
  88. this->DataFunction_ = std::move(dataFunction);
  89. this->EndFunction_ = std::move(endFunction);
  90. int ret = uv_read_start(this->uv_stream(), &cmUVPipeBuffer::UVAlloc,
  91. &cmUVPipeBuffer::UVData);
  92. return (ret == 0);
  93. }
  94. void cmUVPipeBuffer::UVAlloc(uv_handle_t* handle, size_t suggestedSize,
  95. uv_buf_t* buf)
  96. {
  97. auto& pipe = *reinterpret_cast<cmUVPipeBuffer*>(handle->data);
  98. pipe.Buffer_.resize(suggestedSize);
  99. buf->base = pipe.Buffer_.data();
  100. buf->len = static_cast<unsigned long>(pipe.Buffer_.size());
  101. }
  102. void cmUVPipeBuffer::UVData(uv_stream_t* stream, ssize_t nread,
  103. const uv_buf_t* buf)
  104. {
  105. auto& pipe = *reinterpret_cast<cmUVPipeBuffer*>(stream->data);
  106. if (nread > 0) {
  107. if (buf->base) {
  108. // Call data function
  109. pipe.DataFunction_(DataRange(buf->base, buf->base + nread));
  110. }
  111. } else if (nread < 0) {
  112. // Save the end function on the stack before resetting the pipe
  113. EndFunction efunc;
  114. efunc.swap(pipe.EndFunction_);
  115. // Reset pipe before calling the end function
  116. pipe.reset();
  117. // Call end function
  118. efunc((nread == UV_EOF) ? 0 : nread);
  119. }
  120. }
  121. /**
  122. * @brief External process management class
  123. */
  124. class cmUVReadOnlyProcess
  125. {
  126. public:
  127. // -- Types
  128. //! @brief Process settings
  129. struct SetupT
  130. {
  131. std::string WorkingDirectory;
  132. std::vector<std::string> Command;
  133. cmWorkerPool::ProcessResultT* Result = nullptr;
  134. bool MergedOutput = false;
  135. };
  136. // -- Const accessors
  137. SetupT const& Setup() const { return this->Setup_; }
  138. cmWorkerPool::ProcessResultT* Result() const { return this->Setup_.Result; }
  139. bool IsStarted() const { return this->IsStarted_; }
  140. bool IsFinished() const { return this->IsFinished_; }
  141. // -- Runtime
  142. void setup(cmWorkerPool::ProcessResultT* result, bool mergedOutput,
  143. std::vector<std::string> const& command,
  144. std::string const& workingDirectory = std::string());
  145. bool start(uv_loop_t* uv_loop, std::function<void()> finishedCallback);
  146. private:
  147. // -- Libuv callbacks
  148. static void UVExit(uv_process_t* handle, int64_t exitStatus, int termSignal);
  149. void UVPipeOutData(cmUVPipeBuffer::DataRange data) const;
  150. void UVPipeOutEnd(ssize_t error);
  151. void UVPipeErrData(cmUVPipeBuffer::DataRange data) const;
  152. void UVPipeErrEnd(ssize_t error);
  153. void UVTryFinish();
  154. // -- Setup
  155. SetupT Setup_;
  156. // -- Runtime
  157. bool IsStarted_ = false;
  158. bool IsFinished_ = false;
  159. std::function<void()> FinishedCallback_;
  160. std::vector<const char*> CommandPtr_;
  161. std::array<uv_stdio_container_t, 3> UVOptionsStdIO_;
  162. uv_process_options_t UVOptions_;
  163. cm::uv_process_ptr UVProcess_;
  164. cmUVPipeBuffer UVPipeOut_;
  165. cmUVPipeBuffer UVPipeErr_;
  166. };
  167. void cmUVReadOnlyProcess::setup(cmWorkerPool::ProcessResultT* result,
  168. bool mergedOutput,
  169. std::vector<std::string> const& command,
  170. std::string const& workingDirectory)
  171. {
  172. this->Setup_.WorkingDirectory = workingDirectory;
  173. this->Setup_.Command = command;
  174. this->Setup_.Result = result;
  175. this->Setup_.MergedOutput = mergedOutput;
  176. }
  177. bool cmUVReadOnlyProcess::start(uv_loop_t* uv_loop,
  178. std::function<void()> finishedCallback)
  179. {
  180. if (this->IsStarted() || !this->Result()) {
  181. return false;
  182. }
  183. // Reset result before the start
  184. this->Result()->reset();
  185. // Fill command string pointers
  186. if (!this->Setup().Command.empty()) {
  187. this->CommandPtr_.reserve(this->Setup().Command.size() + 1);
  188. for (std::string const& arg : this->Setup().Command) {
  189. this->CommandPtr_.push_back(arg.c_str());
  190. }
  191. this->CommandPtr_.push_back(nullptr);
  192. } else {
  193. this->Result()->ErrorMessage = "Empty command";
  194. }
  195. if (!this->Result()->error()) {
  196. if (!this->UVPipeOut_.init(uv_loop)) {
  197. this->Result()->ErrorMessage = "libuv stdout pipe initialization failed";
  198. }
  199. }
  200. if (!this->Result()->error()) {
  201. if (!this->UVPipeErr_.init(uv_loop)) {
  202. this->Result()->ErrorMessage = "libuv stderr pipe initialization failed";
  203. }
  204. }
  205. if (!this->Result()->error()) {
  206. // -- Setup process stdio options
  207. // stdin
  208. this->UVOptionsStdIO_[0].flags = UV_IGNORE;
  209. this->UVOptionsStdIO_[0].data.stream = nullptr;
  210. // stdout
  211. this->UVOptionsStdIO_[1].flags =
  212. static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
  213. this->UVOptionsStdIO_[1].data.stream = this->UVPipeOut_.uv_stream();
  214. // stderr
  215. this->UVOptionsStdIO_[2].flags =
  216. static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
  217. this->UVOptionsStdIO_[2].data.stream = this->UVPipeErr_.uv_stream();
  218. // -- Setup process options
  219. std::fill_n(reinterpret_cast<char*>(&this->UVOptions_),
  220. sizeof(this->UVOptions_), 0);
  221. this->UVOptions_.exit_cb = &cmUVReadOnlyProcess::UVExit;
  222. this->UVOptions_.file = this->CommandPtr_[0];
  223. this->UVOptions_.args = const_cast<char**>(this->CommandPtr_.data());
  224. this->UVOptions_.cwd = this->Setup_.WorkingDirectory.c_str();
  225. this->UVOptions_.flags = UV_PROCESS_WINDOWS_HIDE;
  226. #if UV_VERSION_MAJOR > 1 || !defined(CMAKE_USE_SYSTEM_LIBUV)
  227. this->UVOptions_.flags |= UV_PROCESS_WINDOWS_USE_PARENT_ERROR_MODE;
  228. #endif
  229. this->UVOptions_.stdio_count =
  230. static_cast<int>(this->UVOptionsStdIO_.size());
  231. this->UVOptions_.stdio = this->UVOptionsStdIO_.data();
  232. // -- Spawn process
  233. int uvErrorCode = this->UVProcess_.spawn(*uv_loop, this->UVOptions_, this);
  234. if (uvErrorCode != 0) {
  235. this->Result()->ErrorMessage = "libuv process spawn failed";
  236. if (const char* uvErr = uv_strerror(uvErrorCode)) {
  237. this->Result()->ErrorMessage += ": ";
  238. this->Result()->ErrorMessage += uvErr;
  239. }
  240. }
  241. }
  242. // -- Start reading from stdio streams
  243. if (!this->Result()->error()) {
  244. if (!this->UVPipeOut_.startRead(
  245. [this](cmUVPipeBuffer::DataRange range) {
  246. this->UVPipeOutData(range);
  247. },
  248. [this](ssize_t error) { this->UVPipeOutEnd(error); })) {
  249. this->Result()->ErrorMessage =
  250. "libuv start reading from stdout pipe failed";
  251. }
  252. }
  253. if (!this->Result()->error()) {
  254. if (!this->UVPipeErr_.startRead(
  255. [this](cmUVPipeBuffer::DataRange range) {
  256. this->UVPipeErrData(range);
  257. },
  258. [this](ssize_t error) { this->UVPipeErrEnd(error); })) {
  259. this->Result()->ErrorMessage =
  260. "libuv start reading from stderr pipe failed";
  261. }
  262. }
  263. if (!this->Result()->error()) {
  264. this->IsStarted_ = true;
  265. this->FinishedCallback_ = std::move(finishedCallback);
  266. } else {
  267. // Clear libuv handles and finish
  268. this->UVProcess_.reset();
  269. this->UVPipeOut_.reset();
  270. this->UVPipeErr_.reset();
  271. this->CommandPtr_.clear();
  272. }
  273. return this->IsStarted();
  274. }
  275. void cmUVReadOnlyProcess::UVExit(uv_process_t* handle, int64_t exitStatus,
  276. int termSignal)
  277. {
  278. auto& proc = *reinterpret_cast<cmUVReadOnlyProcess*>(handle->data);
  279. if (proc.IsStarted() && !proc.IsFinished()) {
  280. // Set error message on demand
  281. proc.Result()->ExitStatus = exitStatus;
  282. proc.Result()->TermSignal = termSignal;
  283. if (proc.Result()->ErrorMessage.empty()) {
  284. if (termSignal != 0) {
  285. proc.Result()->ErrorMessage = cmStrCat(
  286. "Process was terminated by signal ", proc.Result()->TermSignal);
  287. } else if (exitStatus != 0) {
  288. proc.Result()->ErrorMessage = cmStrCat(
  289. "Process failed with return value ", proc.Result()->ExitStatus);
  290. }
  291. }
  292. // Reset process handle
  293. proc.UVProcess_.reset();
  294. // Try finish
  295. proc.UVTryFinish();
  296. }
  297. }
  298. void cmUVReadOnlyProcess::UVPipeOutData(cmUVPipeBuffer::DataRange data) const
  299. {
  300. this->Result()->StdOut.append(data.begin(), data.end());
  301. }
  302. void cmUVReadOnlyProcess::UVPipeOutEnd(ssize_t error)
  303. {
  304. // Process pipe error
  305. if ((error != 0) && !this->Result()->error()) {
  306. this->Result()->ErrorMessage = cmStrCat(
  307. "Reading from stdout pipe failed with libuv error code ", error);
  308. }
  309. // Try finish
  310. this->UVTryFinish();
  311. }
  312. void cmUVReadOnlyProcess::UVPipeErrData(cmUVPipeBuffer::DataRange data) const
  313. {
  314. std::string* str = this->Setup_.MergedOutput ? &this->Result()->StdOut
  315. : &this->Result()->StdErr;
  316. str->append(data.begin(), data.end());
  317. }
  318. void cmUVReadOnlyProcess::UVPipeErrEnd(ssize_t error)
  319. {
  320. // Process pipe error
  321. if ((error != 0) && !this->Result()->error()) {
  322. this->Result()->ErrorMessage = cmStrCat(
  323. "Reading from stderr pipe failed with libuv error code ", error);
  324. }
  325. // Try finish
  326. this->UVTryFinish();
  327. }
  328. void cmUVReadOnlyProcess::UVTryFinish()
  329. {
  330. // There still might be data in the pipes after the process has finished.
  331. // Therefore check if the process is finished AND all pipes are closed
  332. // before signaling the worker thread to continue.
  333. if ((this->UVProcess_.get()) || (this->UVPipeOut_.uv_pipe()) ||
  334. (this->UVPipeErr_.uv_pipe())) {
  335. return;
  336. }
  337. this->IsFinished_ = true;
  338. this->FinishedCallback_();
  339. }
  340. /**
  341. * @brief Worker pool worker thread
  342. */
  343. class cmWorkerPoolWorker
  344. {
  345. public:
  346. cmWorkerPoolWorker(uv_loop_t& uvLoop);
  347. ~cmWorkerPoolWorker();
  348. cmWorkerPoolWorker(cmWorkerPoolWorker const&) = delete;
  349. cmWorkerPoolWorker& operator=(cmWorkerPoolWorker const&) = delete;
  350. /**
  351. * Set the internal thread
  352. */
  353. void SetThread(std::thread&& aThread) { this->Thread_ = std::move(aThread); }
  354. /**
  355. * Run an external process
  356. */
  357. bool RunProcess(cmWorkerPool::ProcessResultT& result,
  358. std::vector<std::string> const& command,
  359. std::string const& workingDirectory);
  360. private:
  361. // -- Libuv callbacks
  362. static void UVProcessStart(uv_async_t* handle);
  363. void UVProcessFinished();
  364. // -- Process management
  365. struct
  366. {
  367. std::mutex Mutex;
  368. cm::uv_async_ptr Request;
  369. std::condition_variable Condition;
  370. std::unique_ptr<cmUVReadOnlyProcess> ROP;
  371. } Proc_;
  372. // -- System thread
  373. std::thread Thread_;
  374. };
  375. cmWorkerPoolWorker::cmWorkerPoolWorker(uv_loop_t& uvLoop)
  376. {
  377. this->Proc_.Request.init(uvLoop, &cmWorkerPoolWorker::UVProcessStart, this);
  378. }
  379. cmWorkerPoolWorker::~cmWorkerPoolWorker()
  380. {
  381. if (this->Thread_.joinable()) {
  382. this->Thread_.join();
  383. }
  384. }
  385. bool cmWorkerPoolWorker::RunProcess(cmWorkerPool::ProcessResultT& result,
  386. std::vector<std::string> const& command,
  387. std::string const& workingDirectory)
  388. {
  389. if (command.empty()) {
  390. return false;
  391. }
  392. // Create process instance
  393. {
  394. std::lock_guard<std::mutex> lock(this->Proc_.Mutex);
  395. this->Proc_.ROP = cm::make_unique<cmUVReadOnlyProcess>();
  396. this->Proc_.ROP->setup(&result, true, command, workingDirectory);
  397. }
  398. // Send asynchronous process start request to libuv loop
  399. this->Proc_.Request.send();
  400. // Wait until the process has been finished and destroyed
  401. {
  402. std::unique_lock<std::mutex> ulock(this->Proc_.Mutex);
  403. while (this->Proc_.ROP) {
  404. this->Proc_.Condition.wait(ulock);
  405. }
  406. }
  407. return !result.error();
  408. }
  409. void cmWorkerPoolWorker::UVProcessStart(uv_async_t* handle)
  410. {
  411. auto* worker = reinterpret_cast<cmWorkerPoolWorker*>(handle->data);
  412. bool startFailed = false;
  413. {
  414. auto& Proc = worker->Proc_;
  415. std::lock_guard<std::mutex> lock(Proc.Mutex);
  416. if (Proc.ROP && !Proc.ROP->IsStarted()) {
  417. startFailed = !Proc.ROP->start(
  418. handle->loop, [worker] { worker->UVProcessFinished(); });
  419. }
  420. }
  421. // Clean up if starting of the process failed
  422. if (startFailed) {
  423. worker->UVProcessFinished();
  424. }
  425. }
  426. void cmWorkerPoolWorker::UVProcessFinished()
  427. {
  428. std::lock_guard<std::mutex> lock(this->Proc_.Mutex);
  429. if (this->Proc_.ROP &&
  430. (this->Proc_.ROP->IsFinished() || !this->Proc_.ROP->IsStarted())) {
  431. this->Proc_.ROP.reset();
  432. }
  433. // Notify idling thread
  434. this->Proc_.Condition.notify_one();
  435. }
  436. /**
  437. * @brief Private worker pool internals
  438. */
  439. class cmWorkerPoolInternal
  440. {
  441. public:
  442. // -- Constructors
  443. cmWorkerPoolInternal(cmWorkerPool* pool);
  444. ~cmWorkerPoolInternal();
  445. /**
  446. * Runs the libuv loop.
  447. */
  448. bool Process();
  449. /**
  450. * Clear queue and abort threads.
  451. */
  452. void Abort();
  453. /**
  454. * Push a job to the queue and notify a worker.
  455. */
  456. bool PushJob(cmWorkerPool::JobHandleT&& jobHandle);
  457. /**
  458. * Worker thread main loop method.
  459. */
  460. void Work(unsigned int workerIndex);
  461. // -- Request slots
  462. static void UVSlotBegin(uv_async_t* handle);
  463. static void UVSlotEnd(uv_async_t* handle);
  464. // -- UV loop
  465. std::unique_ptr<uv_loop_t> UVLoop;
  466. cm::uv_async_ptr UVRequestBegin;
  467. cm::uv_async_ptr UVRequestEnd;
  468. // -- Thread pool and job queue
  469. std::mutex Mutex;
  470. bool Processing = false;
  471. bool Aborting = false;
  472. bool FenceProcessing = false;
  473. unsigned int WorkersRunning = 0;
  474. unsigned int WorkersIdle = 0;
  475. unsigned int JobsProcessing = 0;
  476. std::deque<cmWorkerPool::JobHandleT> Queue;
  477. std::condition_variable Condition;
  478. std::condition_variable ConditionFence;
  479. std::vector<std::unique_ptr<cmWorkerPoolWorker>> Workers;
  480. // -- References
  481. cmWorkerPool* Pool = nullptr;
  482. };
  483. void cmWorkerPool::ProcessResultT::reset()
  484. {
  485. this->ExitStatus = 0;
  486. this->TermSignal = 0;
  487. if (!this->StdOut.empty()) {
  488. this->StdOut.clear();
  489. this->StdOut.shrink_to_fit();
  490. }
  491. if (!this->StdErr.empty()) {
  492. this->StdErr.clear();
  493. this->StdErr.shrink_to_fit();
  494. }
  495. if (!this->ErrorMessage.empty()) {
  496. this->ErrorMessage.clear();
  497. this->ErrorMessage.shrink_to_fit();
  498. }
  499. }
  500. cmWorkerPoolInternal::cmWorkerPoolInternal(cmWorkerPool* pool)
  501. : Pool(pool)
  502. {
  503. // Initialize libuv loop
  504. uv_disable_stdio_inheritance();
  505. this->UVLoop = cm::make_unique<uv_loop_t>();
  506. uv_loop_init(this->UVLoop.get());
  507. }
  508. cmWorkerPoolInternal::~cmWorkerPoolInternal()
  509. {
  510. uv_loop_close(this->UVLoop.get());
  511. }
  512. bool cmWorkerPoolInternal::Process()
  513. {
  514. // Reset state flags
  515. this->Processing = true;
  516. this->Aborting = false;
  517. // Initialize libuv asynchronous request
  518. this->UVRequestBegin.init(*this->UVLoop, &cmWorkerPoolInternal::UVSlotBegin,
  519. this);
  520. this->UVRequestEnd.init(*this->UVLoop, &cmWorkerPoolInternal::UVSlotEnd,
  521. this);
  522. // Send begin request
  523. this->UVRequestBegin.send();
  524. // Run libuv loop
  525. bool success = (uv_run(this->UVLoop.get(), UV_RUN_DEFAULT) == 0);
  526. // Update state flags
  527. this->Processing = false;
  528. this->Aborting = false;
  529. return success;
  530. }
  531. void cmWorkerPoolInternal::Abort()
  532. {
  533. // Clear all jobs and set abort flag
  534. std::lock_guard<std::mutex> guard(this->Mutex);
  535. if (!this->Aborting) {
  536. // Register abort and clear queue
  537. this->Aborting = true;
  538. this->Queue.clear();
  539. this->Condition.notify_all();
  540. }
  541. }
  542. inline bool cmWorkerPoolInternal::PushJob(cmWorkerPool::JobHandleT&& jobHandle)
  543. {
  544. std::lock_guard<std::mutex> guard(this->Mutex);
  545. if (this->Aborting) {
  546. return false;
  547. }
  548. // Append the job to the queue
  549. this->Queue.emplace_back(std::move(jobHandle));
  550. // Notify an idle worker if there's one
  551. if (this->WorkersIdle != 0) {
  552. this->Condition.notify_one();
  553. }
  554. // Return success
  555. return true;
  556. }
  557. void cmWorkerPoolInternal::UVSlotBegin(uv_async_t* handle)
  558. {
  559. auto& gint = *reinterpret_cast<cmWorkerPoolInternal*>(handle->data);
  560. // Create worker threads
  561. {
  562. unsigned int const num = gint.Pool->ThreadCount();
  563. // Create workers
  564. gint.Workers.reserve(num);
  565. for (unsigned int ii = 0; ii != num; ++ii) {
  566. gint.Workers.emplace_back(
  567. cm::make_unique<cmWorkerPoolWorker>(*gint.UVLoop));
  568. }
  569. // Start worker threads
  570. for (unsigned int ii = 0; ii != num; ++ii) {
  571. gint.Workers[ii]->SetThread(
  572. std::thread(&cmWorkerPoolInternal::Work, &gint, ii));
  573. }
  574. }
  575. // Destroy begin request
  576. gint.UVRequestBegin.reset();
  577. }
  578. void cmWorkerPoolInternal::UVSlotEnd(uv_async_t* handle)
  579. {
  580. auto& gint = *reinterpret_cast<cmWorkerPoolInternal*>(handle->data);
  581. // Join and destroy worker threads
  582. gint.Workers.clear();
  583. // Destroy end request
  584. gint.UVRequestEnd.reset();
  585. }
  586. void cmWorkerPoolInternal::Work(unsigned int workerIndex)
  587. {
  588. cmWorkerPool::JobHandleT jobHandle;
  589. std::unique_lock<std::mutex> uLock(this->Mutex);
  590. // Increment running workers count
  591. ++this->WorkersRunning;
  592. // Enter worker main loop
  593. while (true) {
  594. // Abort on request
  595. if (this->Aborting) {
  596. break;
  597. }
  598. // Wait for new jobs on the main CV
  599. if (this->Queue.empty()) {
  600. ++this->WorkersIdle;
  601. this->Condition.wait(uLock);
  602. --this->WorkersIdle;
  603. continue;
  604. }
  605. // If there is a fence currently active or waiting,
  606. // sleep on the main CV and try again.
  607. if (this->FenceProcessing) {
  608. this->Condition.wait(uLock);
  609. continue;
  610. }
  611. // Pop next job from queue
  612. jobHandle = std::move(this->Queue.front());
  613. this->Queue.pop_front();
  614. // Check for fence jobs
  615. bool raisedFence = false;
  616. if (jobHandle->IsFence()) {
  617. this->FenceProcessing = true;
  618. raisedFence = true;
  619. // Wait on the Fence CV until all pending jobs are done.
  620. while (this->JobsProcessing != 0 && !this->Aborting) {
  621. this->ConditionFence.wait(uLock);
  622. }
  623. // When aborting, explicitly kick all threads alive once more.
  624. if (this->Aborting) {
  625. this->FenceProcessing = false;
  626. this->Condition.notify_all();
  627. break;
  628. }
  629. }
  630. // Unlocked scope for job processing
  631. ++this->JobsProcessing;
  632. {
  633. uLock.unlock();
  634. jobHandle->Work(this->Pool, workerIndex); // Process job
  635. jobHandle.reset(); // Destroy job
  636. uLock.lock();
  637. }
  638. --this->JobsProcessing;
  639. // If this was the thread that entered fence processing
  640. // originally, notify all idling workers that the fence
  641. // is done.
  642. if (raisedFence) {
  643. this->FenceProcessing = false;
  644. this->Condition.notify_all();
  645. }
  646. // If fence processing is still not done, notify the
  647. // the fencing worker when all active jobs are done.
  648. if (this->FenceProcessing && this->JobsProcessing == 0) {
  649. this->ConditionFence.notify_all();
  650. }
  651. }
  652. // Decrement running workers count
  653. if (--this->WorkersRunning == 0) {
  654. // Last worker thread about to finish. Send libuv event.
  655. this->UVRequestEnd.send();
  656. }
  657. }
  658. cmWorkerPool::JobT::~JobT() = default;
  659. bool cmWorkerPool::JobT::RunProcess(ProcessResultT& result,
  660. std::vector<std::string> const& command,
  661. std::string const& workingDirectory)
  662. {
  663. // Get worker by index
  664. auto* worker = this->Pool_->Int_->Workers.at(this->WorkerIndex_).get();
  665. return worker->RunProcess(result, command, workingDirectory);
  666. }
  667. cmWorkerPool::cmWorkerPool()
  668. : Int_(cm::make_unique<cmWorkerPoolInternal>(this))
  669. {
  670. }
  671. cmWorkerPool::~cmWorkerPool() = default;
  672. void cmWorkerPool::SetThreadCount(unsigned int threadCount)
  673. {
  674. if (!this->Int_->Processing) {
  675. this->ThreadCount_ = (threadCount > 0) ? threadCount : 1u;
  676. }
  677. }
  678. bool cmWorkerPool::Process(void* userData)
  679. {
  680. // Setup user data
  681. this->UserData_ = userData;
  682. // Run libuv loop
  683. bool success = this->Int_->Process();
  684. // Clear user data
  685. this->UserData_ = nullptr;
  686. // Return
  687. return success;
  688. }
  689. bool cmWorkerPool::PushJob(JobHandleT&& jobHandle)
  690. {
  691. return this->Int_->PushJob(std::move(jobHandle));
  692. }
  693. void cmWorkerPool::Abort()
  694. {
  695. this->Int_->Abort();
  696. }