Fixing return type calculation for bulk_then_execute.

- flyby: added util::functional::unwrap[_n|_all] - some restructuring required by circular #include dependencies
STEllAR-GROUP · Feb 28, 2018 · 7b7c183 · 7b7c183
1 parent 9cdd6bc
commit 7b7c183
Show file tree

Hide file tree

Showing 56 changed files with 982 additions and 436 deletions.
diff --git a/docs/CMakeLists.txt b/docs/CMakeLists.txt
@@ -143,7 +143,7 @@ set(doxygen_dependencies
     "${PROJECT_SOURCE_DIR}/hpx/performance_counters/manage_counter_type.hpp"
     "${PROJECT_SOURCE_DIR}/hpx/runtime_fwd.hpp"
     "${PROJECT_SOURCE_DIR}/hpx/runtime/applier_fwd.hpp"
-    "${PROJECT_SOURCE_DIR}/hpx/runtime/basename_registration.hpp"
+    "${PROJECT_SOURCE_DIR}/hpx/runtime/basename_registration_fwd.hpp"
     "${PROJECT_SOURCE_DIR}/hpx/runtime/find_here.hpp"
     "${PROJECT_SOURCE_DIR}/hpx/runtime/find_localities.hpp"
     "${PROJECT_SOURCE_DIR}/hpx/runtime/get_colocation_id.hpp"

diff --git a/hpx/apply.hpp b/hpx/apply.hpp
@@ -7,11 +7,11 @@
 #define HPX_APPLY_APR_16_20012_0943AM
 
 #include <hpx/config.hpp>
-#include <hpx/lcos/future.hpp>
 #include <hpx/runtime/applier/apply.hpp>
 #include <hpx/runtime/applier/apply_continue.hpp>
 #include <hpx/runtime/threads/thread_executor.hpp>
 #include <hpx/runtime/threads/thread_helpers.hpp>
+#include <hpx/runtime_fwd.hpp>
 #include <hpx/traits/is_executor.hpp>
 #include <hpx/traits/is_launch_policy.hpp>
 #include <hpx/util/bind_action.hpp>

diff --git a/hpx/compute/cuda/target_distribution_policy.hpp b/hpx/compute/cuda/target_distribution_policy.hpp
@@ -12,7 +12,7 @@
 
 #if defined(HPX_HAVE_CUDA)
 
-#include <hpx/dataflow.hpp>
+#include <hpx/lcos/dataflow.hpp>
 #include <hpx/lcos/future.hpp>
 #include <hpx/runtime/components/stubs/stub_base.hpp>
 #include <hpx/runtime/serialization/base_object.hpp>

diff --git a/hpx/compute/host/target_distribution_policy.hpp b/hpx/compute/host/target_distribution_policy.hpp
@@ -10,7 +10,7 @@
 
 #include <hpx/config.hpp>
 
-#include <hpx/dataflow.hpp>
+#include <hpx/lcos/dataflow.hpp>
 #include <hpx/lcos/future.hpp>
 #include <hpx/runtime/components/stubs/stub_base.hpp>
 #include <hpx/runtime/serialization/base_object.hpp>

diff --git a/hpx/dataflow.hpp b/hpx/dataflow.hpp
@@ -6,6 +6,7 @@
 #ifndef HPX_DATAFLOW_HPP
 #define HPX_DATAFLOW_HPP
 
+#include <hpx/lcos/future.hpp>
 #include <hpx/lcos/dataflow.hpp>
 #include <hpx/lcos/local/dataflow.hpp>
 

diff --git a/hpx/hpx.hpp b/hpx/hpx.hpp
@@ -6,6 +6,9 @@
 #if !defined(HPX_MAR_24_2008_1118AM)
 #define HPX_MAR_24_2008_1118AM
 
+#include <hpx/apply.hpp>
+#include <hpx/async.hpp>
+#include <hpx/dataflow.hpp>
 #include <hpx/exception.hpp>
 #include <hpx/include/lcos.hpp>
 #include <hpx/include/performance_counters.hpp>

diff --git a/hpx/lcos/broadcast.hpp b/hpx/lcos/broadcast.hpp
@@ -135,6 +135,7 @@ namespace hpx { namespace lcos
 #define HPX_LCOS_BROADCAST_HPP
 
 #include <hpx/config.hpp>
+#include <hpx/apply.hpp>
 #include <hpx/lcos/detail/async_colocated.hpp>
 #include <hpx/lcos/future.hpp>
 #include <hpx/lcos/when_all.hpp>

diff --git a/hpx/lcos/dataflow.hpp b/hpx/lcos/dataflow.hpp
@@ -1,18 +1,21 @@
-//  Copyright (c) 2007-2017 Hartmut Kaiser
+//  Copyright (c) 2007-2018 Hartmut Kaiser
 //
 //  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)
 
 // hpxinspect:nodeprecatedinclude:boost/ref.hpp
 // hpxinspect:nodeprecatedname:boost::reference_wrapper
 
+#include <hpx/config.hpp>
+
+// Intentionally #include future.hpp outside of the guards as it may #include
+// dataflow.hpp itself
+#include <hpx/lcos/future.hpp>
+
 #ifndef HPX_LCOS_DATAFLOW_HPP
 #define HPX_LCOS_DATAFLOW_HPP
 
-#include <hpx/config.hpp>
-#include <hpx/apply.hpp>
 #include <hpx/lcos/detail/future_transforms.hpp>
-#include <hpx/lcos/future.hpp>
 #include <hpx/runtime/get_worker_thread_num.hpp>
 #include <hpx/runtime/launch_policy.hpp>
 #include <hpx/traits/acquire_future.hpp>
@@ -36,6 +39,7 @@
 #include <hpx/parallel/executors/v1/executor_traits.hpp>
 #endif
 #include <hpx/parallel/executors/execution.hpp>
+#include <hpx/parallel/executors/parallel_executor.hpp>
 
 #include <boost/intrusive_ptr.hpp>
 #include <boost/ref.hpp>

diff --git a/hpx/lcos/detail/future_transforms.hpp b/hpx/lcos/detail/future_transforms.hpp
@@ -6,6 +6,7 @@
 #ifndef HPX_LCOS_DETAIL_FUTURE_TRANSFORMS_HPP
 #define HPX_LCOS_DETAIL_FUTURE_TRANSFORMS_HPP
 
+#include <hpx/lcos/detail/future_traits.hpp>
 #include <hpx/lcos_fwd.hpp>
 #include <hpx/traits/acquire_future.hpp>
 #include <hpx/traits/acquire_shared_state.hpp>

diff --git a/hpx/lcos/future.hpp b/hpx/lcos/future.hpp
@@ -37,6 +37,7 @@
 #include <hpx/util/invoke.hpp>
 #include <hpx/util/lazy_enable_if.hpp>
 #include <hpx/util/result_of.hpp>
+#include <hpx/util/serialize_exception.hpp>
 #include <hpx/util/steady_clock.hpp>
 #include <hpx/util/void_guard.hpp>
 

diff --git a/hpx/lcos/when_all_fwd.hpp b/hpx/lcos/when_all_fwd.hpp
@@ -0,0 +1,26 @@
+//  Copyright (c) 2018 Hartmut Kaiser
+//
+//  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)
+
+#if !defined(HPX_LCOS_WHEN_ALL_FWD_HPP)
+#define HPX_LCOS_WHEN_ALL_FWD_HPP
+
+#include <hpx/config.hpp>
+#include <hpx/lcos_fwd.hpp>
+
+#include <vector>
+
+///////////////////////////////////////////////////////////////////////////////
+namespace hpx { namespace lcos
+{
+    // special forwarding function to break #include dependencies
+    HPX_API_EXPORT hpx::future<void> when_all_fwd(
+        std::vector<hpx::future<void>>&);
+
+    HPX_API_EXPORT hpx::future<void> when_all_fwd(
+        std::vector<hpx::future<void>>&&);
+}}
+
+#endif
+
diff --git a/hpx/parallel/executors/default_executor.hpp b/hpx/parallel/executors/default_executor.hpp
@@ -9,6 +9,7 @@
 #define HPX_PARALLEL_EXECUTORS_DEFAULT_EXECUTOR_AUG_24_2015_0624PM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/future.hpp>
 #include <hpx/parallel/executors/execution_parameters.hpp>
 #include <hpx/parallel/executors/thread_execution.hpp>
 #include <hpx/parallel/executors/thread_execution_information.hpp>

diff --git a/hpx/parallel/executors/execution.hpp b/hpx/parallel/executors/execution.hpp
@@ -13,9 +13,11 @@
 #include <hpx/parallel/executors/execution_fwd.hpp>
 
 #include <hpx/exception_list.hpp>
+#include <hpx/lcos/dataflow.hpp>
 #include <hpx/lcos/future.hpp>
 #include <hpx/lcos/wait_all.hpp>
 #include <hpx/traits/detail/wrap_int.hpp>
+#include <hpx/traits/executor_traits.hpp>
 #include <hpx/traits/future_access.hpp>
 #include <hpx/traits/future_then_result.hpp>
 #include <hpx/traits/future_traits.hpp>
@@ -1133,6 +1135,7 @@ namespace hpx { namespace parallel { namespace execution
                 typedef typename bulk_then_execute_result<
                         F, Shape, Future, Ts...
                     >::type result_type;
+
                 typedef typename hpx::traits::detail::shared_state_ptr<
                         result_type
                     >::type shared_state_type;
@@ -1304,15 +1307,17 @@ namespace hpx { namespace parallel { namespace execution
                     Future&& predecessor, Ts &&... ts)
             ->  typename hpx::traits::executor_future<
                     Executor,
-                    typename then_bulk_function_result<
+                    typename bulk_then_execute_result<
                         F, Shape, Future, Ts...
                     >::type
                 >::type
             {
+                // result_of_t<F(Shape::value_type, Future)>
                 typedef typename then_bulk_function_result<
                         F, Shape, Future, Ts...
                     >::type func_result_type;
 
+                // std::vector<future<func_result_type>>
                 typedef std::vector<typename hpx::traits::executor_future<
                         Executor, func_result_type, Ts...
                     >::type> result_type;
@@ -1321,18 +1326,33 @@ namespace hpx { namespace parallel { namespace execution
                     exec, std::forward<F>(f), shape,
                     hpx::util::make_tuple(std::forward<Ts>(ts)...));
 
+                // void or std::vector<func_result_type>
+                typedef typename bulk_then_execute_result<
+                        F, Shape, Future, Ts...
+                    >::type vector_result_type;
+
+                // future<vector_result_type>
+                typedef typename hpx::traits::executor_future<
+                        Executor, vector_result_type
+                    >::type result_future_type;
+
                 typedef typename hpx::traits::detail::shared_state_ptr<
-                        result_type
+                        result_future_type
                     >::type shared_state_type;
 
+                typedef typename std::decay<Future>::type future_type;
+
                 shared_state_type p =
-                    lcos::detail::make_continuation_exec<result_type>(
+                    lcos::detail::make_continuation_exec<result_future_type>(
                         std::forward<Future>(predecessor),
-                        std::forward<BulkExecutor>(exec), std::move(func));
-
-                typedef typename hpx::traits::executor_future<
-                        Executor, result_type
-                    >::type result_future_type;
+                        std::forward<BulkExecutor>(exec),
+                        [HPX_CAPTURE_MOVE(func)](future_type&& predecessor) mutable
+                        ->  result_future_type
+                        {
+                            return hpx::dataflow(
+                                hpx::util::functional::unwrap{},
+                                func(std::move(predecessor)));
+                        });
 
                 return hpx::traits::future_access<result_future_type>::create(
                     std::move(p));
@@ -1376,7 +1396,5 @@ namespace hpx { namespace parallel { namespace execution
     /// \endcond
 }}}
 
-
-
 #endif
 
diff --git a/hpx/parallel/executors/parallel_executor.hpp b/hpx/parallel/executors/parallel_executor.hpp
@@ -11,7 +11,7 @@
 #include <hpx/config.hpp>
 #include <hpx/async_launch_policy_dispatch.hpp>
 #include <hpx/lcos/future.hpp>
-#include <hpx/lcos/when_all.hpp>
+#include <hpx/lcos/when_all_fwd.hpp>
 #include <hpx/parallel/algorithms/detail/predicates.hpp>
 #include <hpx/parallel/executors/post_policy_dispatch.hpp>
 #include <hpx/parallel/executors/static_chunk_size.hpp>
@@ -197,7 +197,7 @@ namespace hpx { namespace parallel { namespace execution
 
                 HPX_ASSERT(size == 0);
 
-                return hpx::when_all(tasks);
+                return hpx::lcos::when_all_fwd(std::move(tasks));
             }
 
             // spawn all tasks sequentially

diff --git a/hpx/parallel/executors/pool_executor.hpp b/hpx/parallel/executors/pool_executor.hpp
@@ -9,6 +9,7 @@
 #define HPX_PARALLEL_EXECUTORS_POOL_EXECUTOR_FEB_17_2018_0327PM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/future.hpp>
 #include <hpx/parallel/executors/execution_parameters.hpp>
 #include <hpx/parallel/executors/thread_execution.hpp>
 #include <hpx/parallel/executors/thread_execution_information.hpp>

diff --git a/hpx/parallel/executors/service_executors.hpp b/hpx/parallel/executors/service_executors.hpp
@@ -9,6 +9,7 @@
 #define HPX_PARALLEL_EXECUTORS_SERVICE_EXECUTORS_MAY_15_2015_0548PM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/future.hpp>
 #include <hpx/parallel/executors/static_chunk_size.hpp>
 #include <hpx/parallel/executors/thread_execution.hpp>
 #include <hpx/runtime/threads/executors/service_executors.hpp>

diff --git a/hpx/parallel/executors/static_chunk_size.hpp b/hpx/parallel/executors/static_chunk_size.hpp
@@ -12,7 +12,7 @@
 #include <hpx/runtime/serialization/serialize.hpp>
 #include <hpx/traits/is_executor_parameters.hpp>
 
-#include <hpx/parallel/executors/execution_parameters.hpp>
+#include <hpx/parallel/executors/execution_parameters_fwd.hpp>
 
 #include <cstddef>
 #include <type_traits>

diff --git a/hpx/parallel/executors/this_thread_executors.hpp b/hpx/parallel/executors/this_thread_executors.hpp
@@ -7,6 +7,7 @@
 #define HPX_PARALLEL_EXECUTORS_THIS_THREAD_EXECUTORS_JUL_16_2015_0809PM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/future.hpp>
 #include <hpx/parallel/executors/execution_parameters.hpp>
 #include <hpx/parallel/executors/thread_execution.hpp>
 #include <hpx/parallel/executors/thread_execution_information.hpp>

diff --git a/hpx/parallel/executors/thread_execution.hpp b/hpx/parallel/executors/thread_execution.hpp
@@ -9,6 +9,7 @@
 #define HPX_PARALLEL_EXECUTORS_THREAD_EXECUTION_JAN_03_2017_1145AM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/dataflow.hpp>
 #include <hpx/lcos/future.hpp>
 #include <hpx/lcos/local/futures_factory.hpp>
 #include <hpx/runtime/threads/thread_executor.hpp>
@@ -192,22 +193,37 @@ namespace hpx { namespace threads
             >::type func_result_type;
 
         typedef std::vector<hpx::lcos::future<func_result_type> > result_type;
-        typedef hpx::lcos::future<result_type> result_future_type;
 
         auto func =
             parallel::execution::detail::make_fused_bulk_async_execute_helper<
                 result_type
             >(exec, std::forward<F>(f), shape,
                 hpx::util::make_tuple(std::forward<Ts>(ts)...));
 
+        // void or std::vector<func_result_type>
+        typedef typename parallel::execution::detail::bulk_then_execute_result<
+                F, Shape, Future, Ts...
+            >::type vector_result_type;
+
+        typedef hpx::future<vector_result_type> result_future_type;
+
         typedef typename hpx::traits::detail::shared_state_ptr<
-                result_type
+                result_future_type
             >::type shared_state_type;
 
+        typedef typename std::decay<Future>::type future_type;
+
         shared_state_type p =
-            lcos::detail::make_continuation_exec<result_type>(
+            lcos::detail::make_continuation_exec<result_future_type>(
                 std::forward<Future>(predecessor),
-                std::forward<Executor>(exec), std::move(func));
+                std::forward<Executor>(exec),
+                [HPX_CAPTURE_MOVE(func)](future_type&& predecessor) mutable
+                ->  result_future_type
+                {
+                    return hpx::dataflow(
+                        hpx::util::functional::unwrap{},
+                        func(std::move(predecessor)));
+                });
 
         return hpx::traits::future_access<result_future_type>::create(
             std::move(p));

diff --git a/hpx/parallel/executors/thread_pool_executors.hpp b/hpx/parallel/executors/thread_pool_executors.hpp
@@ -9,6 +9,7 @@
 #define HPX_PARALLEL_EXECUTORS_THREAD_POOL_EXECUTORS_MAY_15_2015_0548PM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/future.hpp>
 #include <hpx/parallel/executors/execution_parameters.hpp>
 #include <hpx/parallel/executors/thread_execution.hpp>
 #include <hpx/parallel/executors/thread_execution_information.hpp>

diff --git a/hpx/parallel/executors/thread_pool_os_executors.hpp b/hpx/parallel/executors/thread_pool_os_executors.hpp
@@ -9,6 +9,7 @@
 #define HPX_PARALLEL_EXECUTORS_THREAD_POOL_OS_EXECUTORS_AUG_22_2015_0739PM
 
 #include <hpx/config.hpp>
+#include <hpx/lcos/future.hpp>
 #include <hpx/parallel/executors/execution_parameters.hpp>
 #include <hpx/parallel/executors/thread_execution.hpp>
 #include <hpx/parallel/executors/thread_execution_information.hpp>

diff --git a/hpx/parallel/task_block.hpp b/hpx/parallel/task_block.hpp
@@ -10,8 +10,8 @@
 
 #include <hpx/config.hpp>
 #include <hpx/async.hpp>
-#include <hpx/dataflow.hpp>
 #include <hpx/exception.hpp>
+#include <hpx/lcos/dataflow.hpp>
 #include <hpx/lcos/future.hpp>
 #include <hpx/lcos/local/spinlock.hpp>
 #include <hpx/lcos/when_all.hpp>