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

   1 #ifndef DALI_THREAD_POOL_H
   2 #define DALI_THREAD_POOL_H
   3
   4 /*
   5  * Copyright (c) 2020 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 <algorithm>
  26 #include <condition_variable>
  27 #include <functional>
  28 #include <future>
  29 #include <iostream>
  30 #include <memory>
  31 #include <mutex>
  32 #include <queue>
  33 #include <thread>
  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    * @brief Constructor of Future
  54    */
  55   Future()
  56   {
  57     mFuture = mPromise.get_future();
  58   }
  59
  60   /**
  61    * @brief Destructor of Future
  62    */
  63   ~Future()
  64   {
  65     Wait();
  66   }
  67
  68   /**
  69    * @brief Returns value of future, blocks if needed.
  70    * @return Value stored by the future
  71    */
  72   T Get() const
  73   {
  74     return mFuture.get();
  75   }
  76
  77   /**
  78    * @brief Waits until the value of future is ready. This function
  79    * is a fencing mechanism.
  80    */
  81   void Wait() const
  82   {
  83     if(IsValid())
  84     {
  85       mFuture.wait();
  86     }
  87   }
  88
  89   /**
  90    * @brief Tests whether the future is valid
  91    * @return True if valid, False otherwise
  92    */
  93   bool IsValid() const
  94   {
  95     return mFuture.valid();
  96   }
  97
  98   /**
  99    * @brief Resets the future bringing it to the initial state.
 100    * It's required in order to reuse the same Future object.
 101    */
 102   void Reset()
 103   {
 104     mPromise = std::promise<T>();
 105     mFuture  = mPromise.get_future();
 106   }
 107
 108 private:
 109   std::promise<T> mPromise{};
 110   std::future<T>  mFuture{};
 111 };
 112
 113 using SharedFuture = std::shared_ptr<Future<void>>;
 114
 115 /**
 116  * FutureGroup binds many Future objects and applies synchronization.
 117  */
 118 template<typename T>
 119 class FutureGroup final
 120 {
 121   friend class ThreadPool;
 122
 123 public:
 124   /**
 125    * Waits for all the Futures to complete.
 126    */
 127   void Wait()
 128   {
 129     for(auto& future : mFutures)
 130     {
 131       future->Wait();
 132     }
 133   }
 134
 135 private:
 136   std::vector<std::shared_ptr<Future<T>>> mFutures;
 137 };
 138
 139 using UniqueFutureGroup = std::unique_ptr<FutureGroup<void>>;
 140
 141 /**
 142  * ThreadPool creates and manages worker threads and tasks submitted for execution.
 143  */
 144 class DALI_CORE_API ThreadPool final
 145 {
 146 public:
 147   /**
 148    * @brief Constructor of thread pool.
 149    */
 150   ThreadPool();
 151
 152   /**
 153    * @brief Destructor of thread pool.
 154    */
 155   ~ThreadPool();
 156
 157   /**
 158    * @brief Intializes thread pool
 159    * @param threadCount Number of worker threads to use. If 0 then thread count equals hardware thread count.
 160    * @return True if success
 161    */
 162   bool Initialize(uint32_t threadCount = 0u);
 163
 164   /**
 165    * @brief Waits until all threads finish execution and go back to the idle state.
 166    */
 167   void Wait();
 168
 169   /**
 170    * @brief Submits a single task to specified ( by the index ) worker thread.
 171    * @param workerIndex Index of thread to be used
 172    * @param task Task submitted for execution
 173    * @return Shared pointer to the Future object
 174    */
 175   SharedFuture SubmitTask(uint32_t workerIndex, const Task& task);
 176
 177   /**
 178    * @brief Submits vector of tasks to the pool
 179    * @param tasks Vector containing tasks to be executed
 180    * @return Shared pointer to the Future object
 181    */
 182   SharedFuture SubmitTasks(const std::vector<Task>& tasks);
 183
 184   /**
 185    * @brief Submits tasks to threads specified by thread mask.
 186    * @param tasks Vector of tasks
 187    * @param threadMask Mask of threads to be used or 0 to use all available threads
 188    * @return Unique pointer to the FutureGroup object
 189    */
 190   UniqueFutureGroup SubmitTasks(const std::vector<Task>& tasks, uint32_t threadMask);
 191
 192   /**
 193    * @brief Returns number of worker threads
 194    * @return Number of worker threads
 195    */
 196   size_t GetWorkerCount() const;
 197
 198 private:
 199   struct Impl;
 200   std::unique_ptr<Impl> mImpl;
 201 };
 202
 203 } //namespace Dali
 204
 205 #endif // DALI_THREAD_POOL_H