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 ]
;
}