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

   1 // Copyright (C) 2013,2014 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 user scheduler.
   8 // 2013/11 Vicente J. Botet Escriba
   9 //    rename loop_executor.
  10
  11 #ifndef BOOST_THREAD_EXECUTORS_LOOP_EXECUTOR_HPP
  12 #define BOOST_THREAD_EXECUTORS_LOOP_EXECUTOR_HPP
  13
  14 #include <boost/thread/detail/config.hpp>
  15 #include <boost/thread/detail/delete.hpp>
  16 #include <boost/thread/detail/move.hpp>
  17 #include <boost/thread/concurrent_queues/sync_queue.hpp>
  18 #include <boost/thread/executors/work.hpp>
  19 #include <boost/assert.hpp>
  20
  21 #include <boost/config/abi_prefix.hpp>
  22
  23 namespace boost
  24 {
  25 namespace executors
  26 {
  27
  28   class loop_executor
  29   {
  30   public:
  31     /// type-erasure to store the works to do
  32     typedef  executors::work work;
  33   private:
  34     /// the thread safe work queue
  35     concurrent::sync_queue<work > work_queue;
  36
  37   public:
  38     /**
  39      * Effects: try to execute one task.
  40      * Returns: whether a task has been executed.
  41      * Throws: whatever the current task constructor throws or the task() throws.
  42      */
  43     bool try_executing_one()
  44     {
  45       return execute_one(/*wait:*/false);
  46     }
  47
  48   private:
  49     /**
  50      * Effects: Execute one task.
  51      * Remark: If wait is true, waits until a task is available or the executor
  52      *         is closed. If wait is false, returns false immediately if no
  53      *         task is available.
  54      * Returns: whether a task has been executed (if wait is true, only returns false if closed).
  55      * Throws: whatever the current task constructor throws or the task() throws.
  56      */
  57     bool execute_one(bool wait)
  58     {
  59       work task;
  60       try
  61       {
  62         queue_op_status status = wait ?
  63           work_queue.wait_pull(task) :
  64           work_queue.try_pull(task);
  65         if (status == queue_op_status::success)
  66         {
  67           task();
  68           return true;
  69         }
  70         BOOST_ASSERT(!wait || status == queue_op_status::closed);
  71         return false;
  72       }
  73       catch (...)
  74       {
  75         std::terminate();
  76         //return false;
  77       }
  78     }
  79
  80   public:
  81     /// loop_executor is not copyable.
  82     BOOST_THREAD_NO_COPYABLE(loop_executor)
  83
  84     /**
  85      * \b Effects: creates a thread pool that runs closures using one of its closure-executing methods.
  86      *
  87      * \b Throws: Whatever exception is thrown while initializing the needed resources.
  88      */
  89     loop_executor()
  90     {
  91     }
  92     /**
  93      * \b Effects: Destroys the thread pool.
  94      *
  95      * \b Synchronization: The completion of all the closures happen before the completion of the \c loop_executor destructor.
  96      */
  97     ~loop_executor()
  98     {
  99       // signal to all the worker thread that there will be no more submissions.
 100       close();
 101     }
 102
 103     /**
 104      * The main loop of the worker thread
 105      */
 106     void loop()
 107     {
 108       while (execute_one(/*wait:*/true))
 109       {
 110       }
 111       BOOST_ASSERT(closed());
 112       while (try_executing_one())
 113       {
 114       }
 115     }
 116
 117     /**
 118      * \b Effects: close the \c loop_executor for submissions.
 119      * The loop will work until there is no more closures to run.
 120      */
 121     void close()
 122     {
 123       work_queue.close();
 124     }
 125
 126     /**
 127      * \b Returns: whether the pool is closed for submissions.
 128      */
 129     bool closed()
 130     {
 131       return work_queue.closed();
 132     }
 133
 134     /**
 135      * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible.
 136      *
 137      * \b Effects: The specified \c closure will be scheduled for execution at some point in the future.
 138      * If invoked closure throws an exception the \c loop_executor will call \c std::terminate, as is the case with threads.
 139      *
 140      * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables.
 141      *
 142      * \b Throws: \c sync_queue_is_closed if the thread pool is closed.
 143      * Whatever exception that can be throw while storing the closure.
 144      */
 145     void submit(BOOST_THREAD_RV_REF(work) closure)  {
 146       work_queue.push(boost::move(closure));
 147     }
 148
 149 #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
 150     template <typename Closure>
 151     void submit(Closure & closure)
 152     {
 153       submit(work(closure));
 154    }
 155 #endif
 156
 157     void submit(void (*closure)())
 158     {
 159       submit(work(closure));
 160     }
 161
 162     template <typename Closure>
 163     void submit(BOOST_THREAD_FWD_REF(Closure) closure)
 164     {
 165       //work_queue.push(work(boost::forward<Closure>(closure)));
 166       work w((boost::forward<Closure>(closure)));
 167       submit(boost::move(w));
 168     }
 169
 170     /**
 171      * \b Requires: This must be called from an scheduled task.
 172      *
 173      * \b Effects: reschedule functions until pred()
 174      */
 175     template <typename Pred>
 176     bool reschedule_until(Pred const& pred)
 177     {
 178       do {
 179         if ( ! try_executing_one())
 180         {
 181           return false;
 182         }
 183       } while (! pred());
 184       return true;
 185     }
 186
 187     /**
 188      * run queued closures
 189      */
 190     void run_queued_closures()
 191     {
 192       sync_queue<work>::underlying_queue_type q = work_queue.underlying_queue();
 193       while (! q.empty())
 194       {
 195         work& task = q.front();
 196         task();
 197         q.pop_front();
 198       }
 199     }
 200
 201   };
 202 }
 203 using executors::loop_executor;
 204
 205 }
 206
 207 #include <boost/config/abi_suffix.hpp>
 208
 209 #endif