diff --git a/CMakeLists.txt b/CMakeLists.txt index 3d80e7dd..7afed45c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,12 @@ option( ${PROJECT_IS_TOP_LEVEL} ) +option( + BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + "Enable building default parallel_scheduler backend implementation. Default: ON. Values: { ON, OFF }." + ON +) + add_subdirectory(src/beman/execution) if(NOT BEMAN_USE_MODULES) diff --git a/README.md b/README.md index 65302974..63a125fb 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception --> -[![Library Status](https://raw.githubusercontent.com/bemanproject/beman/refs/heads/main/images/badges/beman_badge-beman_library_under_development.svg)](https://github.com/bemanproject/beman/blob/main/docs/beman_library_maturity_model.md#the-beman-library-maturity-model)![Standard Target](https://github.com/bemanproject/beman/blob/main/images/badges/cpp26.svg)[![Coverage](https://coveralls.io/repos/github/bemanproject/execution/badge.svg?branch=main)](https://coveralls.io/github/bemanproject/execution?branch=main)[![Compiler Explorer Example](https://img.shields.io/badge/Try%20it%20on%20Compiler%20Explorer-grey?logo=compilerexplorer&logoColor=67c52a)](https://godbolt.org/z/jeMEWGYbM) +[![Library Status](https://raw.githubusercontent.com/bemanproject/beman/refs/heads/main/images/badges/beman_badge-beman_library_under_development.svg)](https://github.com/bemanproject/beman/blob/main/docs/beman_library_maturity_model.md#the-beman-library-maturity-model)![Standard Target](https://github.com/bemanproject/beman/blob/main/images/badges/cpp26.svg)![Build Status](https://github.com/bemanproject/execution/actions/workflows/ci_tests.yml/badge.svg)[![Coverage](https://coveralls.io/repos/github/bemanproject/execution/badge.svg?branch=main)](https://coveralls.io/github/bemanproject/execution?branch=main)[![Compiler Explorer Example](https://img.shields.io/badge/Try%20it%20on%20Compiler%20Explorer-grey?logo=compilerexplorer&logoColor=67c52a)](https://godbolt.org/z/jeMEWGYbM) @@ -64,10 +64,12 @@ You can disable building tests by setting CMake option `BEMAN_EXECUTION_BUILD_TE You can disable building examples by setting CMake option `BEMAN_EXECUTION_BUILD_EXAMPLES` to `OFF` when configuring the project. - -| Library | Linux | MacOS | Windows | -| ------- | ----- | ----- | ------- | -| build | ![Linux build status](https://github.com/bemanproject/execution/actions/workflows/linux.yml/badge.svg) | ![MacOS build status](https://github.com/bemanproject/execution/actions/workflows/macos.yml/badge.svg) | ![Window build status](https://github.com/bemanproject/execution/actions/workflows/windows.yml/badge.svg) | +By default, `beman.execution` +provides [query_parallel_scheduler_backend](https://eel.is/c++draft/exec.parschedrepl.query). If you want to use a +custom +`parallel_scheduler_backend`, you can disable the default `query_parallel_scheduler_backend` +implementation by setting CMake option +`BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND` to `OFF` when configuring the project. The following instructions build the library and the examples: diff --git a/include/beman/execution/detail/default_parallel_scheduler_backend.hpp b/include/beman/execution/detail/default_parallel_scheduler_backend.hpp new file mode 100644 index 00000000..ae1b321d --- /dev/null +++ b/include/beman/execution/detail/default_parallel_scheduler_backend.hpp @@ -0,0 +1,34 @@ +// include/beman/execution/detail/default_parallel_scheduler_backend.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#define INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#ifdef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.parallel_scheduler_replacement; +import beman.execution.detail.thread_pool_backend; +#else +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::parallel_scheduler_replacement { +inline auto query_parallel_scheduler_backend() -> ::std::shared_ptr { + static auto backend = ::std::make_shared<::beman::execution::detail::thread_pool_backend>(); + return backend; +} +} // namespace beman::execution::parallel_scheduler_replacement + +// ---------------------------------------------------------------------------- + +#endif // BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND diff --git a/include/beman/execution/detail/parallel_scheduler.hpp b/include/beman/execution/detail/parallel_scheduler.hpp index 2110f523..c1d2e704 100644 --- a/include/beman/execution/detail/parallel_scheduler.hpp +++ b/include/beman/execution/detail/parallel_scheduler.hpp @@ -45,6 +45,7 @@ import beman.execution.detail.stop_token_of_t; #else #include #include +#include // IWYU pragma: keep #include #include #include diff --git a/include/beman/execution/detail/parallel_scheduler_replacement.hpp b/include/beman/execution/detail/parallel_scheduler_replacement.hpp index b1951b31..27bce3a5 100644 --- a/include/beman/execution/detail/parallel_scheduler_replacement.hpp +++ b/include/beman/execution/detail/parallel_scheduler_replacement.hpp @@ -67,7 +67,6 @@ struct parallel_scheduler_backend { -> void = 0; }; -// TODO(P2079R10): provide the project-supported link-time replaceability hook. auto query_parallel_scheduler_backend() -> ::std::shared_ptr; } // namespace beman::execution::parallel_scheduler_replacement diff --git a/include/beman/execution/detail/thread_pool_backend.hpp b/include/beman/execution/detail/thread_pool_backend.hpp new file mode 100644 index 00000000..9a89af17 --- /dev/null +++ b/include/beman/execution/detail/thread_pool_backend.hpp @@ -0,0 +1,283 @@ +// include/beman/execution/detail/thread_pool_backend.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND +#define INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND + +#include +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.parallel_scheduler_replacement; +import beman.execution.detail.psched_bulk_sender; +#else +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail { +class thread_pool_backend_base + : public ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend { + protected: + struct task_base { + task_base() noexcept : next(nullptr) {} + + task_base(const task_base&) = delete; + + task_base(task_base&&) = delete; + + virtual ~task_base() = default; + + auto operator=(const task_base&) -> task_base& = delete; + + auto operator=(task_base&&) -> task_base& = delete; + + virtual auto exec() noexcept -> void = 0; + + task_base* next; + }; + + struct schedule_task : task_base { + explicit schedule_task(::beman::execution::parallel_scheduler_replacement::receiver_proxy& p) noexcept + : proxy(p) {} + + auto exec() noexcept -> void override { + auto& proxy_ref = proxy; + ::std::destroy_at(this); + proxy_ref.set_value(); + } + + ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy; + }; + + // `schedule_task` is small enough to fit directly into the pre-allocated storage provided by `schedule()` + static_assert(sizeof(schedule_task) <= psched_storage_size && alignof(schedule_task) <= psched_storage_alignment); + + struct single_bulk_task : task_base { + single_bulk_task(::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& p, + ::std::size_t shape) noexcept + : proxy(p), shape(shape) {} + + auto exec() noexcept -> void override { + proxy.execute(0uz, shape); + auto& proxy_ref = proxy; + ::std::destroy_at(this); + proxy_ref.set_value(); + } + + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy; + ::std::size_t shape; + }; + + // `single_bulk_task` is small enough to fit directly into the pre-allocated storage provided by + // `schedule_bulk_chunked()`/`schedule_bulk_unchunked()` + static_assert(sizeof(single_bulk_task) <= psched_storage_size && + alignof(single_bulk_task) <= psched_storage_alignment); + + struct batched_bulk_task : task_base { + struct cookie_type { + cookie_type(batched_bulk_task* head, ::std::size_t chunk_count) noexcept + : head(head), chunk_count(chunk_count), ref_count(chunk_count) {} + batched_bulk_task* head; + ::std::size_t chunk_count; + ::std::atomic<::std::size_t> ref_count; + }; + + batched_bulk_task(cookie_type* cookie, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::size_t i, + ::std::size_t j) noexcept + : cookie(cookie), proxy(proxy), i(i), j(j) {} + + auto exec() noexcept -> void override { + proxy.execute(i, j); + if (cookie->ref_count.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) { + auto head = cookie->head; + const auto chunk_count = cookie->chunk_count; + auto& proxy_ref = proxy; + ::std::destroy_at(cookie); + ::std::destroy_n(head, chunk_count); + ::operator delete( + head, chunk_count * sizeof(batched_bulk_task), ::std::align_val_t{alignof(batched_bulk_task)}); + proxy_ref.set_value(); + } + } + + cookie_type* cookie; + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy; + ::std::size_t i; + ::std::size_t j; + }; + + // The cookie of a batch lives in the pre-allocated storage too, so a batch costs exactly one allocation: + // the chunk array itself. + static_assert(sizeof(batched_bulk_task::cookie_type) <= psched_storage_size && + alignof(batched_bulk_task::cookie_type) <= psched_storage_alignment); + + public: + thread_pool_backend_base() = default; + + thread_pool_backend_base(const thread_pool_backend_base&) = delete; + + thread_pool_backend_base(thread_pool_backend_base&&) = delete; + + ~thread_pool_backend_base() override = default; + + auto operator=(const thread_pool_backend_base&) -> thread_pool_backend_base& = delete; + + auto operator=(thread_pool_backend_base&&) -> thread_pool_backend_base& = delete; + + auto shutdown() noexcept -> void { + ::std::unique_lock guard{mtx}; + shutdown_requested = true; + guard.unlock(); + cv.notify_all(); + } + + auto schedule(::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + push_back(::std::construct_at(reinterpret_cast(storage.data()), proxy)); // nothrow! + } + + auto schedule_bulk_chunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + const ::std::size_t chunk_length = (shape + num_threads() - 1uz) / num_threads(); + schedule_bulk(shape, chunk_length, proxy, storage); + } + + auto schedule_bulk_unchunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + schedule_bulk_chunked(shape, proxy, storage); + } + + protected: + [[nodiscard]] static auto num_threads() noexcept -> ::std::size_t { + static const ::std::size_t count = ::std::max(1u, ::std::thread::hardware_concurrency()); + return count; + } + + auto schedule_bulk(::std::size_t shape, + ::std::size_t chunk_length, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void { + if (shape == 0uz) { + schedule(proxy, storage); + return; + } + + const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; + try { + if (chunk_count == 1uz) { + push_back(::std::construct_at(reinterpret_cast(storage.data()), proxy, shape)); + } else { + auto head = static_cast(::operator new( + chunk_count * sizeof(batched_bulk_task), ::std::align_val_t{alignof(batched_bulk_task)})); + // NOLINTBEGIN(*-reinterpret-cast, *-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + auto cookie = ::std::construct_at( + reinterpret_cast(storage.data()), head, chunk_count); + + batched_bulk_task* prev = nullptr; + for (::std::size_t i = 0; i < chunk_count; ++i) { + const ::std::size_t begin = i * chunk_length; + const ::std::size_t end = ::std::min(begin + chunk_length, shape); + auto task = ::std::construct_at(head + i, cookie, proxy, begin, end); + if (prev) { + prev->next = task; + } + prev = task; + } + // NOLINTEND(*-reinterpret-cast, *-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + push_back(head, chunk_count); + } + } catch (...) { + proxy.set_error(::std::current_exception()); + } + } + + auto push_back(task_base* t, ::std::size_t n = 1uz) noexcept -> void { + ::std::unique_lock guard{mtx}; + for (::std::size_t i = 0; i < n; ++i) { + if (auto prev_back = ::std::exchange(back, t)) { + prev_back->next = t; + } else { + front = t; + } + t = t->next; + } + assert(t == nullptr); + guard.unlock(); + if (n == 1uz) { + cv.notify_one(); + } else { + cv.notify_all(); + } + } + + [[nodiscard]] auto pop_front() noexcept -> task_base* { + ::std::unique_lock guard{mtx}; + cv.wait(guard, [this] { return front != nullptr || shutdown_requested; }); + if (front == back) { + back = nullptr; + } + return front ? ::std::exchange(front, front->next) : nullptr; + } + + protected: + bool shutdown_requested = false; + ::std::mutex mtx; + ::std::condition_variable cv; + task_base* front = nullptr; + task_base* back = nullptr; +}; + +struct thread_pool_backend : ::beman::execution::detail::thread_pool_backend_base { + explicit thread_pool_backend(::std::in_place_t) : workers(::new ::std::thread[num_threads()]) {} + + thread_pool_backend() : thread_pool_backend(::std::in_place) { + for (auto& worker : ::std::span(workers.get(), num_threads())) { + worker = ::std::thread(&thread_pool_backend::run, this); + } + } + + ~thread_pool_backend() override { + shutdown(); + for (auto& worker : ::std::span(workers.get(), num_threads())) { + if (worker.joinable()) { + worker.join(); + } + } + } + + private: + auto run() noexcept -> void { + while (auto task = pop_front()) { + task->exec(); + } + } + + ::std::unique_ptr<::std::thread[]> workers; +}; + +} // namespace beman::execution::detail + +// ---------------------------------------------------------------------------- + +#endif // #ifdef INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index fe0d2430..f3eed017 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -75,6 +75,7 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/decays_to.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/default_domain.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/default_impls.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/default_parallel_scheduler_backend.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/dependent_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/dependent_sender_error.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/execution_policy.hpp @@ -222,6 +223,7 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/task.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/task_scheduler.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/then.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/thread_pool_backend.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/transform_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/try_query.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/type_list.hpp @@ -244,6 +246,13 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/write_env.hpp ) +if(BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND) + target_compile_definitions( + ${BEMAN_EXECUTION_TARGET_NAME} + INTERFACE BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + ) +endif() + if(BEMAN_USE_MODULES) target_sources( ${BEMAN_EXECUTION_TARGET_PREFIX} @@ -446,6 +455,7 @@ if(BEMAN_USE_MODULES) task.cppm task_scheduler.cppm then.cppm + thread_pool_backend.cppm transform_sender.cppm try_query.cppm type_list.cppm diff --git a/src/beman/execution/parallel_scheduler.cppm b/src/beman/execution/parallel_scheduler.cppm index 14076f84..c5380546 100644 --- a/src/beman/execution/parallel_scheduler.cppm +++ b/src/beman/execution/parallel_scheduler.cppm @@ -2,6 +2,7 @@ module; // src/beman/execution/parallel_scheduler.cppm -*-C++-*- // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception +#include // IWYU pragma: keep #include export module beman.execution.detail.parallel_scheduler; diff --git a/src/beman/execution/thread_pool_backend.cppm b/src/beman/execution/thread_pool_backend.cppm new file mode 100644 index 00000000..4ce67e23 --- /dev/null +++ b/src/beman/execution/thread_pool_backend.cppm @@ -0,0 +1,12 @@ +module; +// src/beman/execution/thread_pool_backend.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.thread_pool_backend; + +namespace beman::execution::detail { +export using beman::execution::detail::thread_pool_backend_base; +export using beman::execution::detail::thread_pool_backend; +} // namespace beman::execution::detail diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index dae8fe12..1b6c60c0 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -27,8 +27,10 @@ import std; #ifdef BEMAN_HAS_MODULES import beman.execution; import beman.execution.detail.schedule_result_t; +import beman.execution.detail.thread_pool_backend; #else #include +#include #endif namespace { @@ -57,226 +59,6 @@ struct backend : replaceability::parallel_scheduler_backend { ::std::span<::std::byte>) noexcept -> void override {} }; -struct thread_pool_base : replaceability::parallel_scheduler_backend { - struct task { - task() = default; - - task(const task&) = delete; - - task(task&&) = delete; - - virtual ~task() = default; - - auto operator=(const task&) -> task& = delete; - - auto operator=(task&&) -> task& = delete; - - virtual auto exec() noexcept -> void = 0; - }; - - struct schedule_task : task { - explicit schedule_task(replaceability::receiver_proxy& p) noexcept : proxy(p) {} - - auto exec() noexcept -> void override { proxy.set_value(); } - - replaceability::receiver_proxy& proxy; - }; - - struct bulk_shared_state { - std::atomic counter; - std::exception_ptr exception; - }; - - struct bulk_task : task { - bulk_task(std::shared_ptr counter, - replaceability::bulk_item_receiver_proxy& proxy, - std::size_t i, - std::size_t j) noexcept - : shared_state(std::move(counter)), proxy(proxy), i(i), j(j) {} - - auto exec() noexcept -> void override { - proxy.execute(i, j); - if (shared_state->counter.fetch_sub(1uz, std::memory_order_acq_rel) == 1uz) { - if (shared_state->exception) { - proxy.set_error(shared_state->exception); - } else { - proxy.set_value(); - } - } - } - - std::shared_ptr shared_state; - replaceability::bulk_item_receiver_proxy& proxy; - std::size_t i; - std::size_t j; - }; - - thread_pool_base() = default; - - thread_pool_base(const thread_pool_base&) = delete; - - ~thread_pool_base() override = 0; - - auto operator=(const thread_pool_base&) = delete; - - auto shutdown() -> void { - std::unique_lock guard{mtx}; - shutdown_requested = true; - guard.unlock(); - cv.notify_all(); - } - - auto schedule(replaceability::receiver_proxy& proxy, ::std::span<::std::byte>) noexcept -> void override { - try { - auto t = std::make_unique(proxy); - std::unique_lock guard{mtx}; - tasks.push(std::move(t)); - guard.unlock(); - cv.notify_one(); - } catch (...) { - proxy.set_error(std::current_exception()); - } - } - - auto schedule_bulk(std::size_t shape, - std::size_t chunk_length, - replaceability::bulk_item_receiver_proxy& proxy, - std::span<::std::byte> storage) noexcept -> void { - if (shape == 0uz) { - schedule(proxy, storage); - return; - } - const std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; - std::shared_ptr shared_state; - try { - shared_state = std::make_shared(chunk_count); - } catch (...) { - proxy.set_error(std::current_exception()); - return; - } - std::unique_lock guard{mtx}; - for (std::size_t i = 0; i < chunk_count; ++i) { - try { - const std::size_t begin = i * chunk_length; - const std::size_t end = std::min(begin + chunk_length, shape); - tasks.push(std::make_unique(shared_state, proxy, begin, end)); - } catch (...) { - const std::size_t n = chunk_count - i; // the count of `bulk_task` which are not enqueued successfully - - guard.unlock(); - if (i == 1uz) { - cv.notify_one(); - } else if (i > 1uz) { - cv.notify_all(); - } - - // happens-before `proxy.set_value()/proxy.set_error(...)` in `bulk_task::exec` - shared_state->exception = std::current_exception(); - - if (shared_state->counter.fetch_sub(n, std::memory_order_acq_rel) == n) { - proxy.set_error(shared_state->exception); - } - return; - } - } - guard.unlock(); - cv.notify_all(); - } - - auto schedule_bulk_chunked(::std::size_t shape, - replaceability::bulk_item_receiver_proxy& proxy, - ::std::span<::std::byte> storage) noexcept -> void override { - const std::size_t chunk_length = (shape + num_threads - 1uz) / num_threads; - schedule_bulk(shape, chunk_length, proxy, storage); - } - - auto schedule_bulk_unchunked(::std::size_t shape, - replaceability::bulk_item_receiver_proxy& proxy, - ::std::span<::std::byte> storage) noexcept -> void override { - schedule_bulk(shape, 1uz, proxy, storage); - } - - protected: - static constexpr std::size_t num_threads = 4uz; - bool shutdown_requested = false; - std::mutex mtx; - std::condition_variable cv; - std::queue> tasks; -}; - -thread_pool_base::~thread_pool_base() = default; - -struct thread_pool_backend : thread_pool_base { - thread_pool_backend() { - for (std::size_t i = 0; i < num_threads; ++i) { - workers[i] = std::thread([this]() noexcept { this->run(); }); - } - } - - ~thread_pool_backend() override { - shutdown(); - for (auto& worker : workers) { - worker.join(); - } - } - - private: - auto run() noexcept -> void { - while (true) { - std::unique_lock guard{mtx}; - cv.wait(guard, [this]() noexcept { return !tasks.empty() || shutdown_requested; }); - if (shutdown_requested && tasks.empty()) { - return; - } - auto task = std::move(tasks.front()); - tasks.pop(); - guard.unlock(); - task->exec(); - } - } - - std::thread workers[num_threads]; -}; - -// for GCC and Clang, enable -fopenmp for both compiling and linking; for MSVC, use the /openmp:llvm compiler option. -#ifdef _OPENMP -struct openmp_backend : thread_pool_base { - openmp_backend() { - designee = std::thread{[this]() noexcept { -#pragma omp parallel num_threads(num_threads) - { -#pragma omp single - { - while (true) { - std::unique_lock guard{mtx}; - cv.wait(guard, [this]() noexcept { return !tasks.empty() || shutdown_requested; }); - if (shutdown_requested && tasks.empty()) { - break; - } - auto front = std::move(tasks.front()); - tasks.pop(); - guard.unlock(); - auto front_ptr = front.release(); -#pragma omp task firstprivate(front_ptr) - { - std::unique_ptr{front_ptr}->exec(); - } - } - } - } - }}; - } - - ~openmp_backend() override { - shutdown(); - designee.join(); - } - - private: - std::thread designee; -}; -#endif - auto test_parallel_scheduler_synopsis() -> void { static_assert(!::std::default_initializable); static_assert(::std::copy_constructible); @@ -345,16 +127,14 @@ auto test_parallel_scheduler_schedule() -> void { } } // namespace +#ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND namespace beman::execution::parallel_scheduler_replacement { auto query_parallel_scheduler_backend() -> std::shared_ptr { -#ifdef _OPENMP - static auto backend = std::make_shared<::openmp_backend>(); -#else - static auto backend = std::make_shared<::thread_pool_backend>(); -#endif + static auto backend = std::make_shared<::test_detail::thread_pool_backend>(); return backend; } } // namespace beman::execution::parallel_scheduler_replacement +#endif // BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND TEST(exec_parallel_scheduler) { test_parallel_scheduler_synopsis();