+/*\r
+* copyright (c) 2010 Sveriges Television AB <info@casparcg.com>\r
+*\r
+* This file is part of CasparCG.\r
+*\r
+* CasparCG is free software: you can redistribute it and/or modify\r
+* it under the terms of the GNU General Public License as published by\r
+* the Free Software Foundation, either version 3 of the License, or\r
+* (at your option) any later version.\r
+*\r
+* CasparCG is distributed in the hope that it will be useful,\r
+* but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the\r
+* GNU General Public License for more details.\r
+\r
+* You should have received a copy of the GNU General Public License\r
+* along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
+*\r
+*/\r
#pragma once\r
\r
#include "../exception/win32_exception.h"\r
+#include "../utility/string.h"\r
+#include "../utility/move_on_copy.h"\r
#include "../log/log.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
-#include <array>\r
\r
namespace caspar {\r
\r
-class executor : boost::noncopyable\r
+namespace detail {\r
+\r
+typedef struct tagTHREADNAME_INFO\r
{\r
-public:\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
- enum stop_policy\r
- {\r
- wait,\r
- no_wait\r
- };\r
- \r
- explicit executor(const std::function<void()>& f = nullptr)\r
+inline void SetThreadName(DWORD dwThreadID, LPCSTR szThreadName)\r
+{\r
+ THREADNAME_INFO info;\r
{\r
- is_running_ = false;\r
- f_ = f != nullptr ? f : [this]{run();};\r
+ info.dwType = 0x1000;\r
+ info.szName = szThreadName;\r
+ info.dwThreadID = dwThreadID;\r
+ info.dwFlags = 0;\r
}\r
-\r
- ~executor()\r
+ __try\r
{\r
- stop();\r
+ RaiseException( 0x406D1388, 0, sizeof(info)/sizeof(DWORD), (DWORD*)&info );\r
}\r
+ __except (EXCEPTION_CONTINUE_EXECUTION){} \r
+}\r
\r
- void set_capacity(size_t capacity)\r
- {\r
- execution_queue_.set_capacity(capacity);\r
- }\r
+}\r
\r
- void start() // noexcept\r
- {\r
- if(is_running_.fetch_and_store(true))\r
- return;\r
- thread_ = boost::thread(f_);\r
- }\r
- \r
- void stop(stop_policy policy = wait) // noexcept\r
- {\r
- is_running_ = false; \r
- execution_queue_.push([]{});\r
- if(policy == wait && boost::this_thread::get_id() != thread_.get_id())\r
- thread_.join();\r
- }\r
+enum task_priority\r
+{\r
+ high_priority,\r
+ normal_priority,\r
+ priority_count\r
+};\r
\r
- void clear()\r
- {\r
- std::function<void()> func;\r
- while(execution_queue_.try_pop(func)){}\r
- }\r
- \r
+enum thread_priority\r
+{\r
+ high_priority_class,\r
+ above_normal_priority_class,\r
+ normal_priority_class,\r
+ below_normal_priority_class\r
+};\r
+\r
+class executor : boost::noncopyable\r
+{\r
+ const std::string name_;\r
+ boost::thread thread_;\r
+ tbb::atomic<bool> is_running_;\r
+ \r
+ typedef tbb::concurrent_bounded_queue<std::function<void()>> function_queue;\r
+ function_queue execution_queue_[priority_count];\r
+ \r
template<typename Func>\r
- auto begin_invoke(Func&& func) -> boost::unique_future<decltype(func())> // noexcept\r
+ auto create_task(Func&& func) -> boost::packaged_task<decltype(func())> // noexcept\r
{ \r
typedef boost::packaged_task<decltype(func())> task_type;\r
\r
auto task = task_type(std::forward<Func>(func));\r
- auto future = task.get_future();\r
\r
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
{\r
catch(boost::task_already_started&){}\r
}));\r
\r
- struct task_adaptor_t\r
+ return std::move(task);\r
+ }\r
+\r
+public:\r
+ \r
+ explicit executor(const std::wstring& name) : name_(narrow(name)) // noexcept\r
+ {\r
+ thread_ = boost::thread([this]{run();});\r
+ is_running_ = true;\r
+ }\r
+ \r
+ virtual ~executor() // noexcept\r
+ {\r
+ stop();\r
+ join();\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
- task_adaptor_t(const task_adaptor_t& other) : task(std::move(other.task)){}\r
- task_adaptor_t(task_type&& task) : task(std::move(task)){}\r
- void operator()() const { task(); }\r
- mutable task_type task;\r
- } task_adaptor(std::move(task));\r
+ if(p == high_priority_class)\r
+ SetThreadPriority(GetCurrentThread(), HIGH_PRIORITY_CLASS);\r
+ 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
+ }\r
+ \r
+ void stop() // noexcept\r
+ {\r
+ is_running_ = false; \r
+ execution_queue_[normal_priority].try_push([]{}); // Wake the execution thread.\r
+ }\r
+\r
+ void wait() // noexcept\r
+ {\r
+ invoke([]{});\r
+ }\r
+\r
+ void join()\r
+ {\r
+ if(boost::this_thread::get_id() != thread_.get_id())\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
+ { \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
\r
- execution_queue_.push([=]\r
+ auto future = task_adaptor.value.get_future();\r
+\r
+ execution_queue_[priority].push([=]\r
{\r
- try{task_adaptor();}\r
+ try{task_adaptor.value();}\r
catch(boost::task_already_started&){}\r
catch(...){CASPAR_LOG_CURRENT_EXCEPTION();}\r
});\r
\r
+ if(priority != normal_priority)\r
+ execution_queue_[normal_priority].push(nullptr);\r
+ \r
return std::move(future); \r
}\r
\r
template<typename Func>\r
- auto invoke(Func&& func) -> decltype(func())\r
+ auto invoke(Func&& func, task_priority prioriy = 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)).get();\r
+ return begin_invoke(std::forward<Func>(func), prioriy).get();\r
}\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;\r
\r
- tbb::concurrent_bounded_queue<std::function<void()>>::size_type size() const { return execution_queue_.size(); }\r
- bool empty() const { return execution_queue_.empty(); }\r
- bool is_running() const { return is_running_; } \r
+ std::function<void()> func;\r
+ while(execution_queue_[high_priority].try_pop(func))\r
+ {\r
+ if(func)\r
+ func();\r
+ } \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
{\r
std::function<void()> func;\r
- execution_queue_.pop(func); \r
- func();\r
+ execution_queue_[normal_priority].pop(func); \r
+\r
+ yield();\r
+\r
+ if(func)\r
+ func();\r
}\r
\r
void run() // noexcept\r
{\r
- win32_exception::install_handler();\r
+ win32_exception::install_handler(); \r
+ detail::SetThreadName(GetCurrentThreadId(), name_.c_str());\r
while(is_running_)\r
- execute();\r
- }\r
- \r
- std::function<void()> f_;\r
- boost::thread thread_;\r
- tbb::atomic<bool> is_running_;\r
- tbb::concurrent_bounded_queue<std::function<void()>> execution_queue_;\r
+ {\r
+ try\r
+ {\r
+ execute();\r
+ }\r
+ catch(...)\r
+ {\r
+ CASPAR_LOG_CURRENT_EXCEPTION();\r
+ }\r
+ }\r
+ } \r
};\r
\r
}
\ No newline at end of file