diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index a349754..c66df80 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -73,4 +73,8 @@ if(BUILD_CUDA AND BUILD_TBB) add_executable(exec_reco_stdexec exec_reco.cpp) target_link_libraries(exec_reco_stdexec PRIVATE CoroutineTests CUDA::cudart TBB::tbb STDEXEC::stdexec) endif() + if(BUILD_CAPY) + add_executable(capy_reco capy_reco.cpp) + target_link_libraries(capy_reco PRIVATE CoroutineTests CUDA::cudart TBB::tbb Boost::capy) + endif() endif() diff --git a/examples/README.md b/examples/README.md index a3886e5..1d9ce3e 100644 --- a/examples/README.md +++ b/examples/README.md @@ -158,6 +158,6 @@ This is a variant of ["Exec tbb" example](#exec-tbb) using Boost.Capy and IoAwai ## Reconstruction -Link: [alien_reco.cpp](alien_reco.cpp) and [exec_reco.cpp](exec_reco.cpp) +Link: [alien_reco.cpp](alien_reco.cpp), [exec_reco.cpp](exec_reco.cpp), [capy_reco.cpp](capy_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. +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)), in `exec_reco_stdexec` the application is implemented with C++26 execution, in `capy_reco` the implementation uses Boost.Capy and IoAwaitables protocol ([p4003](https://www.open-std.org/JTC1/SC22/WG21/docs/papers/2026/p4003r0.pdf)). diff --git a/examples/capy_reco.cpp b/examples/capy_reco.cpp new file mode 100644 index 0000000..9146fa8 --- /dev/null +++ b/examples/capy_reco.cpp @@ -0,0 +1,241 @@ +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "capy_stream_await.hpp" // StreamIoAwaitable +#include "capy_task_arena_executor.hpp" // TaskArenaExecutor +#include "logging_utils.hpp" // log, format_name +#include "statuscode.hpp" // StatusCodeImpl + +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; +}; + +boost::capy::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 StreamIoAwaitable{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)}; +} + +boost::capy::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 StreamIoAwaitable{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)}; +} + +boost::capy::task reconstruct(cudaStream_t stream, + std::string 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 StreamIoAwaitable{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; + + auto task_arena = tbb::task_arena{2}; + auto context = TaskArenaContext(task_arena); + auto executor = TaskArenaExecutor(context); + + { + std::cout << "--- Single event, synchronous wait for completion ---\n"; + cudaStream_t stream; + ERROR_CHECK_CUDA(cudaStreamCreate(&stream)); + auto condition = std::condition_variable(); + auto mutex = std::mutex(); + auto final_result = tools::StatusCode{}; + auto result_handler = [&condition, &mutex, + &final_result](tools::StatusCode code) { + { + std::lock_guard lock(mutex); + final_result = code; + } + condition.notify_one(); + }; + + log() << "main Launching algorithm..." << std::endl; + boost::capy::run_async(executor, + result_handler)(reconstruct(stream, "main")); + { + auto lock = std::unique_lock(mutex); + condition.wait(lock, [&final_result]() { + return final_result != tools::StatusCode::UNDEFINED; + }); + } + log() << "main Final status of algorithm " << final_result << "" + << 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 condition = std::condition_variable(); + auto mutex = std::mutex(); + auto counter = streams.size(); + log() << "main Launching algorithms..." << std::endl; + + for (std::size_t i = 0; i < streams.size(); ++i) { + auto result_handler = [&condition, &mutex, &status, &counter, + i](tools::StatusCode code) { + auto done = false; + { + std::lock_guard lock(mutex); + status.at(i) = code; + done = (--counter == 0); + } + if (done) { + condition.notify_one(); + } + }; + boost::capy::run_async(executor, result_handler)( + reconstruct(streams.at(i), "event" + std::to_string(i))); + } + { + auto lock = std::unique_lock(mutex); + condition.wait(lock, [&counter]() { return counter == 0; }); + } + 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/capy_stream_await.hpp b/examples/capy_stream_await.hpp new file mode 100644 index 0000000..ea9cf30 --- /dev/null +++ b/examples/capy_stream_await.hpp @@ -0,0 +1,39 @@ +#pragma once +#include + +#include +#include + +class StreamIoAwaitable { + public: + StreamIoAwaitable(cudaStream_t stream) : m_stream(stream) {} + + bool await_ready() const noexcept { return false; } + + void await_suspend(std::coroutine_handle<> handle, + boost::capy::io_env const* env) noexcept { + m_context.handle = handle; + m_context.env = env; + m_error = cudaLaunchHostFunc(m_stream, resumption_callback, &m_context); + // If the callback couldn't be registered, we need to reschedule the + // coroutine immediately to avoid deadlock. + if (m_error != cudaSuccess) { + resumption_callback(&m_context); + } + } + cudaError_t await_resume() const noexcept { return m_error; } + + private: + struct context { + std::coroutine_handle<> handle; + boost::capy::io_env const* env; + }; + cudaStream_t m_stream; + cudaError_t m_error = cudaSuccess; + context m_context; + + static void resumption_callback(void* userData) { + auto* ctx = static_cast(userData); + ctx->env->executor.post(ctx->handle); + } +};