Source/cmWorkerPool.h

   1 /* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
   2    file Copyright.txt or https://cmake.org/licensing for details.  */
   3 #pragma once
   4
   5 #include "cmConfigure.h" // IWYU pragma: keep
   6
   7 #include <cstdint>
   8 #include <string>
   9 #include <utility>
  10 #include <vector>
  11
  12 #include <cm/memory>
  13
  14 // -- Types
  15 class cmWorkerPoolInternal;
  16
  17 /** @class cmWorkerPool
  18  * @brief Thread pool with job queue
  19  */
  20 class cmWorkerPool
  21 {
  22 public:
  23   /**
  24    * Return value and output of an external process.
  25    */
  26   struct ProcessResultT
  27   {
  28     void reset();
  29     bool error() const
  30     {
  31       return (this->ExitStatus != 0) || (this->TermSignal != 0) ||
  32         !this->ErrorMessage.empty();
  33     }
  34
  35     std::int64_t ExitStatus = 0;
  36     int TermSignal = 0;
  37     std::string StdOut;
  38     std::string StdErr;
  39     std::string ErrorMessage;
  40   };
  41
  42   /**
  43    * Abstract job class for concurrent job processing.
  44    */
  45   class JobT
  46   {
  47   public:
  48     JobT(JobT const&) = delete;
  49     JobT& operator=(JobT const&) = delete;
  50
  51     /**
  52      * Virtual destructor.
  53      */
  54     virtual ~JobT();
  55
  56     /**
  57      * Fence job flag
  58      *
  59      * Fence jobs require that:
  60      * - all jobs before in the queue have been processed
  61      * - no jobs later in the queue will be processed before this job was
  62      *   processed
  63      */
  64     bool IsFence() const { return this->Fence_; }
  65
  66   protected:
  67     /**
  68      * Protected default constructor
  69      */
  70     JobT(bool fence = false)
  71       : Fence_(fence)
  72     {
  73     }
  74
  75     /**
  76      * Abstract processing interface that must be implement in derived classes.
  77      */
  78     virtual void Process() = 0;
  79
  80     /**
  81      * Get the worker pool.
  82      * Only valid during the JobT::Process() call!
  83      */
  84     cmWorkerPool* Pool() const { return this->Pool_; }
  85
  86     /**
  87      * Get the user data.
  88      * Only valid during the JobT::Process() call!
  89      */
  90     void* UserData() const { return this->Pool_->UserData(); }
  91
  92     /**
  93      * Get the worker index.
  94      * This is the index of the thread processing this job and is in the range
  95      * [0..ThreadCount).
  96      * Concurrently processing jobs will never have the same WorkerIndex().
  97      * Only valid during the JobT::Process() call!
  98      */
  99     unsigned int WorkerIndex() const { return this->WorkerIndex_; }
 100
 101     /**
 102      * Run an external read only process.
 103      * Use only during JobT::Process() call!
 104      */
 105     bool RunProcess(ProcessResultT& result,
 106                     std::vector<std::string> const& command,
 107                     std::string const& workingDirectory);
 108
 109   private:
 110     //! Needs access to Work()
 111     friend class cmWorkerPoolInternal;
 112     //! Worker thread entry method.
 113     void Work(cmWorkerPool* pool, unsigned int workerIndex)
 114     {
 115       this->Pool_ = pool;
 116       this->WorkerIndex_ = workerIndex;
 117       this->Process();
 118     }
 119
 120     cmWorkerPool* Pool_ = nullptr;
 121     unsigned int WorkerIndex_ = 0;
 122     bool Fence_ = false;
 123   };
 124
 125   /**
 126    * Job handle type
 127    */
 128   using JobHandleT = std::unique_ptr<JobT>;
 129
 130   /**
 131    * Fence job base class
 132    */
 133   class JobFenceT : public JobT
 134   {
 135   public:
 136     JobFenceT()
 137       : JobT(true)
 138     {
 139     }
 140     //! Does nothing
 141     void Process() override {}
 142   };
 143
 144   /**
 145    * Fence job that aborts the worker pool.
 146    *
 147    * Useful as the last job in the job queue.
 148    */
 149   class JobEndT : JobFenceT
 150   {
 151   public:
 152     //! Does nothing
 153     void Process() override { this->Pool()->Abort(); }
 154   };
 155
 156   // -- Methods
 157   cmWorkerPool();
 158   ~cmWorkerPool();
 159
 160   /**
 161    * Number of worker threads.
 162    */
 163   unsigned int ThreadCount() const { return this->ThreadCount_; }
 164
 165   /**
 166    * Set the number of worker threads.
 167    *
 168    * Calling this method during Process() has no effect.
 169    */
 170   void SetThreadCount(unsigned int threadCount);
 171
 172   /**
 173    * Blocking function that starts threads to process all Jobs in the queue.
 174    *
 175    * This method blocks until a job calls the Abort() method.
 176    * @arg threadCount Number of threads to process jobs.
 177    * @arg userData Common user data pointer available in all Jobs.
 178    */
 179   bool Process(void* userData = nullptr);
 180
 181   /**
 182    * User data reference passed to Process().
 183    *
 184    * Only valid during Process().
 185    */
 186   void* UserData() const { return this->UserData_; }
 187
 188   // -- Job processing interface
 189
 190   /**
 191    * Clears the job queue and aborts all worker threads.
 192    *
 193    * This method is thread safe and can be called from inside a job.
 194    */
 195   void Abort();
 196
 197   /**
 198    * Push job to the queue.
 199    *
 200    * This method is thread safe and can be called from inside a job or before
 201    * Process().
 202    */
 203   bool PushJob(JobHandleT&& jobHandle);
 204
 205   /**
 206    * Push job to the queue
 207    *
 208    * This method is thread safe and can be called from inside a job or before
 209    * Process().
 210    */
 211   template <class T, typename... Args>
 212   bool EmplaceJob(Args&&... args)
 213   {
 214     return this->PushJob(cm::make_unique<T>(std::forward<Args>(args)...));
 215   }
 216
 217 private:
 218   void* UserData_ = nullptr;
 219   unsigned int ThreadCount_ = 1;
 220   std::unique_ptr<cmWorkerPoolInternal> Int_;
 221 };