diff --git a/CMakeLists.txt b/CMakeLists.txt index 106be98..a1aec6a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2,6 +2,7 @@ cmake_minimum_required(VERSION 3.25) project(CoroutineTests) set(CMAKE_CXX_STANDARD 20 CACHE STRING "") +set(CMAKE_CXX_EXTENSIONS FALSE CACHE BOOL "Disable C++ extensions" ) set(CMAKE_CXX_STANDARD_REQUIRED ON) if(CMAKE_CXX_STANDARD LESS 20) message(FATAL_ERROR "Unsupported C++ standard: ${CMAKE_CXX_STANDARD}. Must be at least C++20.") @@ -81,6 +82,13 @@ if(BUILD_TBB) endif() endif() +### CUDA options + +option(BUILD_CUDA "Build with CUDA support" OFF) +if(BUILD_CUDA) + find_package(CUDAToolkit REQUIRED) +endif() + ## targets add_library(CoroutineTests INTERFACE) diff --git a/CMakePresets.json b/CMakePresets.json index 6d00f48..ea4e89d 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -12,7 +12,8 @@ "BUILD_STDEXEC": "ON", "BUILD_TBB": "ON", "BUILD_EXAMPLES": "ON", - "BUILD_PERF": "ON" + "BUILD_PERF": "ON", + "BUILD_CUDA": "OFF" } }, { @@ -26,6 +27,16 @@ "BUILD_BEMAN_TASK": "OFF" } }, + { + "name": "20-cuda", + "displayName": "Configure preset constrained to C++20 with CUDA support", + "inherits": [ + "20" + ], + "cacheVariables": { + "BUILD_CUDA": "ON" + } + }, { "name": "23", "displayName": "Configure preset constrained to C++23", @@ -39,7 +50,8 @@ } ], "buildPresets": [ - { "name": "default", + { + "name": "default", "displayName": "Default Build Preset", "hidden": true, "configurePreset": "default", @@ -53,6 +65,14 @@ "default" ] }, + { + "name": "20-cuda", + "displayName": "Build with C++20 and CUDA support", + "configurePreset": "20-cuda", + "inherits": [ + "20" + ] + }, { "name": "23", "displayName": "Build with C++23", @@ -76,6 +96,19 @@ } ] }, + { + "name": "20-cuda", + "steps": [ + { + "type": "configure", + "name": "20-cuda" + }, + { + "type": "build", + "name": "20-cuda" + } + ] + }, { "name": "23", "steps": [ diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 58615c4..788a397 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -55,3 +55,12 @@ if(BUILD_BEMAN_TASK) target_compile_definitions(exec_tbb_beman PUBLIC "USE_BEMAN") endif() endif() + +if(BUILD_CUDA AND BUILD_TBB) + add_executable(alien_reco alien_reco.cpp) + target_link_libraries(alien_reco PRIVATE CoroutineTests CUDA::cudart TBB::tbb) + if(BUILD_STDEXEC) + add_executable(exec_reco_stdexec exec_reco.cpp) + target_link_libraries(exec_reco_stdexec PRIVATE CoroutineTests CUDA::cudart TBB::tbb STDEXEC::stdexec) + endif() +endif() diff --git a/examples/README.md b/examples/README.md index 9e5d01b..b39b05c 100644 --- a/examples/README.md +++ b/examples/README.md @@ -143,3 +143,9 @@ This example demonstrates a `sync_wait` algorithm compatible with coroutine sema Link: [alien_counting_scope.cpp](alien_counting_scope.cpp) This example demonstrates dynamic work submitting with `counting_scope` compatible with coroutine semantic as in ["Alien" example](#alien). `spawn` schedules execution of a coroutine that returns `void`. `join()` blocks the current thread until all submitted coroutines are finished (and rethrows the first exception captured from submitted work, if any). + +## Reconstruction + +Link: [alien_reco.cpp](alien_reco.cpp) and [exec_reco.cpp](exec_reco.cpp) + +These examples show a setup and coroutine chain loosely inspired by track reconstruction on GPU in high energy physics experiments. The `reconstruct` coroutine prepares a mockup input data on a CUDA device, then `co_await`s the `clustering` coroutine, then receives output from it and `co_await`s the `seeding` coroutine. Both `clustering` and `seeding` coroutines receive a device buffer, copy it asynchronously back to host, suspend until the copy is done, then count non-zero elements and allocate new buffer of that size for their results. In `main` the `reconstruct` coroutines are executed in a TBB task arena either synchronously waiting for the result from the submitting or dynamically starting a few `reconstruct` coroutines and waiting until all the work is finished. In the `alien_reco` the application is implemented with "alien" coroutines (as in [alien examples](#alien)) , while in `exec_reco_stdexec` the application is implemented with C++26 execution. diff --git a/examples/alien_reco.cpp b/examples/alien_reco.cpp new file mode 100644 index 0000000..4d76f64 --- /dev/null +++ b/examples/alien_reco.cpp @@ -0,0 +1,237 @@ +#include +#include + +#include +#include + +#include "CoroutineTests/alien/counting_scope.hpp" +#include "CoroutineTests/alien/subtool.hpp" +#include "CoroutineTests/alien/sync_wait.hpp" +#include "CoroutineTests/alien/tool.hpp" +#include "logging_utils.hpp" // log, format_name + +#define ERROR_CHECK_CUDA(EXP) \ + do { \ + cudaError_t errorCode = EXP; \ + if (errorCode != cudaSuccess) { \ + throw std::runtime_error( \ + std::format("CUDA error at {}:{}: {}", __FILE__, __LINE__, \ + cudaGetErrorString(errorCode))); \ + } \ + } while (false) + +using namespace CoroutineTests::alien; + +/// Awaitable that resumes a coroutine when a CUDA stream reaches a certain +/// point. Internally cudaLaunchHostFunc is used to set up a resumption callback +/// on the stream. +class StreamAwaitable { + public: + StreamAwaitable(cudaStream_t stream) : m_stream(stream) {} + + bool await_ready() const noexcept { return false; } + template + void await_suspend(std::coroutine_handle handle) { + m_error = cudaLaunchHostFunc(m_stream, resumption_callback, + handle.address()); + // If the callback couldn't be registered, we need to reschedule the + // coroutine immediately to avoid deadlock. + if (m_error != cudaSuccess) { + handle.promise().reschedule(); + } + } + cudaError_t await_resume() const noexcept { return m_error; } + + private: + cudaStream_t m_stream; + cudaError_t m_error = cudaSuccess; + + template + static void resumption_callback(void* userData) { + auto handle = std::coroutine_handle::from_address(userData); + handle.promise().reschedule(); + } +}; + +template +struct DeviceBuffer { + T* ptr = nullptr; + std::size_t size = 0; +}; + +subtool::Task> clusterization(DeviceBuffer cells, + cudaStream_t stream, + std::string_view parent) { + + const auto self = format_name(parent, "clusterization"); + log(self) << "Starting clusterization" << std::endl; + + const auto nCells = static_cast(cells.size); + + // Copy cells back to host to count non-zero entries + auto h_cells = std::vector(nCells); + + ERROR_CHECK_CUDA(cudaMemcpyAsync(h_cells.data(), cells.ptr, + nCells * sizeof(int), + cudaMemcpyDeviceToHost, stream)); + + ERROR_CHECK_CUDA(co_await StreamAwaitable{stream}); + + auto nClusters = 0; + for (auto v : h_cells) + if (v != 0) + ++nClusters; + + log(self) << "Found " << nClusters << " clusters" << std::endl; + + // Allocate clusters of appropiate size on device + int* d_clusters = nullptr; + ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast(&d_clusters), + nClusters * sizeof(int), stream)); + + // Write some dummy data to the clusters buffer to simulate work + ERROR_CHECK_CUDA( + cudaMemsetAsync(d_clusters, 0, nClusters * sizeof(int), stream)); + ERROR_CHECK_CUDA( + cudaMemsetAsync(d_clusters, 1, nClusters / 2 * sizeof(int), stream)); + + co_return DeviceBuffer{d_clusters, + static_cast(nClusters)}; +} + +subtool::Task> seeding(DeviceBuffer clusters, + cudaStream_t stream, + std::string_view parent) { + + const auto self = format_name(parent, "seeding"); + log(self) << "Starting seeding" << std::endl; + + const auto nClusters = static_cast(clusters.size); + + // Copy clusters to host to count non-zero entries + auto h_clusters = std::vector(nClusters); + + ERROR_CHECK_CUDA(cudaMemcpyAsync(h_clusters.data(), clusters.ptr, + nClusters * sizeof(int), + cudaMemcpyDeviceToHost, stream)); + + ERROR_CHECK_CUDA(co_await StreamAwaitable{stream}); + + int nSeeds = 0; + for (auto v : h_clusters) + if (v != 0) + ++nSeeds; + + log(self) << "Found " << nSeeds << " seeds" << std::endl; + + // Allocate clusters of appropiate size on device + int* d_seeds = nullptr; + ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast(&d_seeds), + nSeeds * sizeof(int), stream)); + + // Write some dummy data to the clusters buffer to simulate work + ERROR_CHECK_CUDA(cudaMemsetAsync(d_seeds, 0, nSeeds * sizeof(int), stream)); + ERROR_CHECK_CUDA( + cudaMemsetAsync(d_seeds, 1, nSeeds / 2 * sizeof(int), stream)); + + co_return DeviceBuffer{d_seeds, static_cast(nSeeds)}; +} + +tool::Task reconstruct(cudaStream_t stream, + std::string_view parent) { + const auto self = format_name(parent, "reconstruction"); + log(self) << "Starting reconstruction" << std::endl; + + // Allocate some dummy input data on the device + auto cells = DeviceBuffer{nullptr, 1000}; + ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast(&cells.ptr), + cells.size * sizeof(int), stream)); + ERROR_CHECK_CUDA( + cudaMemsetAsync(cells.ptr, 1, cells.size * sizeof(int), stream)); + + // Run the clusterization and seeding steps + auto clusters = co_await clusterization(cells, stream, self); + auto seeds = co_await seeding(clusters, stream, self); + + // Cleanup + ERROR_CHECK_CUDA(cudaFreeAsync(cells.ptr, stream)); + ERROR_CHECK_CUDA(cudaFreeAsync(clusters.ptr, stream)); + ERROR_CHECK_CUDA(cudaFreeAsync(seeds.ptr, stream)); + + ERROR_CHECK_CUDA(co_await StreamAwaitable{stream}); + + log(self) << "Finishing reconstruction" << std::endl; + co_return tool::StatusCode::SUCCESS; +} + +int main() { + int deviceCount = 0; + auto error_id = cudaGetDeviceCount(&deviceCount); + + if (error_id != cudaSuccess) { + std::cout << "cudaGetDeviceCount returned " + << static_cast(error_id) << "\n" + << cudaGetErrorString(error_id) << "\n"; + return EXIT_FAILURE; + } + + if (deviceCount == 0) { + std::cout << "No CUDA devices found.\n"; + return EXIT_FAILURE; + } + + log() << "main Starting" << std::endl; + + tbb::task_arena task_arena{2}; + + auto scheduler = [&task_arena](std::coroutine_handle<> handle) { + task_arena.enqueue([handle]() { handle.resume(); }); + }; + + { + cudaStream_t stream; + ERROR_CHECK_CUDA(cudaStreamCreate(&stream)); + std::cout << "--- Single event, synchronous wait for completion ---\n"; + log() << "main Launching algorithm..." << std::endl; + auto status = sync_wait(scheduler, reconstruct(stream, "main")); + log() << "main Final status of algorithm " << status << "" << std::endl; + ERROR_CHECK_CUDA(cudaStreamDestroy(stream)); + } + + { + std::cout << "--- Multiple events, wait for all to complete ---\n"; + + auto streams = std::vector(2); + auto status = std::vector(streams.size()); + for (auto& stream : streams) { + ERROR_CHECK_CUDA(cudaStreamCreate(&stream)); + } + + auto scope = counting_scope{}; + log() << "main Launching algorithms..." << std::endl; + + auto payload = [](std::vector streams, + std::vector& statuses, + int i) -> tool::Task { + const auto name = std::format("event{}:main", i); + auto& stream = streams.at(i); + auto& status = statuses.at(i); + status = co_await reconstruct(stream, name); + }; + for (std::size_t i = 0; i < streams.size(); ++i) { + scope.spawn(scheduler, payload(streams, status, i)); + } + scope.join(); + + for (auto& stream : streams) { + ERROR_CHECK_CUDA(cudaStreamDestroy(stream)); + } + + log() << "main All algorithms completed. Final statuses: "; + for (auto s : status) { + std::cout << s << ' '; + } + std::cout << std::endl; + } + return 0; +} diff --git a/examples/exec_reco.cpp b/examples/exec_reco.cpp new file mode 100644 index 0000000..9219ae1 --- /dev/null +++ b/examples/exec_reco.cpp @@ -0,0 +1,214 @@ +#include +#include + +#include +#include +#include +#include +#include + +#include "exec_statuscode.hpp" // StatusCodeImpl +#include "exec_stream_await_sender.hpp" // stream_await_sender +#include "exec_task_arena_scheduler.hpp" // TaskArenaScheduler +#include "logging_utils.hpp" // log, format_name + +namespace tools { +struct Tag { + static constexpr const char* name = "tools"; +}; +using StatusCode = StatusCodeImpl; +} // namespace tools + +#define ERROR_CHECK_CUDA(EXP) \ + do { \ + cudaError_t errorCode = EXP; \ + if (errorCode != cudaSuccess) { \ + throw std::runtime_error( \ + std::format("CUDA error at {}:{}: {}", __FILE__, __LINE__, \ + cudaGetErrorString(errorCode))); \ + } \ + } while (false) + +template +struct DeviceBuffer { + T* ptr = nullptr; + std::size_t size = 0; +}; + +exec::task> clusterization(DeviceBuffer cells, + cudaStream_t stream, + std::string_view parent) { + + const auto self = format_name(parent, "clusterization"); + log(self) << "Starting clusterization" << std::endl; + + const auto nCells = static_cast(cells.size); + + // Copy cells back to host to count non-zero entries + auto h_cells = std::vector(nCells); + + ERROR_CHECK_CUDA(cudaMemcpyAsync(h_cells.data(), cells.ptr, + nCells * sizeof(int), + cudaMemcpyDeviceToHost, stream)); + + ERROR_CHECK_CUDA(co_await stream_await_sender{stream}); + + auto nClusters = 0; + for (auto v : h_cells) + if (v != 0) + ++nClusters; + + log(self) << "Found " << nClusters << " clusters" << std::endl; + + // Allocate clusters of appropiate size on device + int* d_clusters = nullptr; + ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast(&d_clusters), + nClusters * sizeof(int), stream)); + + // Write some dummy data to the clusters buffer to simulate work + ERROR_CHECK_CUDA( + cudaMemsetAsync(d_clusters, 0, nClusters * sizeof(int), stream)); + ERROR_CHECK_CUDA( + cudaMemsetAsync(d_clusters, 1, nClusters / 2 * sizeof(int), stream)); + + co_return DeviceBuffer{d_clusters, + static_cast(nClusters)}; +} + +exec::task> seeding(DeviceBuffer clusters, + cudaStream_t stream, + std::string_view parent) { + + const auto self = format_name(parent, "seeding"); + log(self) << "Starting seeding" << std::endl; + + const auto nClusters = static_cast(clusters.size); + + // Copy clusters to host to count non-zero entries + auto h_clusters = std::vector(nClusters); + + ERROR_CHECK_CUDA(cudaMemcpyAsync(h_clusters.data(), clusters.ptr, + nClusters * sizeof(int), + cudaMemcpyDeviceToHost, stream)); + + ERROR_CHECK_CUDA(co_await stream_await_sender{stream}); + + int nSeeds = 0; + for (auto v : h_clusters) + if (v != 0) + ++nSeeds; + + log(self) << "Found " << nSeeds << " seeds" << std::endl; + + // Allocate clusters of appropiate size on device + int* d_seeds = nullptr; + ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast(&d_seeds), + nSeeds * sizeof(int), stream)); + + // Write some dummy data to the clusters buffer to simulate work + ERROR_CHECK_CUDA(cudaMemsetAsync(d_seeds, 0, nSeeds * sizeof(int), stream)); + ERROR_CHECK_CUDA( + cudaMemsetAsync(d_seeds, 1, nSeeds / 2 * sizeof(int), stream)); + + co_return DeviceBuffer{d_seeds, static_cast(nSeeds)}; +} + +exec::task reconstruct(cudaStream_t stream, + std::string_view parent) { + const auto self = format_name(parent, "reconstruction"); + log(self) << "Starting reconstruction" << std::endl; + + // Allocate some dummy input data on the device + auto cells = DeviceBuffer{nullptr, 1000}; + ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast(&cells.ptr), + cells.size * sizeof(int), stream)); + ERROR_CHECK_CUDA( + cudaMemsetAsync(cells.ptr, 1, cells.size * sizeof(int), stream)); + + // Run the clusterization and seeding steps + auto clusters = co_await clusterization(cells, stream, self); + auto seeds = co_await seeding(clusters, stream, self); + + // Cleanup + ERROR_CHECK_CUDA(cudaFreeAsync(cells.ptr, stream)); + ERROR_CHECK_CUDA(cudaFreeAsync(clusters.ptr, stream)); + ERROR_CHECK_CUDA(cudaFreeAsync(seeds.ptr, stream)); + + ERROR_CHECK_CUDA(co_await stream_await_sender{stream}); + + log(self) << "Finishing reconstruction" << std::endl; + co_return tools::StatusCode::SUCCESS; +} + +int main() { + int deviceCount = 0; + auto error_id = cudaGetDeviceCount(&deviceCount); + + if (error_id != cudaSuccess) { + std::cout << "cudaGetDeviceCount returned " + << static_cast(error_id) << "\n" + << cudaGetErrorString(error_id) << "\n"; + return EXIT_FAILURE; + } + + if (deviceCount == 0) { + std::cout << "No CUDA devices found.\n"; + return EXIT_FAILURE; + } + + log() << "main Starting" << std::endl; + + tbb::task_arena task_arena{2}; + execution::scheduler auto scheduler = get_scheduler(task_arena, false); + + { + cudaStream_t stream; + ERROR_CHECK_CUDA(cudaStreamCreate(&stream)); + std::cout << "--- Single event, synchronous wait for completion ---\n"; + log() << "main Launching algorithm..." << std::endl; + auto [status] = + stdexec::sync_wait( + stdexec::starts_on(scheduler, reconstruct(stream, "main"))) + .value(); + log() << "main Final status of algorithm " << status << "" << std::endl; + ERROR_CHECK_CUDA(cudaStreamDestroy(stream)); + } + + { + std::cout << "--- Multiple events, wait for all to complete ---\n"; + + auto streams = std::vector(2); + auto status = std::vector(streams.size()); + for (auto& stream : streams) { + ERROR_CHECK_CUDA(cudaStreamCreate(&stream)); + } + + auto scope = exec::async_scope{}; + log() << "main Launching algorithms..." << std::endl; + + auto payload = [](std::vector streams, + std::vector& statuses, + int i) -> exec::task { + const auto name = std::format("event{}:", i); + auto& stream = streams.at(i); + auto& status = statuses.at(i); + status = co_await reconstruct(stream, name); + }; + for (std::size_t i = 0; i < streams.size(); ++i) { + scope.spawn( + stdexec::starts_on(scheduler, payload(streams, status, i))); + } + stdexec::sync_wait(scope.on_empty()); + + for (auto& stream : streams) { + ERROR_CHECK_CUDA(cudaStreamDestroy(stream)); + } + + log() << "main All algorithms completed. Final statuses: "; + for (auto s : status) { + std::cout << s << ' '; + } + std::cout << std::endl; + } + return 0; +} diff --git a/examples/exec_statuscode.hpp b/examples/exec_statuscode.hpp index 69152ef..e144c0f 100644 --- a/examples/exec_statuscode.hpp +++ b/examples/exec_statuscode.hpp @@ -8,7 +8,7 @@ class StatusCodeImpl { inline static const StatusCodeImpl SUCCESS{Status::SUCCESS}; inline static const StatusCodeImpl FAILURE{Status::FAILURE}; StatusCodeImpl(Status status = Status::UNDEFINED) : m_status(status) {} - StatusCodeImpl(const StatusCodeImpl&) = default; + Status status() const { return m_status; } private: diff --git a/examples/exec_stream_await_sender.hpp b/examples/exec_stream_await_sender.hpp new file mode 100644 index 0000000..76136e0 --- /dev/null +++ b/examples/exec_stream_await_sender.hpp @@ -0,0 +1,63 @@ +#pragma once + +#include +#include + +#include +/// Wrapper sender suspending execution until all operations on a CUDA +/// stream are complete. + +class stream_await_sender { + public: + // associated operation state + template + class stream_await_operation; + + using sender_concept = stdexec::sender_t; + using completion_signatures = + stdexec::completion_signatures; + + stream_await_sender(const cudaStream_t stream) : m_stream(stream) {} + stdexec::env<> get_env() const noexcept { return {}; } + + template + auto connect(Receiver&& receiver) const { + return stream_await_operation>( + std::forward(receiver), m_stream); + } + + private: + cudaStream_t m_stream; +}; + +/// Operation state associated with @c stream_await_sender +/// +template +class stream_await_sender::stream_await_operation { + public: + using operation_state_concept = stdexec::operation_state_t; + + stream_await_operation(Receiver&& recv, const cudaStream_t stream) + : m_receiver(std::forward(recv)), m_stream(stream) {} + + void start() & noexcept { + + auto err = cudaLaunchHostFunc(m_stream, callback, &m_receiver); + // If setting up the callback failed, we need to propage the error + // immediately + if (err != cudaSuccess) { + stdexec::set_value(std::move(m_receiver), err); + } + } + + private: + std::remove_cvref_t m_receiver; + cudaStream_t m_stream; + + static void callback(void* userData) { + auto& recv = *static_cast(userData); + stdexec::set_value(std::move(recv), cudaSuccess); + } +}; + +static_assert(stdexec::sender); diff --git a/examples/exec_task_arena_scheduler.hpp b/examples/exec_task_arena_scheduler.hpp new file mode 100644 index 0000000..2750564 --- /dev/null +++ b/examples/exec_task_arena_scheduler.hpp @@ -0,0 +1,69 @@ +#pragma once + +#include + +#include "exec_backend.hpp" // std exec backend selection +#include "logging_utils.hpp" // log, format_name + +// Scheduler enqueuing work into a TBB task arena +struct TaskArenaScheduler { + tbb::task_arena* arena = nullptr; + bool log_scheduling = true; + using scheduler_concept = execution::scheduler_t; + + struct Env { + tbb::task_arena* arena; + bool log_scheduling; + template + auto query( + const execution::get_completion_scheduler_t&) const noexcept { + return TaskArenaScheduler{arena, log_scheduling}; + } + }; + + template + struct Operation { + std::remove_cvref_t receiver; + tbb::task_arena* arena; + bool log_scheduling; + using operation_state_concept = execution::operation_state_t; + + void start() & noexcept { + if (log_scheduling) { + log() << "Submitting work to task arena" << std::endl; + } + arena->enqueue([this]() { + if (log_scheduling) { + log() << "Running work item in task arena" << std::endl; + } + execution::set_value(std::move(receiver)); + }); + } + }; + + struct Sender { + tbb::task_arena* arena; + bool log_scheduling; + using sender_concept = execution::sender_t; + using completion_signatures = + execution::completion_signatures; + + template + auto connect(Receiver&& receiver) { + return Operation(std::forward(receiver), arena, + log_scheduling); + } + + auto get_env() const noexcept { return Env{arena, log_scheduling}; } + }; + + auto schedule() const noexcept { return Sender{arena, log_scheduling}; } + bool operator==(const TaskArenaScheduler& other) const = default; +}; + +static_assert(execution::scheduler, + "TaskArenaScheduler should model scheduler"); + +TaskArenaScheduler get_scheduler(tbb::task_arena& arena, bool log = true) { + return TaskArenaScheduler{&arena, log}; +} diff --git a/examples/exec_tbb.cpp b/examples/exec_tbb.cpp index 94cecd8..191ec4e 100644 --- a/examples/exec_tbb.cpp +++ b/examples/exec_tbb.cpp @@ -6,10 +6,11 @@ #include #include -#include "exec_backend.hpp" // std exec backend selection -#include "exec_statuscode.hpp" // exec StatusCodeImpl -#include "exec_timer.hpp" // TimerSender -#include "logging_utils.hpp" // log, format_name +#include "exec_backend.hpp" // std exec backend selection +#include "exec_statuscode.hpp" // exec StatusCodeImpl +#include "exec_task_arena_scheduler.hpp" // TaskArenaScheduler/// TaskArenaSchduler +#include "exec_timer.hpp" // TimerSender +#include "logging_utils.hpp" // log, format_name namespace tools { struct Tag { @@ -25,60 +26,6 @@ struct Tag { using StatusCode = StatusCodeImpl; } // namespace algs -// Scheduler enqueuing work into a TBB task arena -struct TaskArenaScheduler { - tbb::task_arena* arena = nullptr; - using scheduler_concept = execution::scheduler_t; - - struct Env { - tbb::task_arena* arena; - template - auto query( - const execution::get_completion_scheduler_t&) const noexcept { - return TaskArenaScheduler{arena}; - } - }; - - template - struct Operation { - std::remove_cvref_t receiver; - tbb::task_arena* arena; - using operation_state_concept = execution::operation_state_t; - - void start() & noexcept { - log() << "Submitting work to task arena" << std::endl; - arena->enqueue([this]() { - log() << "Running work item in task arena" << std::endl; - execution::set_value(std::move(receiver)); - }); - } - }; - - struct Sender { - tbb::task_arena* arena; - using sender_concept = execution::sender_t; - using completion_signatures = - execution::completion_signatures; - - template - auto connect(Receiver&& receiver) { - return Operation(std::forward(receiver), arena); - } - - auto get_env() const noexcept { return Env{arena}; } - }; - - auto schedule() const noexcept { return Sender{arena}; } - bool operator==(const TaskArenaScheduler& other) const = default; -}; - -static_assert(execution::scheduler, - "TaskArenaScheduler should model scheduler"); - -TaskArenaScheduler get_scheduler(tbb::task_arena& arena) { - return TaskArenaScheduler{&arena}; -} - execution::task tool1_execute(std::string_view parent) { const auto self = format_name(parent, "tool1");