Always return outermost thread id

-flyby: deprecate get_outer_self_id
-flyby: ignoring locks during termination detection

.circleci/config.yml

@@ -497,10 +497,8 @@ jobs:
       - run:
           name: Building Unit Tests (Algorithms)
           command: |
-              ninja -j2 -k 0 \
-                tests.unit.modules.algorithms.algorithms \
-                tests.unit.modules.algorithms.block
-#                tests.unit.modules.algorithms.datapar_algorithms
+              ninja -j 1 -k 0 \
+                tests.unit.modules.algorithms.algorithms
       - run:
           name: Running Unit Tests
           when: always
@@ -512,9 +510,7 @@ jobs:
                 --no-compress-output \
                 --output-on-failure \
                 --tests-regex \
-              "tests.unit.modules.algorithms.algorithms|\
-              tests.unit.modules.algorithms.block"
-#              "|tests.unit.modules.algorithms.datapar_algorithms"
+              "tests.unit.modules.algorithms.algorithms"
       - run:
           <<: *convert_xml
       - run:
@@ -526,6 +522,39 @@ jobs:
       - store_artifacts:
           path: tests.unit.algorithms
 
+  tests.unit.algorithms.block:
+    <<: *defaults
+    steps:
+      - attach_workspace:
+          at: /hpx
+      - run:
+          name: Building Unit Tests (Algorithms)
+          command: |
+              ninja -j2 -k 0 \
+                tests.unit.modules.algorithms.block
+      - run:
+          name: Running Unit Tests
+          when: always
+          command: |
+              ulimit -c unlimited
+              ctest \
+                --timeout 120 \
+                -T test \
+                --no-compress-output \
+                --output-on-failure \
+                --tests-regex \
+              "tests.unit.modules.algorithms.block"
+      - run:
+          <<: *convert_xml
+      - run:
+          <<: *move_core_dump
+      - run:
+          <<: *move_debug_log
+      - store_test_results:
+          path: tests.unit.algorithms.block
+      - store_artifacts:
+          path: tests.unit.algorithms.block
+
   tests.unit.container_algorithms:
     <<: *defaults
     steps:
@@ -913,6 +942,8 @@ workflows:
           <<: *core_dependency
       - tests.unit.algorithms:
           <<: *core_dependency
+      - tests.unit.algorithms.block:
+          <<: *core_dependency
       - tests.unit.container_algorithms:
           <<: *core_dependency
       - tests.unit.segmented_algorithms:
@@ -970,6 +1001,7 @@ workflows:
             - core
             - tests.examples
             - tests.unit.algorithms
+            - tests.unit.algorithms.block
             - tests.unit.container_algorithms
             - tests.unit.segmented_algorithms
             - tests.unit1

components/iostreams/src/server/output_stream.cpp

@@ -34,7 +34,7 @@ namespace hpx::iostreams::detail {
         ar << valid;
         if (valid)
         {
-            ar& data_;
+            ar & data_;
         }
     }
 
@@ -44,7 +44,7 @@ namespace hpx::iostreams::detail {
         ar >> valid;
         if (valid)
         {
-            ar& data_;
+            ar & data_;
         }
     }
 }    // namespace hpx::iostreams::detail
@@ -89,10 +89,9 @@ namespace hpx::iostreams::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_outer_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_self_id())));
 
         // Sleep until the worker thread wakes us up.
         this_thread::suspend(threads::thread_schedule_state::suspended,

examples/cancelable_action/cancelable_action/server/cancelable_action.hpp

@@ -1,4 +1,4 @@
-//  Copyright (c) 2007-2023 Hartmut Kaiser
+//  Copyright (c) 2007-2024 Hartmut Kaiser
 //
 //  SPDX-License-Identifier: BSL-1.0
 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -63,6 +63,9 @@ namespace examples::server {
             }
             ~reset_id()
             {
+                auto const mtx = outer_.mtx_;
+                std::lock_guard<hpx::mutex> l(*mtx);
+
                 [[maybe_unused]] hpx::thread::id const old_value = outer_.id_;
                 outer_.id_ = hpx::thread::id();
                 HPX_ASSERT(old_value != hpx::thread::id());
@@ -104,9 +107,15 @@ namespace examples::server {
             });
 
             auto const mtx = mtx_;
-            std::lock_guard<hpx::mutex> l(*mtx);
-            HPX_ASSERT(id_ != hpx::thread::id());
-            hpx::thread::interrupt(id_);
+
+            std::unique_lock<hpx::mutex> l(*mtx);
+            auto const id = id_;
+
+            if (id != hpx::thread::id())
+            {
+                l.unlock();
+                hpx::thread::interrupt(id);
+            }
         }
 
         HPX_DEFINE_COMPONENT_ACTION(cancelable_action, do_it, do_it_action)

libs/core/config/include/hpx/config/threads_stack.hpp

@@ -40,20 +40,22 @@
 #endif
 
 #if !defined(HPX_SMALL_STACK_SIZE)
-#  if defined(HPX_WINDOWS) && !defined(HPX_HAVE_GENERIC_CONTEXT_COROUTINES)
-#    define HPX_SMALL_STACK_SIZE_TARGET    0x4000        // 16kByte
-#  else
-#    if defined(HPX_DEBUG)
-#      define HPX_SMALL_STACK_SIZE_TARGET  0x20000       // 128kByte
+#  if !defined(HPX_SMALL_STACK_SIZE_TARGET)
+#    if defined(HPX_WINDOWS) && !defined(HPX_HAVE_GENERIC_CONTEXT_COROUTINES)
+#      define HPX_SMALL_STACK_SIZE_TARGET    0x4000        // 16kByte
 #    else
-#      if defined(__powerpc__) || defined(__INTEL_COMPILER)
-#         define HPX_SMALL_STACK_SIZE_TARGET  0x20000       // 128kByte
+#      if defined(HPX_DEBUG)
+#        define HPX_SMALL_STACK_SIZE_TARGET  0x20000       // 128kByte
 #      else
-#         define HPX_SMALL_STACK_SIZE_TARGET  0x10000        // 64kByte
+#        if defined(__powerpc__) || defined(__INTEL_COMPILER)
+#           define HPX_SMALL_STACK_SIZE_TARGET  0x20000       // 128kByte
+#        else
+#           define HPX_SMALL_STACK_SIZE_TARGET  0x10000        // 64kByte
+#        endif
 #      endif
 #    endif
 #  endif
-
+#
 #  if HPX_SMALL_STACK_SIZE_TARGET < (2 * HPX_THREADS_STACK_OVERHEAD)
 #    define HPX_SMALL_STACK_SIZE (2 * HPX_THREADS_STACK_OVERHEAD)
 #  else

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

@@ -1,4 +1,4 @@
-//  Copyright (c) 2007-2023 Hartmut Kaiser
+//  Copyright (c) 2007-2024 Hartmut Kaiser
 //
 //  SPDX-License-Identifier: BSL-1.0
 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -425,6 +425,12 @@ namespace hpx::threads {
             runs_as_child_mode_bits = static_cast<std::uint8_t>(bits);
         }
 
+        void schedule_hint(std::int16_t core) noexcept
+        {
+            mode = thread_schedule_hint_mode::thread;
+            hint = core;
+        }
+
         /// The hint associated with the mode. The interpretation of this hint
         /// depends on the given mode.
         std::int16_t hint = -1;

libs/core/execution/tests/unit/bulk_async.cpp

@@ -1,4 +1,5 @@
 //  Copyright (c) 2015 Daniel Bourgeois
+//  Copyright (c) 2024 Hartmut Kaiser
 //
 //  SPDX-License-Identifier: BSL-1.0
 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -16,16 +17,16 @@
 #include <vector>
 
 ////////////////////////////////////////////////////////////////////////////////
-int bulk_test(
-    hpx::thread::id tid, int value, bool is_par, int passed_through)    //-V813
+int bulk_test(hpx::thread::id const& tid, int value, bool is_par,
+    int passed_through)    //-V813
 {
     HPX_TEST_EQ(is_par, (tid != hpx::this_thread::get_id()));
     HPX_TEST_EQ(passed_through, 42);
     return value;
 }
 
 template <typename Executor>
-void test_bulk_sync(Executor& exec)
+void test_bulk_sync(Executor&& exec)
 {
     hpx::thread::id tid = hpx::this_thread::get_id();
 
@@ -35,14 +36,15 @@ void test_bulk_sync(Executor& exec)
     using hpx::placeholders::_1;
     using hpx::placeholders::_2;
 
-    std::vector<int> results = hpx::parallel::execution::bulk_sync_execute(
-        exec, hpx::bind(&bulk_test, tid, _1, false, _2), v, 42);
+    std::vector<int> results =
+        hpx::parallel::execution::bulk_sync_execute(HPX_FORWARD(Executor, exec),
+            hpx::bind(&bulk_test, tid, _1, false, _2), v, 42);
 
     HPX_TEST(std::equal(std::begin(results), std::end(results), std::begin(v)));
 }
 
 template <typename Executor>
-void test_bulk_async(Executor& exec)
+void test_bulk_async(Executor&& exec)
 {
     hpx::thread::id tid = hpx::this_thread::get_id();
 
@@ -54,31 +56,42 @@ void test_bulk_async(Executor& exec)
 
     std::vector<hpx::future<int>> results =
         hpx::parallel::execution::bulk_async_execute(
-            exec, hpx::bind(&bulk_test, tid, _1, true, _2), v, 42);
+            HPX_FORWARD(Executor, exec),
+            hpx::bind(&bulk_test, tid, _1, true, _2), v, 42);
 
     HPX_TEST(std::equal(std::begin(results), std::end(results), std::begin(v),
         [](hpx::future<int>& lhs, const int& rhs) {
             return lhs.get() == rhs;
         }));
 }
 
+template <typename Executor>
+decltype(auto) disable_run_as_child(Executor&& exec)
+{
+    auto hint = hpx::execution::experimental::get_hint(exec);
+    hint.runs_as_child_mode(hpx::threads::thread_execution_hint::none);
+
+    return hpx::experimental::prefer(hpx::execution::experimental::with_hint,
+        HPX_FORWARD(Executor, exec), hint);
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 int hpx_main()
 {
     hpx::execution::sequenced_executor seq_exec;
-    test_bulk_sync(seq_exec);
+    test_bulk_sync(disable_run_as_child(seq_exec));
 
     hpx::execution::parallel_executor par_exec;
     hpx::execution::parallel_executor par_fork_exec(hpx::launch::fork);
-    test_bulk_async(par_exec);
-    test_bulk_async(par_fork_exec);
+    test_bulk_async(disable_run_as_child(par_exec));
+    test_bulk_async(disable_run_as_child(par_fork_exec));
 
     return hpx::local::finalize();
 }
 
 int main(int argc, char* argv[])
 {
-    // By default this test should run on all available cores
+    // By default, this test should run on all available cores
     std::vector<std::string> const cfg = {"hpx.os_threads=all"};
 
     // Initialize and run HPX