Remove counting when async future blocking.

merge from develop.
then can be now sync. The destructor of a shared state that joins check for thread_ifbefore join. Adde continuation_shared_state to share common behavior.
2026-07-23 13:53:42 +00:00 · 2015-10-17 11:38:06 +02:00 · 2015-10-17 10:51:50 +02:00 · 2015-10-17 00:44:45 +02:00 · 2015-10-17 00:40:53 +02:00 · 2015-10-17 00:40:23 +02:00
7 changed files with 402 additions and 483 deletions
@@ -7,6 +7,7 @@
 #if ! defined  BOOST_NO_CXX11_DECLTYPE
 #define BOOST_RESULT_OF_USE_DECLTYPE
 #endif
+#include <iostream>

 #define BOOST_THREAD_VERSION 4
 #define BOOST_THREAD_PROVIDES_EXECUTORS
@@ -48,8 +49,9 @@ void p2()

 int f1()
 {
-  // std::cout << BOOST_CONTEXTOF << std::endl;
+  std::cout << BOOST_CONTEXTOF << std::endl;
  boost::this_thread::sleep_for(boost::chrono::seconds(1));
+  std::cout << BOOST_CONTEXTOF << std::endl;
  return 1;
 }
 int f2(int i)
@@ -78,61 +80,95 @@ void at_th_entry(boost::basic_thread_pool& )

 int test_executor_adaptor()
 {
-  // std::cout << BOOST_CONTEXTOF << std::endl;
+  std::cout << BOOST_CONTEXTOF << std::endl;
  {
    try
    {
      {
        boost::executor_adaptor < boost::basic_thread_pool > ea(4);
+        std::cout << BOOST_CONTEXTOF << std::endl;
        submit_some( ea);
+        std::cout << BOOST_CONTEXTOF << std::endl;
+#if 1
+      // fixme
+      // ERROR= tr1::bad_weak_ptr
        {
          boost::future<int> t1 = boost::async(ea, &f1);
          boost::future<int> t2 = boost::async(ea, &f1);
-          // std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
-          // std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
        }
+        std::cout << BOOST_CONTEXTOF << std::endl;
        submit_some(ea);
+        std::cout << BOOST_CONTEXTOF << std::endl;
        {
          boost::basic_thread_pool ea3(1);
+          std::cout << BOOST_CONTEXTOF << std::endl;
          boost::future<int> t1 = boost::async(ea3, &f1);
+          std::cout << BOOST_CONTEXTOF << std::endl;
          boost::future<int> t2 = boost::async(ea3, &f1);
+          std::cout << BOOST_CONTEXTOF << std::endl;
          //boost::future<int> t2 = boost::async(ea3, f2, 1); // todo this doesn't compiles yet on C++11
          //boost::future<int> t2 = boost::async(ea3, boost::bind(f2, 1)); // todo this doesn't compiles yet on C++98
-          // std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
-          // std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
        }
+#endif
+        std::cout << BOOST_CONTEXTOF << std::endl;
        submit_some(ea);
+        std::cout << BOOST_CONTEXTOF << std::endl;
      }
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::executor_adaptor < boost::loop_executor > ea2;
        submit_some( ea2);
        ea2.underlying_executor().run_queued_closures();
      }
 #if ! defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::executor_adaptor < boost::basic_thread_pool > ea1(4);
        boost::executor_adaptor < boost::serial_executor > ea2(ea1);
        submit_some(ea2);
      }
 #endif
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::executor_adaptor < boost::inline_executor > ea1;
        submit_some(ea1);
      }
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::executor_adaptor < boost::thread_executor > ea1;
        submit_some(ea1);
      }
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
+#if 1
+      // fixme
+      // ERROR= tr1::bad_weak_ptr
      {
        boost::basic_thread_pool  ea(4, at_th_entry);
        boost::future<int> t1 = boost::async(ea, &f1);
-        // std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
+        std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
      }
+#endif
+      std::cout << BOOST_CONTEXTOF << std::endl;
+      {
+        boost::async(&f1);
+      }
+#if 1
+      // fixme
+      // ERROR= tr1::bad_weak_ptr
+
+      std::cout << BOOST_CONTEXTOF << std::endl;
+      {
+        boost::basic_thread_pool  ea(1);
+        boost::async(ea,&f1);
+      }
+#endif
+      std::cout << BOOST_CONTEXTOF << std::endl;
+      boost::this_thread::sleep_for(boost::chrono::milliseconds(200));
+      std::cout << BOOST_CONTEXTOF << std::endl;
    }
    catch (std::exception& ex)
    {
@@ -73,7 +73,7 @@ void at_th_entry(boost::basic_thread_pool& )

 int test_generic_executor_ref()
 {
-  // std::cout << BOOST_CONTEXTOF << std::endl;
+  std::cout << BOOST_CONTEXTOF << std::endl;
  {
    try
    {
@@ -83,8 +83,8 @@ int test_generic_executor_ref()
        {
          boost::future<int> t1 = boost::async(ea, &f1);
          boost::future<int> t2 = boost::async(ea, &f1);
-          // std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
-          // std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
        }
        submit_some(ea);
        {
@@ -93,41 +93,51 @@ int test_generic_executor_ref()
          boost::future<int> t2 = boost::async(ea3, &f1);
          //boost::future<int> t2 = boost::async(ea3, f2, 1); // todo this doesn't compiles yet on C++11
          //boost::future<int> t2 = boost::async(ea3, boost::bind(f2, 1)); // todo this doesn't compiles yet on C++98
-          // std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
-          // std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
+          std::cout << BOOST_CONTEXTOF << " t2= " << t2.get() << std::endl;
        }
        submit_some(ea);
      }
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::loop_executor ea2;
        submit_some( ea2);
        ea2.run_queued_closures();
      }
 #if ! defined(BOOST_NO_CXX11_RVALUE_REFERENCES)
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::basic_thread_pool ea1(4);
        boost::serial_executor ea2(ea1);
        submit_some(ea2);
      }
 #endif
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::inline_executor ea1;
        submit_some(ea1);
      }
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        //boost::thread_executor ea1;
        //submit_some(ea1);
      }
-      // std::cout << BOOST_CONTEXTOF << std::endl;
+      std::cout << BOOST_CONTEXTOF << std::endl;
      {
        boost::basic_thread_pool  ea(4, at_th_entry);
        boost::future<int> t1 = boost::async(ea, &f1);
-        // std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
+        std::cout << BOOST_CONTEXTOF << " t1= " << t1.get() << std::endl;
      }
+      std::cout << BOOST_CONTEXTOF << std::endl;
+      {
+        boost::basic_thread_pool  ea(4, at_th_entry);
+        boost::async(ea, &f1);
+        std::cout << BOOST_CONTEXTOF << std::endl;
+      }
+      std::cout << BOOST_CONTEXTOF << std::endl;
+      boost::this_thread::sleep_for(boost::chrono::milliseconds(200));
+      std::cout << BOOST_CONTEXTOF << std::endl;
+
    }
    catch (std::exception& ex)
    {
@@ -10,6 +10,7 @@
 // See http://www.boost.org/libs/thread for documentation.
 //
 //////////////////////////////////////////////////////////////////////////////
+#include <iostream>

 #include <boost/thread/detail/config.hpp>
 #include <boost/thread/concurrent_queues/detail/sync_queue_base.hpp>
@@ -150,13 +151,19 @@ namespace concurrent
  template <class ValueType, class Container>
  queue_op_status sync_queue<ValueType, Container>::wait_pull(ValueType& elem, unique_lock<mutex>& lk)
  {
+    //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    if (super::empty(lk))
    {
+      //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
      if (super::closed(lk)) return queue_op_status::closed;
    }
+    //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    bool has_been_closed = super::wait_until_not_empty_or_closed(lk);
+    //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    if (has_been_closed) return queue_op_status::closed;
+    //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    pull(elem, lk);
+    //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    return queue_op_status::success;
  }

@@ -87,7 +87,9 @@ namespace executors
        {
          work task;
          queue_op_status st = work_queue.wait_pull(task);
-          if (st == queue_op_status::closed) return;
+          if (st == queue_op_status::closed) {
+            return;
+          }
          task();
        }
      }
@@ -811,7 +811,7 @@ rule thread-compile ( sources : reqs * : name )
          [ thread-run2 ../example/executor.cpp : ex_executor ]
          [ thread-run2 ../example/generic_executor_ref.cpp : ex_generic_executor_ref ]
          [ thread-run2 ../example/serial_executor.cpp : ex_serial_executor ]
-          [ thread-run2 ../example/serial_executor_cont.cpp : ex_serial_executor_cont ]
+          #[ thread-run2 ../example/serial_executor_cont.cpp : ex_serial_executor_cont ]
          [ thread-run2 ../example/future_when_all.cpp : ex_future_when_all ]
          [ thread-run2 ../example/parallel_accumulate.cpp : ex_parallel_accumulate ]
          [ thread-run2 ../example/parallel_quick_sort.cpp : ex_parallel_quick_sort ]
@@ -958,9 +958,12 @@ rule thread-compile ( sources : reqs * : name )
    explicit ts_ ;
    test-suite ts_
    :
+          #[ thread-run test_11256.cpp ] 
          #[ thread-run test_11256.cpp ]
          #[ thread-run test_11499.cpp ]
-          [ thread-run test_11611.cpp ]
+          #[ thread-run test_11611.cpp ]
+          [ thread-run2-noit ./sync/futures/shared_future/then_executor_pass.cpp : shared_future__then_executor_p2 ]
+          [ thread-run2-noit ./sync/futures/future/then_executor_pass.cpp : future__then_executor_p2 ]
    ;


@@ -38,6 +38,27 @@ struct TestCallback
    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    //boost::future<void> ff = future.get();

+    return boost::make_ready_future();
+  }
+  result_type operator()(boost::shared_future<void> future) const
+  {
+    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+    assert(future.is_ready());
+    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+    future.wait();
+    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+    return boost::make_ready_future();
+  }
+
+  result_type operator()(boost::shared_future<boost::future<void> > future) const
+  {
+    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+    assert(future.is_ready());
+    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+    assert(future.get().is_ready());
+    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+    //boost::future<void> ff = future.get();
+
    return boost::make_ready_future();
  }
 };
@@ -75,11 +96,17 @@ int main()
  std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
  for (int i=0; i< number_of_tests; i++)
  {
-    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
    auto f1 = boost::make_ready_future().then(TestCallback());
    BOOST_STATIC_ASSERT(std::is_same<decltype(f1), boost::future<boost::future<void> > >::value);
    boost::future<void> f2 = f1.get();
-    std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+  }
+  std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
+  {
+    auto f1 = boost::make_ready_future().then(TestCallback());
+    BOOST_STATIC_ASSERT(std::is_same<decltype(f1), boost::future<boost::future<void> > >::value);
+    auto f3 = f1.then(TestCallback());
+    BOOST_STATIC_ASSERT(std::is_same<decltype(f3), boost::future<boost::future<void> > >::value);
+    f3.wait();
  }
  std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
  for (int i=0; i< number_of_tests; i++)
@@ -114,8 +141,6 @@ int main()
    BOOST_STATIC_ASSERT(std::is_same<decltype(f3), boost::future<boost::future<void> > >::value);
    f3.wait();
  }
-
-
  std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
  for (int i=0; i< number_of_tests; i++)
  {
Author	SHA1	Message	Date
Vicente J. Botet Escriba	0ab63b9248	Remove counting when async future blocking.	2015-10-17 11:38:06 +02:00
Vicente J. Botet Escriba	73053e4abe	merge from develop.	2015-10-17 10:51:50 +02:00
Vicente J. Botet Escriba	b4744a2aa8	then can be now sync. The destructor of a shared state that joins check for thread_ifbefore join. Adde continuation_shared_state to share common behavior.	2015-10-17 00:44:45 +02:00
Vicente J. Botet Escriba	4169bcee44	Addoverloadsfor shared.	2015-10-17 00:40:53 +02:00
Vicente J. Botet Escriba	baf516e82f	comment serial conttest.	2015-10-17 00:40:23 +02:00
Vicente J. Botet Escriba	cc309eef8d	refactor ::then(f) in function of ::then(launch_policy, f).	2015-10-15 23:31:38 +02:00
Vicente J. Botet Escriba	45cc1704ef	fix some missing decay. Cleanup adding init.	2015-10-15 19:19:07 +02:00
Vicente J. Botet Escriba	278a06fd47	Merge pull request #71 from jhunold/executors_fix Fix executor forward	2015-10-12 23:49:48 +02:00
Jürgen Hunold	6ed577f4eb	Fix executor forward correct namespace and type	2015-10-12 08:01:19 +02:00
Vicente J. Botet Escriba	bf8459cf23	Merge from develop.	2015-10-11 23:34:45 +02:00
Vicente J. Botet Escriba	773f8bfcb4	replace >> by > >.	2015-09-30 23:57:00 +02:00
Vicente J. Botet Escriba	8afcbe22af	uncomment fixed test after merge from develop.	2015-09-30 00:08:16 +02:00
Vicente J. Botet Escriba	450f34daed	merge from develop.	2015-09-30 00:01:33 +02:00
Vicente J. Botet Escriba	958b773d05	Merge from develop.	2015-05-18 22:35:17 +02:00
Vicente J. Botet Escriba	1480d6fc99	simplify launch_continuation interface.	2015-05-04 18:24:09 +02:00
Vicente J. Botet Escriba	131c92a7d8	propagate unwrap fixes.	2015-05-04 00:19:12 +02:00
Vicente J. Botet Escriba	c50d0ac4ba	try to fix issue with continuation on unwrap shared state.	2015-05-03 16:51:41 +02:00
Vicente J. Botet Escriba	650956bd3b	First draft of non-blocking. Fixed issue with basic_thread_pool scoped threads. Now when the result of async(ex,f) is not stored, there is a lifetime issue with shared_state_nullary_task use of direct pointer to the shared state, however there is some weak_ptr issues :(.	2015-05-03 10:13:08 +02:00