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

   1 #pragma once\r
   2 \r
   3 #include "../enum_class.h"\r
   4 \r
   5 #include <boost/thread/future.hpp>\r
   6 #include <boost/thread/thread.hpp>\r
   7 #include <boost/utility/declval.hpp>\r
   8 \r
   9 #include <functional>\r
  10 #include <memory>\r
  11 #include <tuple>\r
  12 \r
  13 namespace caspar {\r
  14 \r
  15 struct launch_policy_def\r
  16 {\r
  17         enum type\r
  18         {\r
  19                 async,\r
  20                 deferred\r
  21         };\r
  22 };\r
  23 typedef enum_class<launch_policy_def> launch;\r
  24 \r
  25 namespace detail {\r
  26 \r
  27 template<typename R, typename F>\r
  28 struct callback_object: public boost::detail::future_object<R>\r
  29 {       \r
  30         F f;\r
  31         bool done;\r
  32 \r
  33         template<typename F2>\r
  34         callback_object(F2&& f)\r
  35                 : f(std::forward<F2>(f))\r
  36                 , done(false)\r
  37         {\r
  38         }\r
  39                 \r
  40         void operator()()\r
  41         {               \r
  42                 boost::lock_guard<boost::mutex> lock2(mutex);\r
  43 \r
  44                 if(done)\r
  45                         return;\r
  46 \r
  47         try\r
  48         {\r
  49                    this->mark_finished_with_result_internal(f());\r
  50         }\r
  51         catch(...)\r
  52         {\r
  53                         this->mark_exceptional_finish_internal(boost::current_exception());\r
  54         }\r
  55 \r
  56                 done = true;\r
  57         }\r
  58 };\r
  59 \r
  60 template<typename F>\r
  61 struct callback_object<void, F> : public boost::detail::future_object<void>\r
  62 {       \r
  63         F f;\r
  64         bool done;\r
  65 \r
  66         template<typename F2>\r
  67         callback_object(F2&& f)\r
  68                 : f(std::forward<F2>(f))\r
  69                 , done(false)\r
  70         {\r
  71         }\r
  72 \r
  73         void operator()()\r
  74         {\r
  75                 boost::lock_guard<boost::mutex> lock2(mutex);\r
  76                 \r
  77                 if(done)\r
  78                         return;\r
  79 \r
  80         try\r
  81         {\r
  82                         f();\r
  83                         this->mark_finished_with_result_internal();\r
  84                 }\r
  85         catch(...)\r
  86         {\r
  87                         this->mark_exceptional_finish_internal(boost::current_exception());\r
  88         }\r
  89 \r
  90                 done = true;\r
  91         }\r
  92 };\r
  93 \r
  94 }\r
  95         \r
  96 template<typename F>\r
  97 auto async(launch lp, F&& f) -> boost::unique_future<decltype(f())>\r
  98 {               \r
  99         typedef decltype(f()) result_type;\r
 100         \r
 101         if(lp == launch::deferred)\r
 102         {                       \r
 103                 // HACK: THIS IS A MAYOR HACK!\r
 104 \r
 105                 typedef boost::detail::future_object<result_type> future_object_t;\r
 106                                         \r
 107                 auto callback_object     = boost::make_shared<detail::callback_object<result_type, F>>(std::forward<F>(f));\r
 108                 auto callback_object_raw = callback_object.get();\r
 109                 auto future_object               = boost::static_pointer_cast<future_object_t>(std::move(callback_object));\r
 110 \r
 111                 int dummy = 0;\r
 112                 future_object->set_wait_callback(std::function<void(int)>([callback_object_raw](int) mutable\r
 113                 {                                                               \r
 114                         (*callback_object_raw)();\r
 115                 }), &dummy);\r
 116                 \r
 117                 boost::unique_future<result_type> future;\r
 118                 reinterpret_cast<boost::shared_ptr<future_object_t>&>(future) = std::move(future_object); // Get around the "private" encapsulation.\r
 119                 return std::move(future);\r
 120         }\r
 121         else\r
 122         {\r
 123                 typedef boost::packaged_task<result_type> task_t;\r
 124 \r
 125                 auto task       = task_t(std::forward<F>(f));   \r
 126                 auto future = task.get_future();\r
 127                 \r
 128                 boost::thread(std::move(task)).detach();\r
 129         \r
 130                 return std::move(future);\r
 131         }\r
 132 }\r
 133         \r
 134 template<typename F>\r
 135 auto async(F&& f) -> boost::unique_future<decltype(f())>\r
 136 {       \r
 137         return async(launch::async, std::forward<F>(f));\r
 138 }\r
 139 \r
 140 template<typename T>\r
 141 auto make_shared(boost::unique_future<T>&& f) -> boost::shared_future<T>\r
 142 {       \r
 143         return boost::shared_future<T>(std::move(f));\r
 144 }\r
 145 \r
 146 template<typename T>\r
 147 auto flatten(boost::unique_future<T>&& f) -> boost::unique_future<decltype(f.get().get())>\r
 148 {\r
 149         auto shared_f = make_shared(std::move(f));\r
 150         return async(launch::deferred, [=]() mutable\r
 151         {\r
 152                 return shared_f.get().get();\r
 153         });\r
 154 }\r
 155 \r
 156 }