executors: fix performance regressions and optimize parallel_executor

Author: arpittkhandelwal

libs/core/executors/include/hpx/executors/fork_join_executor.hpp

@@ -1270,8 +1270,7 @@ namespace hpx::execution::experimental {
         /// \brief Construct a fork_join_executor.
         ///
         /// \param priority  The priority of the worker threads.
-        /// \param stacksize The stacksize of the worker threads. Must not be
-        ///                  nostack.
+        /// \param stacksize The stacksize of the worker threads.
         /// \param sched     The loop schedule of the parallel regions.
         /// \param yield_delay The time after which the executor yields to other
         ///        work if it has not received any new work for execution.
@@ -1291,8 +1290,7 @@ namespace hpx::execution::experimental {
         ///
         /// \param pu_mask   The PU-mask to use for placing the created threads
         /// \param priority  The priority of the worker threads.
-        /// \param stacksize The stacksize of the worker threads. Must not be
-        ///                  nostack.
+        /// \param stacksize The stacksize of the worker threads.
         /// \param sched     The loop schedule of the parallel regions.
         /// \param yield_delay The time after which the executor yields to other
         ///        work if it has not received any new work for execution.
@@ -1347,6 +1345,23 @@ namespace hpx::execution::experimental {
             return exec.shared_data_->pu_mask_;
         }
 
+        friend std::size_t tag_invoke(
+            hpx::execution::experimental::get_first_core_t,
+            fork_join_executor const& exec) noexcept
+        {
+            return shared_data::get_first_core(exec.shared_data_->pu_mask_);
+        }
+
+        template <typename Parameters>
+        friend std::size_t tag_invoke(
+            hpx::execution::experimental::processing_units_count_t,
+            Parameters&&, fork_join_executor const& exec,
+            hpx::chrono::steady_duration const& = hpx::chrono::null_duration,
+            std::size_t = 0) noexcept
+        {
+            return exec.shared_data_->num_threads_;
+        }
+
         /// \cond NOINTERNAL
         enum class init_mode : std::uint8_t
         {

libs/core/executors/include/hpx/executors/parallel_executor.hpp

@@ -161,7 +161,6 @@ namespace hpx::execution {
           , policy_(rhs.policy_)
           , first_core_(rhs.first_core_)
           , num_cores_(rhs.num_cores_)
-          , mask_(nullptr)    // force recomputing cached pu mask
 #if defined(HPX_HAVE_THREAD_DESCRIPTION)
           , annotation_(rhs.annotation_)
 #endif
@@ -178,7 +177,6 @@ namespace hpx::execution {
                 policy_ = rhs.policy_;
                 first_core_ = rhs.first_core_;
                 num_cores_ = rhs.num_cores_;
-                mask_ = nullptr;    // force recomputing cached pu mask
 
 #if defined(HPX_HAVE_THREAD_DESCRIPTION)
                 annotation_ = rhs.annotation_;
@@ -187,10 +185,7 @@ namespace hpx::execution {
             return *this;
         }
 
-        constexpr ~parallel_policy_executor_base()
-        {
-            delete mask_;
-        }
+        constexpr ~parallel_policy_executor_base() = default;
 
         // backwards compatibility support, will be removed in the future
         template <typename Parameters>
@@ -345,11 +340,6 @@ namespace hpx::execution {
 
         hpx::threads::mask_type pu_mask() const
         {
-            //if (mask_ != nullptr && hpx::threads::any(*mask_))
-            //{
-            //    return *mask_;
-            //}
-
             auto const num_threads = get_num_cores();
             auto const available_threads = static_cast<std::uint32_t>(
                 pool()->get_active_os_thread_count());
@@ -376,13 +366,6 @@ namespace hpx::execution {
                 }
             }
 
-            // `mask_` is conceptually mutable, however in order to constexpr
-            // construct this object we need to use a `const_cast` to cache
-            // the mask.
-
-            //*const_cast<hpx::threads::mask_type**>(&mask_) =
-            //    new hpx::threads::mask_type(mask);
-
             return mask;
         }
 
@@ -397,7 +380,6 @@ namespace hpx::execution {
         Policy policy_;
         std::size_t first_core_ = 0;
         std::size_t num_cores_ = 0;
-        hpx::threads::mask_type* mask_ = nullptr;
 #if defined(HPX_HAVE_THREAD_DESCRIPTION)
         char const* annotation_ = nullptr;
 #endif

libs/core/executors/tests/unit/fork_join_executor.cpp

@@ -49,6 +49,16 @@ void test_processing_mask(ExecutorArgs&&... args)
 
     auto const cores_mask = hpx::execution::experimental::get_cores_mask(exec);
     HPX_TEST(cores_mask == expected_mask);
+
+    auto const expected_pu_count = hpx::threads::count(pus_mask);
+    HPX_TEST_EQ(expected_pu_count, hpx::threads::count(expected_mask));
+
+    auto const pu_count =
+        hpx::execution::experimental::processing_units_count(exec);
+    HPX_TEST_EQ(pu_count, expected_pu_count);
+
+    auto const first_core = hpx::execution::experimental::get_first_core(exec);
+    HPX_TEST_EQ(first_core, hpx::threads::find_first(expected_mask));
 }
 
 ///////////////////////////////////////////////////////////////////////////////

libs/core/executors/tests/unit/parallel_executor.cpp

@@ -187,21 +187,43 @@ void test_processing_mask()
 {
     hpx::execution::parallel_executor exec;
 
+    auto const pool = hpx::threads::detail::get_self_or_default_pool();
+    auto const expected_mask =
+        pool->get_used_processing_units(pool->get_os_thread_count(), false);
+
+    // 1. Initial access returns the expected processing units mask
+    auto const mask1 =
+        hpx::execution::experimental::get_processing_units_mask(exec);
+    HPX_TEST(mask1 == expected_mask);
+
+    // 2. Repeated access returns the same mask
+    auto const mask2 =
+        hpx::execution::experimental::get_processing_units_mask(exec);
+    HPX_TEST(mask2 == expected_mask);
+
+    // 3. Copy construction preserves the observable mask behavior
+    hpx::execution::parallel_executor exec_copy = exec;
+    auto const mask3 =
+        hpx::execution::experimental::get_processing_units_mask(exec_copy);
+    HPX_TEST(mask3 == expected_mask);
+
+    // 4. Copy assignment preserves the observable mask behavior
+    hpx::execution::parallel_executor exec_assign;
+    // Exercise the assignee before assignment to verify repeated use remains correct
+    hpx::execution::experimental::get_processing_units_mask(exec_assign);
+
+    // Assign from another executor and verify the resulting mask is still correct
+    exec_assign = exec;
+    auto const mask4 =
+        hpx::execution::experimental::get_processing_units_mask(exec_assign);
+    HPX_TEST(mask4 == expected_mask);
+
+    // 5. Cores mask test
     {
-        auto const pool = hpx::threads::detail::get_self_or_default_pool();
-        auto const expected_mask =
-            pool->get_used_processing_units(pool->get_os_thread_count(), false);
-        auto const mask =
-            hpx::execution::experimental::get_processing_units_mask(exec);
-        HPX_TEST(mask == expected_mask);
-    }
-
-    {
-        auto const pool = hpx::threads::detail::get_self_or_default_pool();
-        auto const expected_mask =
+        auto const expected_cores =
             pool->get_used_processing_units(pool->get_os_thread_count(), true);
         auto const mask = hpx::execution::experimental::get_cores_mask(exec);
-        HPX_TEST(mask == expected_mask);
+        HPX_TEST(mask == expected_cores);
     }
 }