Skip to content

Commit eaa324f

Browse files
m-filam-fila
authored
add delegation example (#54)
* add delegation example with stdexec * add delegation example with capy * bump capy version, the old had a bug in changing executors inside coroutine * update examples readme * simplify result handler with latch
1 parent 33e12d8 commit eaa324f

5 files changed

Lines changed: 530 additions & 1 deletion

File tree

‎examples/CMakeLists.txt‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,9 +72,13 @@ if(BUILD_CUDA AND BUILD_TBB)
7272
if(BUILD_STDEXEC)
7373
add_executable(exec_reco_stdexec exec_reco.cpp)
7474
target_link_libraries(exec_reco_stdexec PRIVATE CoroutineTests CUDA::cudart TBB::tbb STDEXEC::stdexec)
75+
add_executable(exec_delegate_stdexec exec_delegate.cpp)
76+
target_link_libraries(exec_delegate_stdexec PRIVATE CoroutineTests CUDA::cudart TBB::tbb STDEXEC::stdexec)
7577
endif()
7678
if(BUILD_CAPY)
7779
add_executable(capy_reco capy_reco.cpp)
7880
target_link_libraries(capy_reco PRIVATE CoroutineTests CUDA::cudart TBB::tbb Boost::capy)
81+
add_executable(capy_delegate capy_delegate.cpp)
82+
target_link_libraries(capy_delegate PRIVATE CoroutineTests CUDA::cudart TBB::tbb Boost::capy)
7983
endif()
8084
endif()

‎examples/README.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,3 +161,9 @@ This is a variant of ["Exec tbb" example](#exec-tbb) using Boost.Capy and IoAwai
161161
Link: [alien_reco.cpp](alien_reco.cpp), [exec_reco.cpp](exec_reco.cpp), [capy_reco.cpp](capy_reco.cpp)
162162

163163
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)).
164+
165+
## Delegation
166+
167+
Link: [exec_delegation.cpp](exec_delegation.cpp), [capy_delegation.cpp](capy_delegation.cpp)
168+
169+
These examples are a modification of [reconstuction examples](#reconstruction) to delegate all the calls to CUDA API that are happening inside the task to execute on a designated thread by changing scheduler/executor. This represents a specific optimization: invoking CUDA APIs from multiple threads can incur performance penalties due to contention on internal CUDA locks. By delegating these calls to a single thread, this contention can be reduced. The examples demonstrate how this optimization can be implemented using coroutines schedulers/executors.

‎examples/capy_delegate.cpp‎

Lines changed: 256 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,256 @@
1+
#include <cuda_runtime_api.h>
2+
#include <tbb/task_arena.h>
3+
4+
#include <boost/capy.hpp>
5+
#include <boost/capy/ex/thread_pool.hpp>
6+
#include <cstddef>
7+
#include <iostream>
8+
#include <latch>
9+
#include <string_view>
10+
#include <utility>
11+
12+
#include "capy_stream_await.hpp" // StreamIoAwaitable
13+
#include "capy_task_arena_executor.hpp" // TaskArenaExecutor
14+
#include "logging_utils.hpp" // log, format_name
15+
#include "statuscode.hpp" // StatusCodeImpl
16+
17+
namespace tools {
18+
struct Tag {
19+
static constexpr const char* name = "tools";
20+
};
21+
using StatusCode = StatusCodeImpl<Tag>;
22+
} // namespace tools
23+
24+
#define ERROR_CHECK_CUDA(EXP) \
25+
do { \
26+
cudaError_t errorCode = EXP; \
27+
if (errorCode != cudaSuccess) { \
28+
throw std::runtime_error( \
29+
std::format("CUDA error at {}:{}: {}", __FILE__, __LINE__, \
30+
cudaGetErrorString(errorCode))); \
31+
} \
32+
} while (false)
33+
34+
template <typename T>
35+
struct DeviceBuffer {
36+
T* ptr = nullptr;
37+
std::size_t size = 0;
38+
};
39+
40+
template <typename F>
41+
boost::capy::task<void> coro_wrapper(F&& f) {
42+
std::forward<F>(f)();
43+
co_return;
44+
}
45+
46+
boost::capy::task<DeviceBuffer<int>> clusterization(
47+
DeviceBuffer<int> cells, cudaStream_t stream,
48+
boost::capy::thread_pool& delegation_thread, std::string_view parent) {
49+
50+
const auto self = format_name(parent, "clusterization");
51+
log(self) << "Starting clusterization" << std::endl;
52+
53+
const auto nCells = static_cast<int>(cells.size);
54+
55+
// Copy cells back to host to count non-zero entries
56+
auto h_cells = std::vector<int>(nCells);
57+
58+
co_await boost::capy::run(
59+
delegation_thread.get_executor())(coro_wrapper([&]() {
60+
log(self) << "Delegated copy of cells from device to host" << std::endl;
61+
ERROR_CHECK_CUDA(cudaMemcpyAsync(h_cells.data(), cells.ptr,
62+
nCells * sizeof(int),
63+
cudaMemcpyDeviceToHost, stream));
64+
}));
65+
66+
ERROR_CHECK_CUDA(co_await StreamIoAwaitable{stream});
67+
68+
auto nClusters = 0;
69+
for (auto v : h_cells)
70+
if (v != 0)
71+
++nClusters;
72+
73+
log(self) << "Found " << nClusters << " clusters" << std::endl;
74+
75+
// Allocate clusters of appropiate size on device
76+
int* d_clusters = nullptr;
77+
78+
co_await boost::capy::run(
79+
delegation_thread.get_executor())(coro_wrapper([&]() {
80+
log(self) << "Delegated allocation of clusters on device" << std::endl;
81+
ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast<void**>(&d_clusters),
82+
nClusters * sizeof(int), stream));
83+
84+
// Write some dummy data to the clusters buffer to simulate work
85+
ERROR_CHECK_CUDA(
86+
cudaMemsetAsync(d_clusters, 0, nClusters * sizeof(int), stream));
87+
ERROR_CHECK_CUDA(cudaMemsetAsync(d_clusters, 1,
88+
nClusters / 2 * sizeof(int), stream));
89+
}));
90+
91+
co_return DeviceBuffer<int>{d_clusters,
92+
static_cast<std::size_t>(nClusters)};
93+
}
94+
95+
boost::capy::task<DeviceBuffer<int>> seeding(
96+
DeviceBuffer<int> clusters, cudaStream_t stream,
97+
boost::capy::thread_pool& delegation_thread, std::string_view parent) {
98+
99+
const auto self = format_name(parent, "seeding");
100+
log(self) << "Starting seeding" << std::endl;
101+
102+
const auto nClusters = static_cast<int>(clusters.size);
103+
104+
// Copy clusters to host to count non-zero entries
105+
auto h_clusters = std::vector<int>(nClusters);
106+
107+
co_await boost::capy::run(delegation_thread.get_executor())(
108+
coro_wrapper([&]() {
109+
log(self) << "Delegated copy of clusters to host" << std::endl;
110+
ERROR_CHECK_CUDA(cudaMemcpyAsync(h_clusters.data(), clusters.ptr,
111+
nClusters * sizeof(int),
112+
cudaMemcpyDeviceToHost, stream));
113+
}));
114+
115+
ERROR_CHECK_CUDA(co_await StreamIoAwaitable{stream});
116+
117+
int nSeeds = 0;
118+
for (auto v : h_clusters)
119+
if (v != 0)
120+
++nSeeds;
121+
122+
log(self) << "Found " << nSeeds << " seeds" << std::endl;
123+
124+
// Allocate seeds of appropiate size on device
125+
int* d_seeds = nullptr;
126+
127+
co_await boost::capy::run(delegation_thread.get_executor())(
128+
coro_wrapper([&]() {
129+
log(self) << "Delegated allocation of seeds on device" << std::endl;
130+
ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast<void**>(&d_seeds),
131+
nSeeds * sizeof(int), stream));
132+
133+
// Write some dummy data to the seeds buffer to simulate work
134+
ERROR_CHECK_CUDA(
135+
cudaMemsetAsync(d_seeds, 0, nSeeds * sizeof(int), stream));
136+
ERROR_CHECK_CUDA(
137+
cudaMemsetAsync(d_seeds, 1, nSeeds / 2 * sizeof(int), stream));
138+
}));
139+
140+
co_return DeviceBuffer<int>{d_seeds, static_cast<std::size_t>(nSeeds)};
141+
}
142+
143+
boost::capy::task<tools::StatusCode> reconstruct(
144+
cudaStream_t stream, boost::capy::thread_pool& delegation_thread,
145+
std::string parent) {
146+
const auto self = format_name(parent, "reconstruction");
147+
log(self) << "Starting reconstruction" << std::endl;
148+
149+
// Allocate some dummy input data on the device
150+
auto cells = DeviceBuffer<int>{nullptr, 1000};
151+
co_await boost::capy::run(
152+
delegation_thread.get_executor())(coro_wrapper([&]() {
153+
log(self) << "Delegated allocation of input data on device"
154+
<< std::endl;
155+
ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast<void**>(&cells.ptr),
156+
cells.size * sizeof(int), stream));
157+
ERROR_CHECK_CUDA(
158+
cudaMemsetAsync(cells.ptr, 1, cells.size * sizeof(int), stream));
159+
}));
160+
161+
// Run the clusterization and seeding steps
162+
auto clusters =
163+
co_await clusterization(cells, stream, delegation_thread, self);
164+
auto seeds = co_await seeding(clusters, stream, delegation_thread, self);
165+
166+
// Cleanup
167+
co_await boost::capy::run(delegation_thread.get_executor())(
168+
coro_wrapper([&]() {
169+
log(self) << "Delegated cleanup of device memory" << std::endl;
170+
ERROR_CHECK_CUDA(cudaFreeAsync(cells.ptr, stream));
171+
ERROR_CHECK_CUDA(cudaFreeAsync(clusters.ptr, stream));
172+
ERROR_CHECK_CUDA(cudaFreeAsync(seeds.ptr, stream));
173+
}));
174+
175+
ERROR_CHECK_CUDA(co_await StreamIoAwaitable{stream});
176+
177+
log(self) << "Finishing reconstruction" << std::endl;
178+
co_return tools::StatusCode::SUCCESS;
179+
}
180+
181+
int main() {
182+
int deviceCount = 0;
183+
auto error_id = cudaGetDeviceCount(&deviceCount);
184+
185+
if (error_id != cudaSuccess) {
186+
std::cout << "cudaGetDeviceCount returned "
187+
<< static_cast<int>(error_id) << "\n"
188+
<< cudaGetErrorString(error_id) << "\n";
189+
return EXIT_FAILURE;
190+
}
191+
192+
if (deviceCount == 0) {
193+
std::cout << "No CUDA devices found.\n";
194+
return EXIT_FAILURE;
195+
}
196+
197+
log() << "main Starting" << std::endl;
198+
199+
auto task_arena = tbb::task_arena{2, 0};
200+
auto context = TaskArenaContext(task_arena);
201+
auto executor = TaskArenaExecutor(context);
202+
auto delegation_thread = boost::capy::thread_pool(1);
203+
204+
{
205+
std::cout << "--- Single event, synchronous wait for completion ---\n";
206+
cudaStream_t stream;
207+
ERROR_CHECK_CUDA(cudaStreamCreate(&stream));
208+
auto final_result = tools::StatusCode{};
209+
auto done = std::latch{1};
210+
auto result_handler = [&done, &final_result](tools::StatusCode code) {
211+
final_result = code;
212+
done.count_down();
213+
};
214+
215+
log() << "main Launching algorithm..." << std::endl;
216+
boost::capy::run_async(executor, result_handler)(
217+
reconstruct(stream, delegation_thread, "main"));
218+
done.wait();
219+
log() << "main Final status of algorithm " << final_result << ""
220+
<< std::endl;
221+
ERROR_CHECK_CUDA(cudaStreamDestroy(stream));
222+
}
223+
224+
{
225+
std::cout << "--- Multiple events, wait for all to complete ---\n ";
226+
227+
auto streams = std::vector<cudaStream_t>(2);
228+
auto status = std::vector<tools::StatusCode>(streams.size());
229+
for (auto& stream : streams) {
230+
ERROR_CHECK_CUDA(cudaStreamCreate(&stream));
231+
}
232+
233+
auto done = std::latch{static_cast<std::ptrdiff_t>(streams.size())};
234+
log() << "main Launching algorithms..." << std::endl;
235+
236+
for (std::size_t i = 0; i < streams.size(); ++i) {
237+
auto result_handler = [&done, &status, i](tools::StatusCode code) {
238+
status.at(i) = code;
239+
done.count_down();
240+
};
241+
boost::capy::run_async(executor, result_handler)(reconstruct(
242+
streams.at(i), delegation_thread, "event" + std::to_string(i)));
243+
}
244+
done.wait();
245+
for (auto& stream : streams) {
246+
ERROR_CHECK_CUDA(cudaStreamDestroy(stream));
247+
}
248+
249+
log() << "main All algorithms completed. Final statuses: ";
250+
for (auto s : status) {
251+
std::cout << s << ' ';
252+
}
253+
std::cout << std::endl;
254+
}
255+
return 0;
256+
}

0 commit comments

Comments
 (0)