-#pragma once\r
+/*\r
+* Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>\r
+*\r
+* This file is part of CasparCG (www.casparcg.com).\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
+* Author: Robert Nagy, ronag89@gmail.com\r
+*/\r
\r
-#include "../exception/exceptions.h"\r
-#include "../exception/win32_exception.h"\r
-#include "../log/log.h"\r
+#pragma once\r
\r
-#include <boost/thread.hpp>\r
+#include "../except.h"\r
+#include "../enum_class.h"\r
+#include "../log.h"\r
\r
#include <tbb/atomic.h>\r
#include <tbb/concurrent_queue.h>\r
\r
+#include <boost/thread.hpp>\r
+\r
#include <functional>\r
-#include <array>\r
\r
namespace caspar {\r
\r
-class executor\r
+namespace detail {\r
+ \r
+template<typename T>\r
+struct move_on_copy\r
{\r
-public:\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 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
+ enum type\r
{\r
- low_priority = 0,\r
+ high_priority,\r
normal_priority,\r
- high_priority\r
+ priority_count\r
};\r
+};\r
+typedef enum_class<task_priority_def> task_priority;\r
\r
- explicit executor(const std::function<void()>& f = nullptr)\r
- {\r
- size_ = 0;\r
- is_running_ = false;\r
- f_ = f != nullptr ? f : [this]{run();};\r
+class executor\r
+{\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_[task_priority::priority_count];\r
+ \r
+ template<typename Func>\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
+ \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
+ try\r
+ {\r
+ if(boost::this_thread::get_id() == thread_.get_id()) // Avoids potential deadlock.\r
+ my_task();\r
+ }\r
+ catch(boost::task_already_started&){}\r
+ }));\r
+ \r
+ return std::move(task);\r
}\r
\r
- virtual ~executor()\r
+public: \r
+ executor(const std::wstring& name) // noexcept\r
{\r
- stop();\r
+ name; // TODO: Use to set thread name.\r
+ is_running_ = true;\r
+ thread_ = boost::thread([this]{run();});\r
}\r
-\r
- void start() // noexcept\r
+ \r
+ virtual ~executor() // noexcept\r
{\r
- if(is_running_.fetch_and_store(true))\r
- return;\r
- thread_ = boost::thread(f_);\r
+ stop();\r
+ join();\r
}\r
\r
- bool is_running() const // noexcept\r
+ void set_capacity(size_t capacity) // noexcept\r
{\r
- return is_running_;\r
+ execution_queue_[task_priority::normal_priority].set_capacity(capacity);\r
}\r
\r
- void stop(bool wait = true) // noexcept\r
+ void clear()\r
+ { \r
+ std::function<void()> 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
- begin_invoke([]{}); // wake if sleeping\r
- if(wait && boost::this_thread::get_id() != thread_.get_id())\r
- thread_.join();\r
+ execution_queue_[task_priority::normal_priority].try_push([]{}); // Wake the execution thread.\r
}\r
\r
- void execute() // noexcept\r
+ void wait() // noexcept\r
{\r
- boost::unique_lock<boost::mutex> lock(mut_);\r
- while(size_ < 1) \r
- cond_.wait(lock);\r
- \r
- try_execute();\r
+ invoke([]{});\r
}\r
\r
- bool try_execute() // noexcept\r
+ void join()\r
{\r
- std::function<void()> func;\r
- if(execution_queue_[high_priority].try_pop(func) || execution_queue_[normal_priority].try_pop(func) || execution_queue_[low_priority].try_pop(func))\r
- {\r
- func();\r
- --size_;\r
- }\r
+ if(boost::this_thread::get_id() == thread_.get_id())\r
+ BOOST_THROW_EXCEPTION(invalid_operation());\r
\r
- return func != nullptr;\r
+ thread_.join();\r
}\r
- \r
+ \r
template<typename Func>\r
- auto begin_invoke(Func&& func, priority p = 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
- typedef decltype(func()) result_type; \r
- \r
- auto task = std::make_shared<boost::packaged_task<result_type>>(std::forward<Func>(func)); // boost::packaged_task cannot be moved into lambda, need to used shared_ptr.\r
- auto future = task->get_future();\r
- \r
- task->set_wait_callback(std::function<void(decltype(*task)& task)>([=](decltype(*task)& task) // The std::function wrapper is required in order to add ::result_type to functor class.\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 = detail::make_move_on_copy(create_task(func));\r
+\r
+ auto future = task_adaptor.value.get_future();\r
+\r
+ execution_queue_[priority.value()].push([=]\r
{\r
try\r
{\r
- if(boost::this_thread::get_id() == thread_.get_id()) // Avoids potential deadlock.\r
- task();\r
+ task_adaptor.value();\r
+ }\r
+ catch(boost::task_already_started&)\r
+ {\r
+ }\r
+ catch(...)\r
+ {\r
+ CASPAR_LOG_CURRENT_EXCEPTION();\r
}\r
- catch(boost::task_already_started&){}\r
- }));\r
- execution_queue_[p].push([=]\r
- {\r
- try{(*task)();}\r
- catch(boost::task_already_started&){}\r
- catch(...){CASPAR_LOG_CURRENT_EXCEPTION();}\r
});\r
- ++size_;\r
- cond_.notify_one();\r
\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
- size_t size() const\r
- {\r
- return execution_queue_.size();\r
- }\r
-\r
- bool empty() const\r
- {\r
- return execution_queue_.empty();\r
- }\r
\r
template<typename Func>\r
- auto invoke(Func&& func, priority p = normal_priority) -> decltype(func())\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), p).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())\r
+ BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("Executor can only yield inside of thread context."));\r
+\r
+ std::function<void()> func;\r
+ execution_queue_[task_priority::normal_priority].pop(func); \r
\r
-private:\r
+ std::function<void()> func2;\r
+ while(execution_queue_[task_priority::high_priority].try_pop(func2))\r
+ {\r
+ if(func2)\r
+ func2();\r
+ } \r
\r
- virtual void run() // noexcept\r
+ if(func)\r
+ func();\r
+ }\r
+ \r
+ function_queue::size_type size() const /*noexcept*/\r
{\r
- win32_exception::install_handler();\r
- while(is_running_)\r
- execute();\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
- tbb::atomic<size_t> size_;\r
- boost::condition_variable cond_;\r
- boost::mutex mut_;\r
-\r
- std::function<void()> f_;\r
- boost::thread thread_;\r
- tbb::atomic<bool> is_running_;\r
- std::array<tbb::concurrent_bounded_queue<std::function<void()>>, 3> execution_queue_;\r
+ void run() // noexcept\r
+ {\r
+ win32_exception::install_handler(); \r
+ while(is_running_)\r
+ {\r
+ try\r
+ {\r
+ yield();\r
+ }\r
+ catch(...)\r
+ {\r
+ CASPAR_LOG_CURRENT_EXCEPTION();\r
+ }\r
+ }\r
+ } \r
};\r
\r
}
\ No newline at end of file
projects / casparcg / blobdiff