Adding component base class which ties a component instance to a given executor

examples/quickstart/component_with_executor.cpp

@@ -1,4 +1,4 @@
-//  Copyright (c) 2014 Hartmut Kaiser
+//  Copyright (c) 2014-2017 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)
@@ -7,17 +7,25 @@
 #include <hpx/include/components.hpp>
 #include <hpx/include/actions.hpp>
 #include <hpx/include/iostreams.hpp>
-#include <hpx/include/thread_executors.hpp>
+#include <hpx/include/parallel_executors.hpp>
 
 #include <utility>
 
 ///////////////////////////////////////////////////////////////////////////////
 // Define a base component which exposes the required interface
 struct hello_world_server
-  : hpx::components::component_base<hello_world_server>
+  : hpx::components::executor_component<
+        hpx::parallel::local_priority_queue_executor,
+        hpx::components::component_base<hello_world_server> >
 {
+    typedef hpx::parallel::local_priority_queue_executor executor_type;
+    typedef hpx::components::executor_component<
+            executor_type, hpx::components::component_base<hello_world_server>
+        > base_type;
+
+    // run on all available cores
     hello_world_server()
-      : sched_(hpx::get_num_worker_threads())   // run on all available cores
+      : base_type(executor_type(hpx::get_num_worker_threads()))
     {}
 
     void print() const
@@ -26,34 +34,6 @@ struct hello_world_server
     }
 
     HPX_DEFINE_COMPONENT_ACTION(hello_world_server, print, print_action);
-
-    ///////////////////////////////////////////////////////////////////////////
-    // wrap given function into a nullary function as expected by the executor
-    static void func(hpx::threads::thread_function_type f)
-    {
-        f(hpx::threads::wait_signaled);
-    }
-
-    /// This is the default hook implementation for schedule_thread which
-    /// forwards to the default scheduler.
-    static void schedule_thread(hpx::naming::address::address_type lva,
-        hpx::threads::thread_init_data& data,
-        hpx::threads::thread_state_enum initial_state)
-    {
-        hpx::util::thread_description desc(&hello_world_server::func);
-#ifdef HPX_HAVE_THREAD_DESCRIPTION
-        desc = data.description;
-#endif
-
-        hpx::get_lva<hello_world_server>::call(lva)->sched_.add(
-            hpx::util::bind(
-                hpx::util::one_shot(&hello_world_server::func),
-                std::move(data.func)),
-            desc, initial_state);
-    }
-
-private:
-    hpx::threads::executors::local_priority_queue_executor sched_;
 };
 
 typedef hpx::components::component<hello_world_server> server_type;

hpx/include/components.hpp

@@ -41,6 +41,7 @@
 #include <hpx/runtime/components/server/simple_component_base.hpp>
 
 #include <hpx/runtime/components/server/locking_hook.hpp>
+#include <hpx/runtime/components/server/executor_component.hpp>
 #include <hpx/runtime/components/server/migration_support.hpp>
 
 #include <hpx/runtime/components/copy_component.hpp>

hpx/runtime/components/server/executor_component.hpp

@@ -0,0 +1,98 @@
+//  Copyright (c) 2017 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_RUNTIME_COMPONENTS_SERVER_EXECUTOR_COMPONENT_FEB_09_2017_0839PM)
+#define HPX_RUNTIME_COMPONENTS_SERVER_EXECUTOR_COMPONENT_FEB_09_2017_0839PM
+
+#include <hpx/config.hpp>
+#include <hpx/parallel/executors/executor_traits.hpp>
+#include <hpx/runtime/get_lva.hpp>
+#include <hpx/runtime/naming/address.hpp>
+#include <hpx/runtime/threads/executors.hpp>
+#include <hpx/runtime/threads/thread_data_fwd.hpp>
+#include <hpx/runtime/threads/thread_enums.hpp>
+#include <hpx/traits/is_launch_policy.hpp>
+#include <hpx/util/annotated_function.hpp>
+#include <hpx/util/deferred_call.hpp>
+#include <hpx/util/thread_description.hpp>
+
+#include <type_traits>
+#include <utility>
+
+namespace hpx { namespace components
+{
+    // This is a base class which allows to associate the execution of all
+    // actions for a particular component instance with a given executor.
+    template <typename Executor, typename BaseComponent>
+    struct executor_component :  BaseComponent
+    {
+    private:
+        typedef BaseComponent base_type;
+        typedef Executor executor_type;
+        typedef typename base_type::this_component_type this_component_type;
+
+    public:
+        template <typename ... Arg>
+        executor_component(executor_type const& exec, Arg &&... arg)
+          : base_type(std::forward<Arg>(arg)...),
+            exec_(exec)
+        {}
+
+        ///////////////////////////////////////////////////////////////////////
+        // wrap given function into a nullary function as expected by the
+        // executor
+        static void execute(hpx::threads::thread_function_type const& f)
+        {
+            f(hpx::threads::wait_signaled);
+        }
+
+        /// This is the default hook implementation for schedule_thread which
+        /// forwards to the executor instance associated with this component.
+        template <typename Executor_ = Executor>
+        static typename std::enable_if<
+            traits::is_threads_executor<Executor_>::value
+        >::type
+        schedule_thread(hpx::naming::address::address_type lva,
+            hpx::threads::thread_init_data& data,
+            hpx::threads::thread_state_enum initial_state)
+        {
+            hpx::util::thread_description desc(&executor_component::execute);
+#ifdef HPX_HAVE_THREAD_DESCRIPTION
+            desc = data.description;
+#endif
+            hpx::get_lva<executor_component>::call(lva)->exec_.add(
+                hpx::util::deferred_call(&executor_component::execute,
+                    std::move(data.func)),
+                desc, initial_state);
+        }
+
+        template <typename Executor_ = Executor>
+        static typename std::enable_if<
+            !traits::is_threads_executor<Executor_>::value
+        >::type
+        schedule_thread(hpx::naming::address::address_type lva,
+            hpx::threads::thread_init_data& data,
+            hpx::threads::thread_state_enum initial_state)
+        {
+            hpx::util::thread_description desc(&executor_component::execute);
+#ifdef HPX_HAVE_THREAD_DESCRIPTION
+            desc = data.description;
+#endif
+            hpx::parallel::executor_traits<executor_type>::async_execute(
+                hpx::get_lva<executor_component>::call(lva)->exec_,
+                hpx::util::deferred_call(
+                    hpx::util::annotated_function(
+                        &executor_component::execute, desc.get_description()
+                    ),
+                    std::move(data.func)));
+        }
+
+    protected:
+        executor_type exec_;
+    };
+}}
+
+#endif
+