ceph/src/boost/libs/thread/include/boost/thread/executors/basic_thread_pool.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/09 Vicente J. Botet Escriba
   7 //    Adapt to boost from CCIA C++11 implementation
   8 //    first implementation of a simple pool thread using a vector of threads and a sync_queue.
   9
  10 #ifndef BOOST_THREAD_EXECUTORS_BASIC_THREAD_POOL_HPP
  11 #define BOOST_THREAD_EXECUTORS_BASIC_THREAD_POOL_HPP
  12
  13 #include <boost/thread/detail/config.hpp>
  14 #include <boost/thread/detail/delete.hpp>
  15 #include <boost/thread/detail/move.hpp>
  16 #include <boost/thread/thread.hpp>
  17 #include <boost/thread/concurrent_queues/sync_queue.hpp>
  18 #include <boost/thread/executors/work.hpp>
  19 #include <boost/thread/csbl/vector.hpp>
  20
  21 #include <boost/config/abi_prefix.hpp>
  22
  23 namespace boost
  24 {
  25 namespace executors
  26 {
  27   class basic_thread_pool
  28   {
  29   public:
  30     /// type-erasure to store the works to do
  31     typedef  executors::work work;
  32   private:
  33     typedef thread thread_t;
  34     /// A move aware vector type
  35     typedef csbl::vector<thread_t> thread_vector;
  36
  37     /// A move aware vector
  38     thread_vector threads;
  39     /// the thread safe work queue
  40     concurrent::sync_queue<work > work_queue;
  41
  42   public:
  43     /**
  44      * Effects: try to execute one task.
  45      * Returns: whether a task has been executed.
  46      * Throws: whatever the current task constructor throws or the task() throws.
  47      */
  48     bool try_executing_one()
  49     {
  50       try
  51       {
  52         work task;
  53         if (work_queue.try_pull(task) == queue_op_status::success)
  54         {
  55           task();
  56           return true;
  57         }
  58         return false;
  59       }
  60       catch (...)
  61       {
  62         std::terminate();
  63         //return false;
  64       }
  65     }
  66     /**
  67      * Effects: schedule one task or yields
  68      * Throws: whatever the current task constructor throws or the task() throws.
  69      */
  70     void schedule_one_or_yield()
  71     {
  72         if ( ! try_executing_one())
  73         {
  74           this_thread::yield();
  75         }
  76     }
  77   private:
  78
  79     /**
  80      * The main loop of the worker threads
  81      */
  82     void worker_thread()
  83     {
  84       try
  85       {
  86         for(;;)
  87         {
  88           work task;
  89           queue_op_status st = work_queue.wait_pull(task);
  90           if (st == queue_op_status::closed) {
  91             return;
  92           }
  93           task();
  94         }
  95       }
  96       catch (...)
  97       {
  98         std::terminate();
  99         return;
 100       }
 101     }
 102 #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
 103     template <class AtThreadEntry>
 104     void worker_thread1(AtThreadEntry& at_thread_entry)
 105     {
 106       at_thread_entry(*this);
 107       worker_thread();
 108     }
 109 #endif
 110     void worker_thread2(void(*at_thread_entry)(basic_thread_pool&))
 111     {
 112       at_thread_entry(*this);
 113       worker_thread();
 114     }
 115     template <class AtThreadEntry>
 116     void worker_thread3(BOOST_THREAD_FWD_REF(AtThreadEntry) at_thread_entry)
 117     {
 118       at_thread_entry(*this);
 119       worker_thread();
 120     }
 121     static void do_nothing_at_thread_entry(basic_thread_pool&) {}
 122
 123   public:
 124     /// basic_thread_pool is not copyable.
 125     BOOST_THREAD_NO_COPYABLE(basic_thread_pool)
 126
 127     /**
 128      * \b Effects: creates a thread pool that runs closures on \c thread_count threads.
 129      *
 130      * \b Throws: Whatever exception is thrown while initializing the needed resources.
 131      */
 132     basic_thread_pool(unsigned const thread_count = thread::hardware_concurrency()+1)
 133     {
 134       try
 135       {
 136         threads.reserve(thread_count);
 137         for (unsigned i = 0; i < thread_count; ++i)
 138         {
 139 #if 1
 140           thread th (&basic_thread_pool::worker_thread, this);
 141           threads.push_back(thread_t(boost::move(th)));
 142 #else
 143           threads.push_back(thread_t(&basic_thread_pool::worker_thread, this)); // do not compile
 144 #endif
 145         }
 146       }
 147       catch (...)
 148       {
 149         close();
 150         throw;
 151       }
 152     }
 153     /**
 154      * \b Effects: creates a thread pool that runs closures on \c thread_count threads
 155      * and executes the at_thread_entry function at the entry of each created thread. .
 156      *
 157      * \b Throws: Whatever exception is thrown while initializing the needed resources.
 158      */
 159 #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
 160     template <class AtThreadEntry>
 161     basic_thread_pool( unsigned const thread_count, AtThreadEntry& at_thread_entry)
 162     {
 163       try
 164       {
 165         threads.reserve(thread_count);
 166         for (unsigned i = 0; i < thread_count; ++i)
 167         {
 168           thread th (&basic_thread_pool::worker_thread1<AtThreadEntry>, this, at_thread_entry);
 169           threads.push_back(thread_t(boost::move(th)));
 170           //threads.push_back(thread_t(&basic_thread_pool::worker_thread, this)); // do not compile
 171         }
 172       }
 173       catch (...)
 174       {
 175         close();
 176         throw;
 177       }
 178     }
 179 #endif
 180     basic_thread_pool( unsigned const thread_count, void(*at_thread_entry)(basic_thread_pool&))
 181     {
 182       try
 183       {
 184         threads.reserve(thread_count);
 185         for (unsigned i = 0; i < thread_count; ++i)
 186         {
 187           thread th (&basic_thread_pool::worker_thread2, this, at_thread_entry);
 188           threads.push_back(thread_t(boost::move(th)));
 189           //threads.push_back(thread_t(&basic_thread_pool::worker_thread, this)); // do not compile
 190         }
 191       }
 192       catch (...)
 193       {
 194         close();
 195         throw;
 196       }
 197     }
 198     template <class AtThreadEntry>
 199     basic_thread_pool( unsigned const thread_count, BOOST_THREAD_FWD_REF(AtThreadEntry) at_thread_entry)
 200     {
 201       try
 202       {
 203         threads.reserve(thread_count);
 204         for (unsigned i = 0; i < thread_count; ++i)
 205         {
 206           thread th (&basic_thread_pool::worker_thread3<AtThreadEntry>, this, boost::forward<AtThreadEntry>(at_thread_entry));
 207           threads.push_back(thread_t(boost::move(th)));
 208           //threads.push_back(thread_t(&basic_thread_pool::worker_thread, this)); // do not compile
 209         }
 210       }
 211       catch (...)
 212       {
 213         close();
 214         throw;
 215       }
 216     }
 217     /**
 218      * \b Effects: Destroys the thread pool.
 219      *
 220      * \b Synchronization: The completion of all the closures happen before the completion of the \c basic_thread_pool destructor.
 221      */
 222     ~basic_thread_pool()
 223     {
 224       // signal to all the worker threads that there will be no more submissions.
 225       close();
 226       // joins all the threads before destroying the thread pool resources (e.g. the queue).
 227       join();
 228     }
 229
 230     /**
 231      * \b Effects: join all the threads.
 232      */
 233     void join()
 234     {
 235       for (unsigned i = 0; i < threads.size(); ++i)
 236       {
 237         threads[i].join();
 238       }
 239     }
 240
 241     /**
 242      * \b Effects: close the \c basic_thread_pool for submissions.
 243      * The worker threads will work until there is no more closures to run.
 244      */
 245     void close()
 246     {
 247       work_queue.close();
 248     }
 249
 250     /**
 251      * \b Returns: whether the pool is closed for submissions.
 252      */
 253     bool closed()
 254     {
 255       return work_queue.closed();
 256     }
 257
 258     /**
 259      * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible.
 260      *
 261      * \b Effects: The specified \c closure will be scheduled for execution at some point in the future.
 262      * If invoked closure throws an exception the \c basic_thread_pool will call \c std::terminate, as is the case with threads.
 263      *
 264      * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables.
 265      *
 266      * \b Throws: \c sync_queue_is_closed if the thread pool is closed.
 267      * Whatever exception that can be throw while storing the closure.
 268      */
 269     void submit(BOOST_THREAD_RV_REF(work) closure)  {
 270       work_queue.push(boost::move(closure));
 271     }
 272
 273 #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
 274     template <typename Closure>
 275     void submit(Closure & closure)
 276     {
 277       submit(work(closure));
 278     }
 279 #endif
 280     void submit(void (*closure)())
 281     {
 282       submit(work(closure));
 283     }
 284
 285     template <typename Closure>
 286     void submit(BOOST_THREAD_FWD_REF(Closure) closure)
 287     {
 288       //submit(work(boost::forward<Closure>(closure)));
 289       work w((boost::forward<Closure>(closure)));
 290       submit(boost::move(w));
 291     }
 292
 293     /**
 294      * \b Requires: This must be called from an scheduled task.
 295      *
 296      * \b Effects: reschedule functions until pred()
 297      */
 298     template <typename Pred>
 299     bool reschedule_until(Pred const& pred)
 300     {
 301       do {
 302         if ( ! try_executing_one())
 303         {
 304           return false;
 305         }
 306       } while (! pred());
 307       return true;
 308     }
 309
 310   };
 311 }
 312 using executors::basic_thread_pool;
 313
 314 }
 315
 316 #include <boost/config/abi_suffix.hpp>
 317
 318 #endif