Boost logo

Boost-Commit :

Subject: [Boost-commit] svn:boost r86570 - in trunk: boost/thread libs/thread/example libs/thread/test
From: vicente.botet_at_[hidden]
Date: 2013-11-06 01:37:47


Author: viboes
Date: 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013)
New Revision: 86570
URL: http://svn.boost.org/trac/boost/changeset/86570

Log:
Thread: Added user_scheduler (loop_executor).

Added:
   trunk/boost/thread/user_scheduler.hpp (contents, props changed)
   trunk/libs/thread/example/user_scheduler.cpp (contents, props changed)
Text files modified:
   trunk/boost/thread/executor.hpp | 2
   trunk/boost/thread/sync_queue.hpp | 7 +
   trunk/boost/thread/thread_pool.hpp | 15 --
   trunk/boost/thread/user_scheduler.hpp | 202 ++++++++++++++++++++++++++++++++++++++++
   trunk/libs/thread/example/executor.cpp | 110 +++------------------
   trunk/libs/thread/example/thread_pool.cpp | 104 ++------------------
   trunk/libs/thread/example/user_scheduler.cpp | 70 +++++++++++++
   trunk/libs/thread/test/Jamfile.v2 | 2
   8 files changed, 311 insertions(+), 201 deletions(-)

Modified: trunk/boost/thread/executor.hpp
==============================================================================
--- trunk/boost/thread/executor.hpp Tue Nov 5 18:08:13 2013 (r86569)
+++ trunk/boost/thread/executor.hpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -144,6 +144,8 @@
     template <typename A1>
     executor_adaptor(BOOST_THREAD_FWD_REF(A1) a1) : ex(boost::forward<A1>(a1)) {}
 #endif
+ Executor& underlying_executor() { return ex; }
+
     /**
      * \b Effects: close the \c executor for submissions.
      * The worker threads will work until there is no more closures to run.

Modified: trunk/boost/thread/sync_queue.hpp
==============================================================================
--- trunk/boost/thread/sync_queue.hpp Tue Nov 5 18:08:13 2013 (r86569)
+++ trunk/boost/thread/sync_queue.hpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -33,6 +33,7 @@
   {
   public:
     typedef ValueType value_type;
+ typedef csbl::deque<ValueType> underlying_queue_type;
     typedef std::size_t size_type;
 
     // Constructors/Assignment/Destructors
@@ -68,12 +69,16 @@
     inline bool try_pull(value_type&);
     inline bool try_pull(no_block_tag,value_type&);
     inline shared_ptr<ValueType> try_pull();
+ inline underlying_queue_type underlying_queue() {
+ lock_guard<mutex> lk(mtx_);
+ return boost::move(data_);
+ }
 
   private:
     mutable mutex mtx_;
     condition_variable not_empty_;
     size_type waiting_empty_;
- csbl::deque<ValueType> data_;
+ underlying_queue_type data_;
     bool closed_;
 
     inline bool empty(unique_lock<mutex>& ) const BOOST_NOEXCEPT

Modified: trunk/boost/thread/thread_pool.hpp
==============================================================================
--- trunk/boost/thread/thread_pool.hpp Tue Nov 5 18:08:13 2013 (r86569)
+++ trunk/boost/thread/thread_pool.hpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -16,28 +16,13 @@
 #include <boost/thread/scoped_thread.hpp>
 #include <boost/thread/sync_queue.hpp>
 #include <boost/thread/detail/work.hpp>
-
-
-//#ifdef BOOST_NO_CXX11_HDR_FUNCTIONAL
-//#include <boost/function.hpp>
-//#else
-//#include <functional>
-//#endif
-
 #include <boost/thread/csbl/vector.hpp>
 
-//#if defined BOOST_NO_CXX11_RVALUE_REFERENCES
-//#include <boost/container/vector.hpp>
-//#else
-//#include <vector>
-//#endif
-
 #include <boost/config/abi_prefix.hpp>
 
 namespace boost
 {
 
-
   class thread_pool
   {
     /// type-erasure to store the works to do

Added: trunk/boost/thread/user_scheduler.hpp
==============================================================================
--- /dev/null 00:00:00 1970 (empty, because file is newly added)
+++ trunk/boost/thread/user_scheduler.hpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -0,0 +1,202 @@
+// Copyright (C) 2013 Vicente J. Botet Escriba
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+// 2013/11 Vicente J. Botet Escriba
+// first implementation of a simple serial scheduler.
+
+#ifndef BOOST_THREAD_USER_SCHEDULER_HPP
+#define BOOST_THREAD_USER_SCHEDULER_HPP
+
+#include <boost/thread/detail/config.hpp>
+#include <boost/thread/detail/delete.hpp>
+#include <boost/thread/detail/move.hpp>
+#include <boost/thread/sync_queue.hpp>
+#include <boost/thread/detail/work.hpp>
+
+#include <boost/config/abi_prefix.hpp>
+
+namespace boost
+{
+
+ class user_scheduler
+ {
+ /// type-erasure to store the works to do
+ typedef thread_detail::work work;
+
+ /// the thread safe work queue
+ sync_queue<work > work_queue;
+
+ public:
+ /**
+ * Effects: try to execute one task.
+ * Returns: whether a task has been executed.
+ * Throws: whatever the current task constructor throws or the task() throws.
+ */
+ bool try_executing_one()
+ {
+ work task;
+ try
+ {
+ if (work_queue.try_pull(task))
+ {
+ task();
+ return true;
+ }
+ return false;
+ }
+ catch (std::exception& )
+ {
+ return false;
+ }
+ catch (...)
+ {
+ return false;
+ }
+ }
+ private:
+ /**
+ * Effects: schedule one task or yields
+ * Throws: whatever the current task constructor throws or the task() throws.
+ */
+ void schedule_one_or_yield()
+ {
+ if ( ! try_executing_one())
+ {
+ this_thread::yield();
+ }
+ }
+
+
+ /**
+ * The main loop of the worker thread
+ */
+ void worker_thread()
+ {
+ while (!closed())
+ {
+ schedule_one_or_yield();
+ }
+ while (try_executing_one())
+ {
+ }
+ }
+
+ public:
+ /// user_scheduler is not copyable.
+ BOOST_THREAD_NO_COPYABLE(user_scheduler)
+
+ /**
+ * \b Effects: creates a thread pool that runs closures using one of its closure-executing methods.
+ *
+ * \b Throws: Whatever exception is thrown while initializing the needed resources.
+ */
+ user_scheduler()
+ {
+ }
+ /**
+ * \b Effects: Destroys the thread pool.
+ *
+ * \b Synchronization: The completion of all the closures happen before the completion of the \c user_scheduler destructor.
+ */
+ ~user_scheduler()
+ {
+ // signal to all the worker thread that there will be no more submissions.
+ close();
+ }
+
+ /**
+ * loop
+ */
+ void loop() { worker_thread(); }
+ /**
+ * \b Effects: close the \c user_scheduler for submissions.
+ * The loop will work until there is no more closures to run.
+ */
+ void close()
+ {
+ work_queue.close();
+ }
+
+ /**
+ * \b Returns: whether the pool is closed for submissions.
+ */
+ bool closed()
+ {
+ return work_queue.closed();
+ }
+
+ /**
+ * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible.
+ *
+ * \b Effects: The specified \c closure will be scheduled for execution at some point in the future.
+ * If invoked closure throws an exception the \c user_scheduler will call \c std::terminate, as is the case with threads.
+ *
+ * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables.
+ *
+ * \b Throws: \c sync_queue_is_closed if the thread pool is closed.
+ * Whatever exception that can be throw while storing the closure.
+ */
+
+#if defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
+ template <typename Closure>
+ void submit(Closure & closure)
+ {
+ work w ((closure));
+ work_queue.push(boost::move(w));
+ //work_queue.push(work(closure)); // todo check why this doesn't work
+ }
+#endif
+ void submit(void (*closure)())
+ {
+ work w ((closure));
+ work_queue.push(boost::move(w));
+ //work_queue.push(work(closure)); // todo check why this doesn't work
+ }
+
+ template <typename Closure>
+ void submit(BOOST_THREAD_RV_REF(Closure) closure)
+ {
+ work w =boost::move(closure);
+ work_queue.push(boost::move(w));
+ //work_queue.push(work(boost::move(closure))); // todo check why this doesn't work
+ }
+
+ /**
+ * \b Requires: This must be called from an scheduled task.
+ *
+ * \b Effects: reschedule functions until pred()
+ */
+ template <typename Pred>
+ bool reschedule_until(Pred const& pred)
+ {
+ do {
+ if ( ! try_executing_one())
+ {
+ return false;
+ }
+ } while (! pred());
+ return true;
+ }
+ /**
+ * run queued closures
+ */
+ void run_queued_closures()
+ {
+ sync_queue<work>::underlying_queue_type q = work_queue.underlying_queue();
+ while (q.empty())
+ {
+ work task = q.front();
+ q.pop_front();
+ task();
+ }
+ }
+
+ };
+
+}
+
+#include <boost/config/abi_suffix.hpp>
+
+#endif

Modified: trunk/libs/thread/example/executor.cpp
==============================================================================
--- trunk/libs/thread/example/executor.cpp Tue Nov 5 18:08:13 2013 (r86569)
+++ trunk/libs/thread/example/executor.cpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -9,6 +9,7 @@
 
 #include <boost/thread/detail/log.hpp>
 #include <boost/thread/thread_pool.hpp>
+#include <boost/thread/user_scheduler.hpp>
 #include <boost/thread/executor.hpp>
 #include <boost/assert.hpp>
 #include <string>
@@ -25,109 +26,33 @@
     << boost::this_thread::get_id() << " P2" << BOOST_THREAD_END_LOG;
 }
 
-void push(boost::csbl::deque<boost::thread_detail::work> &data_, BOOST_THREAD_RV_REF(boost::thread_detail::work) closure)
-{
- try
- {
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- boost::thread_detail::work v;
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- //v = boost::move(closure);
- //v = boost::forward<boost::thread_detail::work>(closure);
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-
- data_.push_back(boost::move(closure));
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-
- //data_.push_back(boost::forward<boost::thread_detail::work>(closure));
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-
- }
- catch (std::exception& ex)
- {
- BOOST_THREAD_LOG
- << "ERRORRRRR " << ex.what() << "" << BOOST_THREAD_END_LOG;
- }
- catch (...)
- {
- BOOST_THREAD_LOG
- << " ERRORRRRR exception thrown" << BOOST_THREAD_END_LOG;
- }
-}
-
-template <typename Closure>
-void submit(boost::csbl::deque<boost::thread_detail::work> &data_, BOOST_THREAD_FWD_REF(Closure) closure)
-{
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- //work w =boost::move(closure);
- //work_queue.push(boost::move(w));
- //push(data_, boost::thread_detail::work(boost::forward<Closure>(closure)));
- boost::thread_detail::work v =boost::forward<Closure>(closure);
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- push(data_, boost::move(v));
-
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
+void submit_some(boost::executor& tp) {
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
 }
 
 int main()
 {
   BOOST_THREAD_LOG
     << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-#if 0
   {
     try
     {
- boost::thread_detail::work f(&p1);
-
- boost::csbl::deque<boost::thread_detail::work> data_;
- data_.push_back(boost::move(f));
- data_.push_back(boost::thread_detail::work(&p1));
- submit(data_, &p1);
- }
- catch (std::exception& ex)
- {
- BOOST_THREAD_LOG
- << "ERRORRRRR " << ex.what() << "" << BOOST_THREAD_END_LOG;
- }
- catch (...)
- {
- BOOST_THREAD_LOG
- << " ERRORRRRR exception thrown" << BOOST_THREAD_END_LOG;
- }
+ boost::executor_adaptor<boost::thread_pool> ea;
+ submit_some(ea);
 
- typedef boost::csbl::vector<boost::thread> thread_vector;
- thread_vector threads;
+ boost::executor_adaptor<boost::user_scheduler> ea2;
+ submit_some(ea2);
+ ea2.underlying_executor().run_queued_closures();
 
- }
-#endif
-#if 1
- {
- try
- {
- boost::executor_adaptor<boost::thread_pool> ea;
- boost::executor &tp=ea;
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- tp.submit(&p1);
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
     }
     catch (std::exception& ex)
     {
@@ -142,7 +67,6 @@
       return 2;
     }
   }
-#endif
   BOOST_THREAD_LOG
     << boost::this_thread::get_id() << "MAIN>" << BOOST_THREAD_END_LOG;
   return 0;

Modified: trunk/libs/thread/example/thread_pool.cpp
==============================================================================
--- trunk/libs/thread/example/thread_pool.cpp Tue Nov 5 18:08:13 2013 (r86569)
+++ trunk/libs/thread/example/thread_pool.cpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -24,108 +24,29 @@
     << boost::this_thread::get_id() << " P2" << BOOST_THREAD_END_LOG;
 }
 
-void push(boost::csbl::deque<boost::thread_detail::work> &data_, BOOST_THREAD_RV_REF(boost::thread_detail::work) closure)
-{
- try
- {
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- boost::thread_detail::work v;
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- //v = boost::move(closure);
- //v = boost::forward<boost::thread_detail::work>(closure);
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-
- data_.push_back(boost::move(closure));
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-
- //data_.push_back(boost::forward<boost::thread_detail::work>(closure));
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-
- }
- catch (std::exception& ex)
- {
- BOOST_THREAD_LOG
- << "ERRORRRRR " << ex.what() << "" << BOOST_THREAD_END_LOG;
- }
- catch (...)
- {
- BOOST_THREAD_LOG
- << " ERRORRRRR exception thrown" << BOOST_THREAD_END_LOG;
- }
+void submit_some(boost::thread_pool& tp) {
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
 }
 
-template <typename Closure>
-void submit(boost::csbl::deque<boost::thread_detail::work> &data_, BOOST_THREAD_FWD_REF(Closure) closure)
-{
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- //work w =boost::move(closure);
- //work_queue.push(boost::move(w));
- //push(data_, boost::thread_detail::work(boost::forward<Closure>(closure)));
- boost::thread_detail::work v =boost::forward<Closure>(closure);
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- push(data_, boost::move(v));
-
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-}
 
 int main()
 {
   BOOST_THREAD_LOG
     << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
-#if 0
- {
- try
- {
- boost::thread_detail::work f(&p1);
-
- boost::csbl::deque<boost::thread_detail::work> data_;
- data_.push_back(boost::move(f));
- data_.push_back(boost::thread_detail::work(&p1));
- submit(data_, &p1);
- }
- catch (std::exception& ex)
- {
- BOOST_THREAD_LOG
- << "ERRORRRRR " << ex.what() << "" << BOOST_THREAD_END_LOG;
- }
- catch (...)
- {
- BOOST_THREAD_LOG
- << " ERRORRRRR exception thrown" << BOOST_THREAD_END_LOG;
- }
-
- typedef boost::csbl::vector<boost::thread> thread_vector;
- thread_vector threads;
-
- }
-#endif
-#if 1
   {
     try
     {
       boost::thread_pool tp;
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- tp.submit(&p1);
- BOOST_THREAD_LOG
- << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
- tp.submit(&p1);
- tp.submit(&p2);
+ submit_some(tp);
     }
     catch (std::exception& ex)
     {
@@ -140,7 +61,6 @@
       return 2;
     }
   }
-#endif
   BOOST_THREAD_LOG
     << boost::this_thread::get_id() << "MAIN>" << BOOST_THREAD_END_LOG;
   return 0;

Added: trunk/libs/thread/example/user_scheduler.cpp
==============================================================================
--- /dev/null 00:00:00 1970 (empty, because file is newly added)
+++ trunk/libs/thread/example/user_scheduler.cpp 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -0,0 +1,70 @@
+// Copyright (C) 2013 Vicente Botet
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+
+#define BOOST_THREAD_VERSION 4
+#define BOOST_THREAD_USES_LOG
+#define BOOST_THREAD_USES_LOG_THREAD_ID
+
+#include <boost/thread/detail/log.hpp>
+#include <boost/thread/user_scheduler.hpp>
+#include <boost/assert.hpp>
+#include <string>
+
+void p1()
+{
+ BOOST_THREAD_LOG
+ << boost::this_thread::get_id() << " P1" << BOOST_THREAD_END_LOG;
+}
+
+void p2()
+{
+ BOOST_THREAD_LOG
+ << boost::this_thread::get_id() << " P2" << BOOST_THREAD_END_LOG;
+}
+
+void submit_some(boost::user_scheduler& tp) {
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+ tp.submit(&p1);
+ tp.submit(&p2);
+}
+
+int main()
+{
+ BOOST_THREAD_LOG
+ << boost::this_thread::get_id() << " <MAIN" << BOOST_THREAD_END_LOG;
+ {
+ try
+ {
+ boost::user_scheduler tp;
+ submit_some(tp);
+ tp.run_queued_closures();
+ submit_some(tp);
+ tp.run_queued_closures();
+ }
+ catch (std::exception& ex)
+ {
+ BOOST_THREAD_LOG
+ << "ERRORRRRR " << ex.what() << "" << BOOST_THREAD_END_LOG;
+ return 1;
+ }
+ catch (...)
+ {
+ BOOST_THREAD_LOG
+ << " ERRORRRRR exception thrown" << BOOST_THREAD_END_LOG;
+ return 2;
+ }
+ }
+
+ BOOST_THREAD_LOG
+ << boost::this_thread::get_id() << "MAIN>" << BOOST_THREAD_END_LOG;
+ return 0;
+}

Modified: trunk/libs/thread/test/Jamfile.v2
==============================================================================
--- trunk/libs/thread/test/Jamfile.v2 Tue Nov 5 18:08:13 2013 (r86569)
+++ trunk/libs/thread/test/Jamfile.v2 2013-11-06 01:37:46 EST (Wed, 06 Nov 2013) (r86570)
@@ -721,6 +721,7 @@
           [ thread-run2 ../example/lambda_future.cpp : ex_lambda_future ]
           [ thread-run2 ../example/not_interleaved2.cpp : ex_not_interleaved2 ]
           [ thread-run2 ../example/thread_pool.cpp : ex_thread_pool ]
+ [ thread-run2 ../example/user_scheduler.cpp : ex_user_scheduler ]
           [ thread-run2 ../example/executor.cpp : ex_executor ]
           [ thread-run2 ../example/future_when_all.cpp : future_when_all ]
 
@@ -813,6 +814,7 @@
           #[ thread-run ../example/std_async_test.cpp ]
           #[ compile virtual_noexcept.cpp ]
           #[ thread-run clang_main.cpp ]
+ #[ thread-run test_9303.cpp ]
     ;
 
 }


Boost-Commit list run by bdawes at acm.org, david.abrahams at rcn.com, gregod at cs.rpi.edu, cpdaniel at pacbell.net, john at johnmaddock.co.uk