git.sesse.net Git - casparcg/blob - common/concurrency/executor.h

   1 /*\r
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>\r
   3 *\r
   4 * This file is part of CasparCG (www.casparcg.com).\r
   5 *\r
   6 * CasparCG is free software: you can redistribute it and/or modify\r
   7 * it under the terms of the GNU General Public License as published by\r
   8 * the Free Software Foundation, either version 3 of the License, or\r
   9 * (at your option) any later version.\r
  10 *\r
  11 * CasparCG is distributed in the hope that it will be useful,\r
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of\r
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\r
  14 * GNU General Public License for more details.\r
  15 *\r
  16 * You should have received a copy of the GNU General Public License\r
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
  18 *\r
  19 * Author: Robert Nagy, ronag89@gmail.com\r
  20 */\r
  21 \r
  22 #pragma once\r
  23 \r
  24 #include "../except.h"\r
  25 #include "../log.h"\r
  26 #include "../enum_class.h"\r
  27 #include "../utility/move_on_copy.h"\r
  28 #include "../os/windows/windows.h"\r
  29 \r
  30 #include <tbb/atomic.h>\r
  31 #include <tbb/concurrent_queue.h>\r
  32 \r
  33 #include <boost/thread.hpp>\r
  34 \r
  35 #include <functional>\r
  36 \r
  37 namespace caspar {\r
  38         \r
  39 struct task_priority_def\r
  40 {\r
  41         enum type\r
  42         {\r
  43                 high_priority,\r
  44                 normal_priority,\r
  45                 priority_count\r
  46         };\r
  47 };\r
  48 typedef enum_class<task_priority_def> task_priority;\r
  49 \r
  50 struct thread_priority_def\r
  51 {\r
  52         enum type\r
  53         {\r
  54                 high_priority_class,\r
  55                 above_normal_priority_class,\r
  56                 normal_priority_class,\r
  57                 below_normal_priority_class\r
  58         };\r
  59 };\r
  60 typedef enum_class<thread_priority_def> thread_priority;\r
  61 \r
  62 class executor\r
  63 {\r
  64         executor(const executor&);\r
  65         executor& operator=(const executor&);\r
  66 \r
  67         const std::string name_;\r
  68         boost::thread thread_;\r
  69         tbb::atomic<bool> is_running_;\r
  70         \r
  71         typedef tbb::concurrent_bounded_queue<std::function<void()>> function_queue;\r
  72         function_queue execution_queue_[task_priority::priority_count];\r
  73                 \r
  74         template<typename Func>\r
  75         auto create_task(Func&& func) -> boost::packaged_task<decltype(func())> // noexcept\r
  76         {       \r
  77                 typedef boost::packaged_task<decltype(func())> task_type;\r
  78                                 \r
  79                 auto task = task_type(std::forward<Func>(func));\r
  80                 \r
  81                 task.set_wait_callback(std::function<void(task_type&)>([=](task_type& my_task) // The std::function wrapper is required in order to add ::result_type to functor class.\r
  82                 {\r
  83                         try\r
  84                         {\r
  85                                 if(boost::this_thread::get_id() == thread_.get_id())  // Avoids potential deadlock.\r
  86                                         my_task();\r
  87                         }\r
  88                         catch(boost::task_already_started&){}\r
  89                 }));\r
  90                                 \r
  91                 return std::move(task);\r
  92         }\r
  93 \r
  94 public:\r
  95                 \r
  96         explicit executor(const std::wstring& name) : name_(u8(name)) // noexcept\r
  97         {\r
  98                 is_running_ = true;\r
  99                 thread_ = boost::thread([this]{run();});\r
 100         }\r
 101         \r
 102         virtual ~executor() // noexcept\r
 103         {\r
 104                 stop();\r
 105                 join();\r
 106         }\r
 107 \r
 108         void set_capacity(size_t capacity) // noexcept\r
 109         {\r
 110                 execution_queue_[task_priority::normal_priority].set_capacity(capacity);\r
 111         }\r
 112 \r
 113         void set_priority_class(thread_priority p)\r
 114         {\r
 115                 begin_invoke([=]\r
 116                 {\r
 117                         if(p == thread_priority::high_priority_class)\r
 118                                 SetThreadPriority(GetCurrentThread(), HIGH_PRIORITY_CLASS);\r
 119                         else if(p == thread_priority::above_normal_priority_class)\r
 120                                 SetThreadPriority(GetCurrentThread(), ABOVE_NORMAL_PRIORITY_CLASS);\r
 121                         else if(p == thread_priority::normal_priority_class)\r
 122                                 SetThreadPriority(GetCurrentThread(), NORMAL_PRIORITY_CLASS);\r
 123                         else if(p == thread_priority::below_normal_priority_class)\r
 124                                 SetThreadPriority(GetCurrentThread(), BELOW_NORMAL_PRIORITY_CLASS);\r
 125                 });\r
 126         }\r
 127         \r
 128         void clear()\r
 129         {               \r
 130                 std::function<void()> func;\r
 131                 while(execution_queue_[task_priority::normal_priority].try_pop(func));\r
 132                 while(execution_queue_[task_priority::high_priority].try_pop(func));\r
 133         }\r
 134                                 \r
 135         void stop() // noexcept\r
 136         {\r
 137                 is_running_ = false;    \r
 138                 execution_queue_[task_priority::normal_priority].try_push([]{}); // Wake the execution thread.\r
 139         }\r
 140 \r
 141         void wait() // noexcept\r
 142         {\r
 143                 invoke([]{});\r
 144         }\r
 145 \r
 146         void join()\r
 147         {\r
 148                 if(boost::this_thread::get_id() != thread_.get_id())\r
 149                         thread_.join();\r
 150         }\r
 151                                 \r
 152         template<typename Func>\r
 153         auto begin_invoke(Func&& func, task_priority priority = task_priority::normal_priority) -> boost::unique_future<decltype(func())> // noexcept\r
 154         {       \r
 155                 if(!is_running_)\r
 156                         BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("executor not running."));\r
 157 \r
 158                 // Create a move on copy adaptor to avoid copying the functor into the queue, tbb::concurrent_queue does not support move semantics.\r
 159                 auto task_adaptor = make_move_on_copy(create_task(func));\r
 160 \r
 161                 auto future = task_adaptor.value.get_future();\r
 162 \r
 163                 execution_queue_[priority.value()].push([=]\r
 164                 {\r
 165                         try\r
 166                         {\r
 167                                 task_adaptor.value();\r
 168                         }\r
 169                         catch(boost::task_already_started&)\r
 170                         {\r
 171                         }\r
 172                         catch(...)\r
 173                         {\r
 174                                 CASPAR_LOG_CURRENT_EXCEPTION();\r
 175                         }\r
 176                 });\r
 177 \r
 178                 if(priority != task_priority::normal_priority)\r
 179                         execution_queue_[task_priority::normal_priority].push(nullptr);\r
 180                                         \r
 181                 return std::move(future);               \r
 182         }\r
 183         \r
 184         template<typename Func>\r
 185         auto invoke(Func&& func, task_priority prioriy = task_priority::normal_priority) -> decltype(func()) // noexcept\r
 186         {\r
 187                 if(boost::this_thread::get_id() == thread_.get_id())  // Avoids potential deadlock.\r
 188                         return func();\r
 189                 \r
 190                 return begin_invoke(std::forward<Func>(func), prioriy).get();\r
 191         }\r
 192 \r
 193         void yield() // noexcept\r
 194         {\r
 195                 if(boost::this_thread::get_id() != thread_.get_id())\r
 196                         BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("Executor can only yield inside of thread context."));\r
 197 \r
 198                 std::function<void()> func;\r
 199                 execution_queue_[task_priority::normal_priority].pop(func);     \r
 200                 \r
 201                 std::function<void()> func2;\r
 202                 while(execution_queue_[task_priority::high_priority].try_pop(func2))\r
 203                 {\r
 204                         if(func2)\r
 205                                 func2();\r
 206                 }       \r
 207 \r
 208                 if(func)\r
 209                         func();\r
 210         }\r
 211                 \r
 212         function_queue::size_type size() const /*noexcept*/\r
 213         {\r
 214                 return execution_queue_[task_priority::normal_priority].size() + execution_queue_[task_priority::high_priority].size(); \r
 215         }\r
 216                 \r
 217         bool is_running() const /*noexcept*/ { return is_running_; }    \r
 218                 \r
 219 private:        \r
 220 \r
 221         void run() // noexcept\r
 222         {\r
 223                 win32_exception::install_handler();             \r
 224                 //detail::SetThreadName(GetCurrentThreadId(), name_.c_str());\r
 225                 while(is_running_)\r
 226                 {\r
 227                         try\r
 228                         {\r
 229                                 yield();\r
 230                         }\r
 231                         catch(...)\r
 232                         {\r
 233                                 CASPAR_LOG_CURRENT_EXCEPTION();\r
 234                         }\r
 235                 }\r
 236         }       \r
 237 };\r
 238 \r
 239 }