Futures attempt to execute threads directly if those have not started…

… executing

- adding new API function hpx::threads::get_outer_self_id

.jenkins/cscs/env-clang-13.sh

@@ -30,6 +30,7 @@ configure_extra_options+=" -DHPX_WITH_COMPILER_WARNINGS=ON"
 configure_extra_options+=" -DHPX_WITH_COMPILER_WARNINGS_AS_ERRORS=ON"
 configure_extra_options+=" -DHPX_WITH_SPINLOCK_DEADLOCK_DETECTION=ON"
 configure_extra_options+=" -DHPX_WITH_UNITY_BUILD=ON"
+configure_extra_options+=" -DHPX_COROUTINES_WITH_THREAD_SCHEDULE_HINT_RUNS_AS_CHILD=ON"
 
 # enable extra counters to verify everything compiles
 configure_extra_options+=" -DHPX_WITH_BACKGROUND_THREAD_COUNTERS=ON"

.jenkins/lsu/env-gcc-9.sh

@@ -27,3 +27,5 @@ configure_extra_options+=" -DHPX_WITH_PARCELPORT_LCI_BACKEND=ibv"
 
 # The pwrapi library still needs to be set up properly on rostam
 # configure_extra_options+=" -DHPX_WITH_POWER_COUNTER=ON"
+
+configure_extra_options+=" -DHPX_COROUTINES_WITH_THREAD_SCHEDULE_HINT_RUNS_AS_CHILD=ON"

components/iostreams/src/server/output_stream.cpp

@@ -87,9 +87,10 @@ namespace hpx { namespace iostreams { namespace server {
     {    // {{{
         // Perform the IO in another OS thread.
         detail::buffer in(buf_in);
-        hpx::get_thread_pool("io_pool")->get_io_service().post(hpx::bind_front(
-            &output_stream::call_write_sync, this, locality_id, count,
-            std::ref(in), threads::thread_id_ref_type(threads::get_self_id())));
+        hpx::get_thread_pool("io_pool")->get_io_service().post(
+            hpx::bind_front(&output_stream::call_write_sync, this, locality_id,
+                count, std::ref(in),
+                threads::thread_id_ref_type(threads::get_outer_self_id())));
 
         // Sleep until the worker thread wakes us up.
         this_thread::suspend(threads::thread_schedule_state::suspended,

docs/sphinx/manual/hpx_runtime_and_resources.rst

@@ -42,12 +42,11 @@ turned on, work stealing is done from queues associated with the same NUMA domai
 first, only after that work is stolen from other NUMA domains.
 
 This scheduler is enabled at build time by default using the FIFO
-(first-in-first-out) queing policy. This policy can be invoked using
+(first-in-first-out) queueing policy. This policy can be invoked using
 :option:`--hpx:queuing`\ ``local-priority-fifo``. The scheduler can also be
 enabled using the LIFO (last-in-first-out) policy. This is not the default
 policy and must be invoked using the command line option
-:option:`--hpx:queuing`\
-``=local-priority-lifo``.
+:option:`--hpx:queuing`\ ``local-priority-lifo``.
 
 Static priority scheduling policy
 ---------------------------------
@@ -63,7 +62,7 @@ robin fashion. There is no thread stealing in this policy.
 Local scheduling policy
 -----------------------
 
-* invoke using: :option:`--hpx:queuing`\ ``=local`` (or ``-ql``)
+* invoke using: :option:`--hpx:queuing`\ ``local`` (or ``-ql``)
 * flag to turn on for build: ``HPX_THREAD_SCHEDULERS=all`` or
   ``HPX_THREAD_SCHEDULERS=local``
 
@@ -73,7 +72,7 @@ thread pulls its tasks (user threads).
 Static scheduling policy
 ------------------------
 
-* invoke using: :option:`--hpx:queuing`\ ``=static``
+* invoke using: :option:`--hpx:queuing`\ ``static``
 * flag to turn on for build: ``HPX_THREAD_SCHEDULERS=all`` or
   ``HPX_THREAD_SCHEDULERS=static``
 
@@ -84,7 +83,7 @@ robin fashion. There is no thread stealing in this policy.
 Priority ABP scheduling policy
 ------------------------------
 
-* invoke using: :option:`--hpx:queuing`\ ``=abp-priority-fifo``
+* invoke using: :option:`--hpx:queuing`\ ``abp-priority-fifo``
 * flag to turn on for build: ``HPX_THREAD_SCHEDULERS=all`` or
   ``HPX_THREAD_SCHEDULERS=abp-priority``
 
@@ -102,8 +101,8 @@ domain first, only after that work is stolen from other NUMA domains.
 
 This scheduler can be used with two underlying queuing policies (FIFO:
 first-in-first-out, and LIFO: last-in-first-out). In order to use the LIFO
-policy use the command line option :option:`--hpx:queuing`\
-``=abp-priority-lifo``.
+policy use the command line option
+:option:`--hpx:queuing`\ ``=abp-priority-lifo``.
 
 ..
     Questions, concerns and notes:

libs/core/async_base/include/hpx/async_base/scheduling_properties.hpp

@@ -63,6 +63,11 @@ namespace hpx { namespace execution { namespace experimental {
     {
     } with_priority{};
 
+    template <>
+    struct is_scheduling_property<with_priority_t> : std::true_type
+    {
+    };
+
     inline constexpr struct get_priority_t final
       : hpx::functional::detail::tag_fallback<get_priority_t>
     {
@@ -87,6 +92,11 @@ namespace hpx { namespace execution { namespace experimental {
     {
     } with_stacksize{};
 
+    template <>
+    struct is_scheduling_property<with_stacksize_t> : std::true_type
+    {
+    };
+
     inline constexpr struct get_stacksize_t final
       : hpx::functional::detail::tag_fallback<get_stacksize_t>
     {
@@ -111,6 +121,11 @@ namespace hpx { namespace execution { namespace experimental {
     {
     } with_hint{};
 
+    template <>
+    struct is_scheduling_property<with_hint_t> : std::true_type
+    {
+    };
+
     inline constexpr struct get_hint_t final
       : hpx::functional::detail::tag_fallback<get_hint_t>
     {
@@ -135,6 +150,11 @@ namespace hpx { namespace execution { namespace experimental {
     {
     } with_annotation{};
 
+    template <>
+    struct is_scheduling_property<with_annotation_t> : std::true_type
+    {
+    };
+
     inline constexpr struct get_annotation_t final
       : hpx::functional::detail::tag_fallback<get_annotation_t>
     {

libs/core/coroutines/CMakeLists.txt

@@ -1,4 +1,4 @@
-# Copyright (c) 2019-2020 The STE||AR-Group
+# Copyright (c) 2019-2021 The STE||AR-Group
 #
 # SPDX-License-Identifier: BSL-1.0
 # Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -14,6 +14,23 @@ hpx_option(
   MODULE COROUTINES
 )
 
+hpx_option(
+  HPX_COROUTINES_WITH_THREAD_SCHEDULE_HINT_RUNS_AS_CHILD
+  BOOL
+  "Futures attempt to run associated threads directly if those have not been started (default: OFF)"
+  OFF
+  CATEGORY "Thread Manager"
+  ADVANCED
+  MODULE COROUTINES
+)
+
+if(HPX_COROUTINES_WITH_THREAD_SCHEDULE_HINT_RUNS_AS_CHILD)
+  hpx_add_config_define_namespace(
+    DEFINE HPX_COROUTINES_HAVE_THREAD_SCHEDULE_HINT_RUNS_AS_CHILD
+    NAMESPACE COROUTINES
+  )
+endif()
+
 set(coroutines_headers
     hpx/coroutines/coroutine.hpp
     hpx/coroutines/coroutine_fwd.hpp
@@ -29,6 +46,7 @@ set(coroutines_headers
     hpx/coroutines/detail/coroutine_impl.hpp
     hpx/coroutines/detail/coroutine_self.hpp
     hpx/coroutines/detail/coroutine_stackful_self.hpp
+    hpx/coroutines/detail/coroutine_stackful_self_direct.hpp
     hpx/coroutines/detail/coroutine_stackless_self.hpp
     hpx/coroutines/detail/get_stack_pointer.hpp
     hpx/coroutines/detail/posix_utility.hpp

libs/core/coroutines/include/hpx/coroutines/coroutine.hpp

@@ -148,6 +148,11 @@ namespace hpx { namespace threads { namespace coroutines {
             return impl_.result();
         }
 
+        HPX_FORCEINLINE result_type invoke_directly(arg_type arg = arg_type())
+        {
+            return impl_.invoke_directly(arg);
+        }
+
         bool is_ready() const
         {
             return impl_.is_ready();

libs/core/coroutines/include/hpx/coroutines/detail/context_generic_context.hpp

@@ -243,8 +243,11 @@ namespace hpx { namespace threads { namespace coroutines {
 #endif
             }
 
-            void reset_stack()
+            void reset_stack(bool direct_execution)
             {
+                if (direct_execution)
+                    return;
+
                 if (ctx_)
                 {
 #if defined(HPX_USE_POSIX_STACK_UTILITIES)

libs/core/coroutines/include/hpx/coroutines/detail/context_linux_x86.hpp

@@ -72,11 +72,10 @@
 #include <sanitizer/asan_interface.h>
 #endif
 
-// Defining HPX_COROUTINE_NO_SEPARATE_CALL_SITES will disable separate
-// invoke, and yield swap_context functions. Separate calls sites
-// increase performance by 25% at least on P4 for invoke+yield back loops
-// at the cost of a slightly higher instruction cache use and is thus enabled by
-// default.
+// Defining HPX_COROUTINE_NO_SEPARATE_CALL_SITES will disable separate invoke,
+// and yield swap_context functions. Separate calls sites increase performance
+// by 25% at least on P4 for invoke+yield back loops at the cost of a slightly
+// higher instruction cache use and is thus enabled by default.
 
 #if defined(__x86_64__)
 extern "C" void swapcontext_stack(void***, void**) noexcept;
@@ -111,9 +110,9 @@ namespace hpx::threads::coroutines {
             (sigsegv_ptr - stk_ptr) :
             (stk_ptr - sigsegv_ptr);
 
-        // check the stack addresses, if they're < 10 apart, terminate
-        // program should filter segmentation faults caused by
-        // coroutine stack overflows from 'genuine' stack overflows
+        // check the stack addresses, if they're < 10 apart, terminate program
+        // should filter segmentation faults caused by coroutine stack overflows
+        // from 'genuine' stack overflows
         //
         if (static_cast<size_t>(addr_delta) <
             COROUTINE_STACKOVERFLOW_ADDR_EPSILON)
@@ -231,8 +230,8 @@ namespace hpx::threads::coroutines::detail::lx {
                 static_cast<void**>(m_sp) - 64 / sizeof(void*), 0, 3);
         }
 
-        // Free function. Saves the current context in @p from
-        // and restores the context in @p to.
+        // Free function. Saves the current context in @p from and restores the
+        // context in @p to.
         // @note This function is found by ADL.
         friend void swap_context(x86_linux_context_impl_base& from,
             x86_linux_context_impl_base const& to, default_hint);
@@ -287,8 +286,8 @@ namespace hpx::threads::coroutines::detail::lx {
 
         typedef x86_linux_context_impl_base context_impl_base;
 
-        // Create a context that on restore invokes Functor on
-        //  a new stack. The stack size can be optionally specified.
+        // Create a context that on restore invokes Functor on a new stack. The
+        // stack size can be optionally specified.
         explicit x86_linux_context_impl(std::ptrdiff_t stack_size = -1)
           : m_stack_size(stack_size == -1 ?
                     static_cast<std::ptrdiff_t>(default_stack_size) :
@@ -370,8 +369,11 @@ namespace hpx::threads::coroutines::detail::lx {
             return m_stack_size;
         }
 
-        void reset_stack()
+        void reset_stack(bool direct_execution)
         {
+            if (direct_execution)
+                return;
+
             HPX_ASSERT(m_stack);
             if (posix::reset_stack(
                     m_stack, static_cast<std::size_t>(m_stack_size)))
@@ -384,7 +386,10 @@ namespace hpx::threads::coroutines::detail::lx {
 
         void rebind_stack()
         {
-            HPX_ASSERT(m_stack);
+            // directly executed coroutine, no need to allocate a stack
+            if (m_stack == nullptr)
+                return;
+
 #if defined(HPX_HAVE_COROUTINE_COUNTERS)
             increment_stack_recycle_count();
 #endif
@@ -532,8 +537,8 @@ namespace hpx::threads::coroutines::detail::lx {
 #endif
     };
 
-    // Free function. Saves the current context in @p from
-    // and restores the context in @p to.
+    // Free function. Saves the current context in @p from and restores the
+    // context in @p to.
     // @note This function is found by ADL.
     inline void swap_context(x86_linux_context_impl_base& from,
         x86_linux_context_impl_base const& to, default_hint)

libs/core/coroutines/include/hpx/coroutines/detail/context_posix.hpp

@@ -346,8 +346,11 @@ namespace hpx { namespace threads { namespace coroutines {
 #endif
             }
 
-            void reset_stack()
+            void reset_stack(bool direct_execution)
             {
+                if (direct_execution)
+                    return;
+
                 if (m_stack)
                 {
                     if (posix::reset_stack(

libs/core/coroutines/include/hpx/coroutines/detail/context_windows_fibers.hpp

@@ -290,7 +290,7 @@ namespace hpx { namespace threads { namespace coroutines {
                 return stacksize_;
             }
 
-            constexpr void reset_stack() noexcept {}
+            constexpr void reset_stack(bool) noexcept {}
 
             void rebind_stack() noexcept
             {

libs/core/coroutines/include/hpx/coroutines/detail/coroutine_impl.hpp

@@ -82,6 +82,10 @@ namespace hpx { namespace threads { namespace coroutines { namespace detail {
         // execute the coroutine using normal context switching
         HPX_CORE_EXPORT void operator()() noexcept;
 
+        // execute the coroutine function directly in the context of the calling
+        // thread
+        HPX_CORE_EXPORT result_type invoke_directly(arg_type arg);
+
     public:
         void bind_result(result_type res)
         {
@@ -97,7 +101,7 @@ namespace hpx { namespace threads { namespace coroutines { namespace detail {
         {
             HPX_ASSERT(m_arg);
             return m_arg;
-        };
+        }
 
         void bind_args(arg_type* arg) noexcept
         {
@@ -116,16 +120,18 @@ namespace hpx { namespace threads { namespace coroutines { namespace detail {
             this->super_type::init();
         }
 
-        void reset()
+        void reset(bool direct_execution)
         {
             // First reset the function and arguments
+            m_result =
+                result_type(thread_schedule_state::unknown, invalid_thread_id);
             m_arg = nullptr;
             m_fun.reset();
 
             // Then reset the id and stack as they may be used by the
             // destructors of the thread function above
             this->super_type::reset();
-            this->reset_stack();
+            this->reset_stack(direct_execution);
         }
 
         void rebind(functor_type&& f, thread_id_type id)

libs/core/coroutines/include/hpx/coroutines/detail/coroutine_self.hpp

@@ -120,6 +120,11 @@ namespace hpx { namespace threads { namespace coroutines { namespace detail {
 
         virtual thread_id_type get_thread_id() const = 0;
 
+        virtual thread_id_type get_outer_thread_id() const
+        {
+            return get_thread_id();
+        }
+
         virtual std::size_t get_thread_phase() const = 0;
 
         virtual std::ptrdiff_t get_available_stack_space() = 0;

libs/core/coroutines/include/hpx/coroutines/detail/coroutine_stackful_self.hpp

@@ -24,7 +24,7 @@ namespace hpx { namespace threads { namespace coroutines { namespace detail {
     {
     public:
         explicit coroutine_stackful_self(
-            impl_type* pimpl, coroutine_self* next_self = nullptr)
+            coroutine_impl* pimpl, coroutine_self* next_self = nullptr)
           : coroutine_self(next_self)
           , pimpl_(pimpl)
         {
@@ -133,6 +133,7 @@ namespace hpx { namespace threads { namespace coroutines { namespace detail {
         {
             return pimpl_;
         }
+
         coroutine_impl* pimpl_;
     };
 }}}}    // namespace hpx::threads::coroutines::detail