dali/devel-api/threading/thread-pool.h

   1 #ifndef DALI_THREAD_POOL_H
   2 #define DALI_THREAD_POOL_H
   3
   4 /*
   5  * Copyright (c) 2019 Samsung Electronics Co., Ltd.
   6  *
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  * http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  */
  20
  21 // INTERNAL INCLUDES
  22 #include <dali/public-api/common/dali-common.h>
  23
  24 // EXTERNAL INCLUDES
  25 #include <thread>
  26 #include <functional>
  27 #include <mutex>
  28 #include <queue>
  29 #include <condition_variable>
  30 #include <future>
  31 #include <algorithm>
  32 #include <iostream>
  33 #include <memory>
  34
  35 namespace Dali
  36 {
  37 using Task = std::function<void(uint32_t)>;
  38
  39 using TaskQueue = std::queue<Task>;
  40
  41 /**
  42  * Future contains the result of submitted task. When queried
  43  * it applies internal synchronization mechanism to make sure
  44  * the value is available.
  45  */
  46 template<typename T>
  47 class DALI_INTERNAL Future final
  48 {
  49   friend class ThreadPool;
  50
  51 public:
  52
  53   /**
  54    * @brief Constructor of Future
  55    */
  56   Future()
  57   {
  58     mFuture = mPromise.get_future();
  59   }
  60
  61   /**
  62    * @brief Destructor of Future
  63    */
  64   ~Future()
  65   {
  66     Wait();
  67   }
  68
  69   /**
  70    * @brief Returns value of future, blocks if needed.
  71    * @return Value stored by the future
  72    */
  73   T Get() const
  74   {
  75     return mFuture.get();
  76   }
  77
  78   /**
  79    * @brief Waits until the value of future is ready. This function
  80    * is a fencing mechanism.
  81    */
  82   void Wait() const
  83   {
  84     if( IsValid() )
  85     {
  86       mFuture.wait();
  87     }
  88   }
  89
  90   /**
  91    * @brief Tests whether the future is valid
  92    * @return True if valid, False otherwise
  93    */
  94   bool IsValid() const
  95   {
  96     return mFuture.valid();
  97   }
  98
  99   /**
 100    * @brief Resets the future bringing it to the initial state.
 101    * It's required in order to reuse the same Future object.
 102    */
 103   void Reset()
 104   {
 105     mPromise = std::promise<T>();
 106     mFuture  = mPromise.get_future();
 107   }
 108
 109 private:
 110
 111   std::promise<T> mPromise{};
 112   std::future<T>  mFuture{};
 113 };
 114
 115 using SharedFuture = std::shared_ptr<Future<void>>;
 116
 117 /**
 118  * FutureGroup binds many Future objects and applies synchronization.
 119  */
 120 template<typename T>
 121 class FutureGroup final
 122 {
 123   friend class ThreadPool;
 124
 125 public:
 126
 127   /**
 128    * Waits for all the Futures to complete.
 129    */
 130   void Wait()
 131   {
 132     for (auto &future : mFutures)
 133     {
 134       future->Wait();
 135     }
 136   }
 137
 138 private:
 139
 140   std::vector<std::shared_ptr<Future<T> > > mFutures;
 141 };
 142
 143 using UniqueFutureGroup = std::unique_ptr<FutureGroup<void>>;
 144
 145
 146
 147 /**
 148  * ThreadPool creates and manages worker threads and tasks submitted for execution.
 149  */
 150 class DALI_CORE_API ThreadPool final
 151 {
 152 public:
 153
 154   /**
 155    * @brief Constructor of thread pool.
 156    */
 157   ThreadPool();
 158
 159   /**
 160    * @brief Destructor of thread pool.
 161    */
 162   ~ThreadPool();
 163
 164   /**
 165    * @brief Intializes thread pool
 166    * @param threadCount Number of worker threads to use. If 0 then thread count equals hardware thread count.
 167    * @return True if success
 168    */
 169   bool Initialize( uint32_t threadCount = 0u );
 170
 171   /**
 172    * @brief Waits until all threads finish execution and go back to the idle state.
 173    */
 174   void Wait();
 175
 176   /**
 177    * @brief Submits a single task to specified ( by the index ) worker thread.
 178    * @param workerIndex Index of thread to be used
 179    * @param task Task submitted for execution
 180    * @return Shared pointer to the Future object
 181    */
 182   SharedFuture SubmitTask(uint32_t workerIndex, const Task &task);
 183
 184   /**
 185    * @brief Submits vector of tasks to the pool
 186    * @param tasks Vector containing tasks to be executed
 187    * @return Shared pointer to the Future object
 188    */
 189   SharedFuture SubmitTasks(const std::vector<Task>& tasks);
 190
 191   /**
 192    * @brief Submits tasks to threads specified by thread mask.
 193    * @param tasks Vector of tasks
 194    * @param threadMask Mask of threads to be used or 0 to use all available threads
 195    * @return Unique pointer to the FutureGroup object
 196    */
 197   UniqueFutureGroup SubmitTasks(const std::vector<Task>& tasks, uint32_t threadMask);
 198
 199   /**
 200    * @brief Returns number of worker threads
 201    * @return Number of worker threads
 202    */
 203   size_t GetWorkerCount() const;
 204
 205 private:
 206
 207   struct Impl;
 208   std::unique_ptr<Impl> mImpl;
 209 };
 210
 211 } //namespace Dali
 212
 213 #endif // DALI_THREAD_POOL_H