ceph/src/boost/boost/thread/executors/serial_executor.hpp

   1 // Copyright (C) 2013 Vicente J. Botet Escriba
   2 //
   3 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
   4 //  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   5 //
   6 // 2013/11 Vicente J. Botet Escriba
   7 //    first implementation of a simple serial scheduler.
   8
   9 #ifndef BOOST_THREAD_SERIAL_EXECUTOR_HPP
  10 #define BOOST_THREAD_SERIAL_EXECUTOR_HPP
  11
  12 #include <boost/thread/detail/config.hpp>
  13 #include <boost/thread/detail/delete.hpp>
  14 #include <boost/thread/detail/move.hpp>
  15 #include <boost/thread/concurrent_queues/sync_queue.hpp>
  16 #include <boost/thread/executors/work.hpp>
  17 #include <boost/thread/executors/generic_executor_ref.hpp>
  18 #include <boost/thread/future.hpp>
  19 #include <boost/thread/scoped_thread.hpp>
  20
  21 #include <boost/config/abi_prefix.hpp>
  22
  23 #if defined(BOOST_MSVC)
  24 # pragma warning(push)
  25 # pragma warning(disable: 4355) // 'this' : used in base member initializer list
  26 #endif
  27
  28 namespace boost
  29 {
  30 namespace executors
  31 {
  32   class serial_executor
  33   {
  34   public:
  35     /// type-erasure to store the works to do
  36     typedef  executors::work work;
  37   private:
  38     typedef  scoped_thread<> thread_t;
  39
  40     /// the thread safe work queue
  41     concurrent::sync_queue<work > work_queue;
  42     generic_executor_ref ex;
  43     thread_t thr;
  44
  45     struct try_executing_one_task {
  46       work& task;
  47       boost::promise<void> &p;
  48       try_executing_one_task(work& task, boost::promise<void> &p)
  49       : task(task), p(p) {}
  50       void operator()() {
  51         try {
  52           task();
  53           p.set_value();
  54         } catch (...)
  55         {
  56           p.set_exception(current_exception());
  57         }
  58       }
  59     };
  60   public:
  61     /**
  62      * \par Returns
  63      * The underlying executor wrapped on a generic executor reference.
  64      */
  65     generic_executor_ref& underlying_executor() BOOST_NOEXCEPT { return ex; }
  66
  67     /**
  68      * Effects: try to execute one task.
  69      * Returns: whether a task has been executed.
  70      * Throws: whatever the current task constructor throws or the task() throws.
  71      */
  72     bool try_executing_one()
  73     {
  74       work task;
  75       try
  76       {
  77         if (work_queue.try_pull(task) == queue_op_status::success)
  78         {
  79           boost::promise<void> p;
  80           try_executing_one_task tmp(task,p);
  81           ex.submit(tmp);
  82           p.get_future().wait();
  83           return true;
  84         }
  85         return false;
  86       }
  87       catch (...)
  88       {
  89         std::terminate();
  90         //return false;
  91       }
  92     }
  93   private:
  94     /**
  95      * Effects: schedule one task or yields
  96      * Throws: whatever the current task constructor throws or the task() throws.
  97      */
  98     void schedule_one_or_yield()
  99     {
 100         if ( ! try_executing_one())
 101         {
 102           this_thread::yield();
 103         }
 104     }
 105
 106     /**
 107      * The main loop of the worker thread
 108      */
 109     void worker_thread()
 110     {
 111       while (!closed())
 112       {
 113         schedule_one_or_yield();
 114       }
 115       while (try_executing_one())
 116       {
 117       }
 118     }
 119
 120   public:
 121     /// serial_executor is not copyable.
 122     BOOST_THREAD_NO_COPYABLE(serial_executor)
 123
 124     /**
 125      * \b Effects: creates a thread pool that runs closures using one of its closure-executing methods.
 126      *
 127      * \b Throws: Whatever exception is thrown while initializing the needed resources.
 128      */
 129     template <class Executor>
 130     serial_executor(Executor& ex)
 131     : ex(ex), thr(&serial_executor::worker_thread, this)
 132     {
 133     }
 134     /**
 135      * \b Effects: Destroys the thread pool.
 136      *
 137      * \b Synchronization: The completion of all the closures happen before the completion of the \c serial_executor destructor.
 138      */
 139     ~serial_executor()
 140     {
 141       // signal to the worker thread that there will be no more submissions.
 142       close();
 143     }
 144
 145     /**
 146      * \b Effects: close the \c serial_executor for submissions.
 147      * The loop will work until there is no more closures to run.
 148      */
 149     void close()
 150     {
 151       work_queue.close();
 152     }
 153
 154     /**
 155      * \b Returns: whether the pool is closed for submissions.
 156      */
 157     bool closed()
 158     {
 159       return work_queue.closed();
 160     }
 161
 162     /**
 163      * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible.
 164      *
 165      * \b Effects: The specified \c closure will be scheduled for execution at some point in the future.
 166      * If invoked closure throws an exception the \c serial_executor will call \c std::terminate, as is the case with threads.
 167      *
 168      * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables.
 169      *
 170      * \b Throws: \c sync_queue_is_closed if the thread pool is closed.
 171      * Whatever exception that can be throw while storing the closure.
 172      */
 173     void submit(BOOST_THREAD_RV_REF(work) closure)
 174     {
 175       work_queue.push(boost::move(closure));
 176     }
 177
 178 #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
 179     template <typename Closure>
 180     void submit(Closure & closure)
 181     {
 182       submit(work(closure));
 183     }
 184 #endif
 185     void submit(void (*closure)())
 186     {
 187       submit(work(closure));
 188     }
 189
 190     template <typename Closure>
 191     void submit(BOOST_THREAD_FWD_REF(Closure) closure)
 192     {
 193       work w((boost::forward<Closure>(closure)));
 194       submit(boost::move(w));
 195     }
 196
 197     /**
 198      * \b Requires: This must be called from an scheduled task.
 199      *
 200      * \b Effects: reschedule functions until pred()
 201      */
 202     template <typename Pred>
 203     bool reschedule_until(Pred const& pred)
 204     {
 205       do {
 206         if ( ! try_executing_one())
 207         {
 208           return false;
 209         }
 210       } while (! pred());
 211       return true;
 212     }
 213
 214   };
 215 }
 216 using executors::serial_executor;
 217 }
 218
 219 #if defined(BOOST_MSVC)
 220 # pragma warning(pop)
 221 #endif
 222
 223 #include <boost/config/abi_suffix.hpp>
 224
 225 #endif