\r
#pragma once\r
\r
-#include "../exception/win32_exception.h"\r
-#include "../utility/move_on_copy.h"\r
+#include "../except.h"\r
+#include "../enum_class.h"\r
#include "../log.h"\r
\r
-#include <Windows.h>\r
-\r
#include <tbb/atomic.h>\r
#include <tbb/concurrent_queue.h>\r
\r
#include <boost/thread.hpp>\r
-#include <boost/optional.hpp>\r
-#include <boost/noncopyable.hpp>\r
\r
#include <functional>\r
\r
namespace caspar {\r
\r
namespace detail {\r
-\r
-//typedef struct tagTHREADNAME_INFO\r
-//{\r
-// DWORD dwType; // must be 0x1000\r
-// LPCSTR szName; // pointer to name (in user addr space)\r
-// DWORD dwThreadID; // thread ID (-1=caller thread)\r
-// DWORD dwFlags; // reserved for future use, must be zero\r
-//} THREADNAME_INFO;\r
-//\r
-//inline void SetThreadName(DWORD dwThreadID, LPCSTR szThreadName)\r
-//{\r
-// THREADNAME_INFO info;\r
-// {\r
-// info.dwType = 0x1000;\r
-// info.szName = szThreadName;\r
-// info.dwThreadID = dwThreadID;\r
-// info.dwFlags = 0;\r
-// }\r
-// __try\r
-// {\r
-// RaiseException( 0x406D1388, 0, sizeof(info)/sizeof(DWORD), (DWORD*)&info );\r
-// }\r
-// __except (EXCEPTION_CONTINUE_EXECUTION){} \r
-//}\r
-//\r
-//}\r
-}\r
-\r
-enum task_priority\r
+ \r
+template<typename T>\r
+struct move_on_copy\r
{\r
- high_priority,\r
- normal_priority,\r
- priority_count\r
+ move_on_copy(const move_on_copy<T>& other) : value(std::move(other.value)){}\r
+ move_on_copy(T&& value) : value(std::move(value)){}\r
+ mutable T value;\r
};\r
\r
-enum thread_priority\r
+template<typename T>\r
+move_on_copy<T> make_move_on_copy(T&& value)\r
+{\r
+ return move_on_copy<T>(std::move(value));\r
+}\r
+\r
+}\r
+ \r
+struct task_priority_def\r
{\r
- high_priority_class,\r
- above_normal_priority_class,\r
- normal_priority_class,\r
- below_normal_priority_class\r
+ enum type\r
+ {\r
+ high_priority,\r
+ normal_priority,\r
+ priority_count\r
+ };\r
};\r
+typedef enum_class<task_priority_def> task_priority;\r
\r
-class executor : boost::noncopyable\r
+class executor\r
{\r
- const std::string name_;\r
- boost::thread thread_;\r
- tbb::atomic<bool> is_running_;\r
+ executor(const executor&);\r
+ executor& operator=(const executor&);\r
+ \r
+ tbb::atomic<bool> is_running_;\r
+ boost::thread thread_;\r
\r
typedef tbb::concurrent_bounded_queue<std::function<void()>> function_queue;\r
- function_queue execution_queue_[priority_count];\r
+ function_queue execution_queue_[task_priority::priority_count];\r
\r
template<typename Func>\r
auto create_task(Func&& func) -> boost::packaged_task<decltype(func())> // noexcept\r
return std::move(task);\r
}\r
\r
-public:\r
- \r
- explicit executor(const std::wstring& name) : name_(u8(name)) // noexcept\r
+public: \r
+ executor(const std::wstring& name) // noexcept\r
{\r
+ name; // TODO: Use to set thread name.\r
is_running_ = true;\r
thread_ = boost::thread([this]{run();});\r
}\r
\r
void set_capacity(size_t capacity) // noexcept\r
{\r
- execution_queue_[normal_priority].set_capacity(capacity);\r
- }\r
-\r
- void set_priority_class(thread_priority p)\r
- {\r
- begin_invoke([=]\r
- {\r
- if(p == high_priority_class)\r
- SetThreadPriority(GetCurrentThread(), HIGH_PRIORITY_CLASS);\r
- else if(p == above_normal_priority_class)\r
- SetThreadPriority(GetCurrentThread(), ABOVE_NORMAL_PRIORITY_CLASS);\r
- else if(p == normal_priority_class)\r
- SetThreadPriority(GetCurrentThread(), NORMAL_PRIORITY_CLASS);\r
- else if(p == below_normal_priority_class)\r
- SetThreadPriority(GetCurrentThread(), BELOW_NORMAL_PRIORITY_CLASS);\r
- });\r
+ execution_queue_[task_priority::normal_priority].set_capacity(capacity);\r
}\r
\r
void clear()\r
{ \r
std::function<void()> func;\r
- while(execution_queue_[normal_priority].try_pop(func));\r
- while(execution_queue_[high_priority].try_pop(func));\r
+ while(execution_queue_[task_priority::normal_priority].try_pop(func));\r
+ while(execution_queue_[task_priority::high_priority].try_pop(func));\r
}\r
\r
void stop() // noexcept\r
{\r
is_running_ = false; \r
- execution_queue_[normal_priority].try_push([]{}); // Wake the execution thread.\r
+ execution_queue_[task_priority::normal_priority].try_push([]{}); // Wake the execution thread.\r
}\r
\r
void wait() // noexcept\r
\r
void join()\r
{\r
- if(boost::this_thread::get_id() != thread_.get_id())\r
- thread_.join();\r
+ if(boost::this_thread::get_id() == thread_.get_id())\r
+ BOOST_THROW_EXCEPTION(invalid_operation());\r
+\r
+ thread_.join();\r
}\r
\r
template<typename Func>\r
- auto begin_invoke(Func&& func, task_priority priority = normal_priority) -> boost::unique_future<decltype(func())> // noexcept\r
+ auto begin_invoke(Func&& func, task_priority priority = task_priority::normal_priority) -> boost::unique_future<decltype(func())> // noexcept\r
{ \r
if(!is_running_)\r
BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("executor not running."));\r
\r
// Create a move on copy adaptor to avoid copying the functor into the queue, tbb::concurrent_queue does not support move semantics.\r
- auto task_adaptor = make_move_on_copy(create_task(func));\r
+ auto task_adaptor = detail::make_move_on_copy(create_task(func));\r
\r
auto future = task_adaptor.value.get_future();\r
\r
- execution_queue_[priority].push([=]\r
+ execution_queue_[priority.value()].push([=]\r
{\r
try\r
{\r
}\r
});\r
\r
- if(priority != normal_priority)\r
- execution_queue_[normal_priority].push(nullptr);\r
+ if(priority != task_priority::normal_priority)\r
+ execution_queue_[task_priority::normal_priority].push(nullptr);\r
\r
return std::move(future); \r
}\r
\r
template<typename Func>\r
- auto invoke(Func&& func, task_priority prioriy = normal_priority) -> decltype(func()) // noexcept\r
+ auto invoke(Func&& func, task_priority prioriy = task_priority::normal_priority) -> decltype(func()) // noexcept\r
{\r
if(boost::this_thread::get_id() == thread_.get_id()) // Avoids potential deadlock.\r
return func();\r
\r
return begin_invoke(std::forward<Func>(func), prioriy).get();\r
}\r
- \r
- bool yield() // noexcept\r
+\r
+ void yield() // noexcept\r
{\r
- if(boost::this_thread::get_id() != thread_.get_id()) // Only yield when calling from execution thread.\r
- return false;\r
+ if(boost::this_thread::get_id() != thread_.get_id())\r
+ BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("Executor can only yield inside of thread context."));\r
\r
std::function<void()> func;\r
- while(execution_queue_[high_priority].try_pop(func))\r
+ execution_queue_[task_priority::normal_priority].pop(func); \r
+ \r
+ std::function<void()> func2;\r
+ while(execution_queue_[task_priority::high_priority].try_pop(func2))\r
{\r
- if(func)\r
- func();\r
+ if(func2)\r
+ func2();\r
} \r
\r
- if(!func)\r
- {\r
- execution_queue_[normal_priority].try_pop(func);\r
- if(func)\r
- func();\r
- }\r
-\r
- return func != nullptr;\r
+ if(func)\r
+ func();\r
}\r
- \r
- function_queue::size_type capacity() const /*noexcept*/ { return execution_queue_[normal_priority].capacity(); }\r
- function_queue::size_type size() const /*noexcept*/ { return execution_queue_[normal_priority].size(); }\r
- bool empty() const /*noexcept*/ { return execution_queue_[normal_priority].empty(); }\r
- bool is_running() const /*noexcept*/ { return is_running_; } \r
\r
-private:\r
- \r
- void execute() // noexcept\r
+ function_queue::size_type size() const /*noexcept*/\r
{\r
- std::function<void()> func;\r
- execution_queue_[normal_priority].pop(func); \r
-\r
- yield();\r
-\r
- if(func)\r
- func();\r
+ return execution_queue_[task_priority::normal_priority].size() + execution_queue_[task_priority::high_priority].size(); \r
}\r
+ \r
+ bool is_running() const /*noexcept*/ { return is_running_; } \r
+ \r
+private: \r
\r
void run() // noexcept\r
{\r
win32_exception::install_handler(); \r
- //detail::SetThreadName(GetCurrentThreadId(), name_.c_str());\r
while(is_running_)\r
{\r
try\r
{\r
- execute();\r
+ yield();\r
}\r
catch(...)\r
{\r