Skip to content

Commit 96ef496

Browse files
m-filam-fila
authored
add cuda examples using query instead of callback (#60)
* add polling example with alien coroutine * add polling example with capy * update readme * fix format
1 parent e80fac8 commit 96ef496

4 files changed

Lines changed: 595 additions & 1 deletion

File tree

‎examples/CMakeLists.txt‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,8 @@ if(BUILD_CUDA AND BUILD_TBB)
7272
target_link_libraries(alien_reco PRIVATE CoroutineTests CUDA::cudart TBB::tbb)
7373
add_executable(alien_delegate alien_delegate.cpp)
7474
target_link_libraries(alien_delegate PRIVATE CoroutineTests CUDA::cudart TBB::tbb)
75+
add_executable(alien_event_poll alien_event_poll.cpp)
76+
target_link_libraries(alien_event_poll PRIVATE CoroutineTests CUDA::cudart TBB::tbb)
7577
if(BUILD_STDEXEC)
7678
add_executable(exec_reco_stdexec exec_reco.cpp)
7779
target_link_libraries(exec_reco_stdexec PRIVATE CoroutineTests CUDA::cudart TBB::tbb STDEXEC::stdexec)
@@ -83,5 +85,7 @@ if(BUILD_CUDA AND BUILD_TBB)
8385
target_link_libraries(capy_reco PRIVATE CoroutineTests CUDA::cudart TBB::tbb Boost::capy)
8486
add_executable(capy_delegate capy_delegate.cpp)
8587
target_link_libraries(capy_delegate PRIVATE CoroutineTests CUDA::cudart TBB::tbb Boost::capy)
88+
add_executable(capy_event_poll capy_event_poll.cpp)
89+
target_link_libraries(capy_event_poll PRIVATE CoroutineTests CUDA::cudart TBB::tbb Boost::capy)
8690
endif()
8791
endif()

‎examples/README.md‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,4 +172,10 @@ These examples show a setup and coroutine chain loosely inspired by track recons
172172

173173
Link: [alien_delegate.cpp](alien_delegate.cpp), [exec_delegate.cpp](exec_delegate.cpp), [capy_delegate.cpp](capy_delegate.cpp)
174174

175-
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. This contention can be reduced by delegating the calls to a single thread. The examples demonstrate how this optimization can be implemented using coroutines schedulers/executors.
175+
These examples are a modification of [reconstruction 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. This contention can be reduced by delegating the calls to a single thread. The examples demonstrate how this optimization can be implemented using coroutines schedulers/executors.
176+
177+
## Event poll
178+
179+
Link: [alien_event_poll.cpp](alien_event_poll.cpp), [capy_event_poll.cpp](capy_event_poll.cpp)
180+
181+
These examples are a variant of [delegate examples](#delegate) in which awaiting completion of CUDA operations is done by repeatedly querying the event state instead of using a callback as in earlier examples. All the queries are executed by a specific thread.

‎examples/alien_event_poll.cpp‎

Lines changed: 303 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,303 @@
1+
#include <cuda_runtime_api.h>
2+
#include <driver_types.h>
3+
#include <tbb/task_arena.h>
4+
5+
#include <cstddef>
6+
#include <format>
7+
#include <iostream>
8+
#include <stdexcept>
9+
#include <string_view>
10+
#include <vector>
11+
12+
#include "CoroutineTests/alien/counting_scope.hpp"
13+
#include "CoroutineTests/alien/schedule_on.hpp"
14+
#include "CoroutineTests/alien/subtool.hpp"
15+
#include "CoroutineTests/alien/sync_wait.hpp"
16+
#include "CoroutineTests/alien/tool.hpp"
17+
#include "CoroutineTests/threadpool.hpp"
18+
#include "logging_utils.hpp" // log, format_name
19+
20+
#define ERROR_CHECK_CUDA(EXP) \
21+
do { \
22+
cudaError_t errorCode = EXP; \
23+
if (errorCode != cudaSuccess) { \
24+
throw std::runtime_error( \
25+
std::format("CUDA error at {}:{}: {}", __FILE__, __LINE__, \
26+
cudaGetErrorString(errorCode))); \
27+
} \
28+
} while (false)
29+
30+
using namespace CoroutineTests::alien;
31+
32+
template <typename T>
33+
struct DeviceBuffer {
34+
T* ptr = nullptr;
35+
std::size_t size = 0;
36+
};
37+
38+
template <typename F>
39+
tool::Task<void> delegate(F f) {
40+
f();
41+
co_return;
42+
}
43+
44+
struct Retry {
45+
bool await_ready() const noexcept { return false; }
46+
template <typename Promise>
47+
void await_suspend(std::coroutine_handle<Promise> handle) const {
48+
handle.promise().reschedule();
49+
}
50+
void await_resume() const noexcept {}
51+
};
52+
53+
tool::Task<void> poll(cudaEvent_t event) {
54+
auto status = cudaSuccess;
55+
log() << "Polling for event completion..." << std::endl;
56+
while ((status = cudaEventQuery(event)) == cudaErrorNotReady) {
57+
log() << "Event not ready, retrying..." << std::endl;
58+
co_await Retry{};
59+
}
60+
ERROR_CHECK_CUDA(status);
61+
}
62+
63+
subtool::Task<DeviceBuffer<int>> clusterization(
64+
DeviceBuffer<int> cells, cudaStream_t stream, cudaEvent_t event,
65+
std::function<void(std::coroutine_handle<>)> delegation_scheduler,
66+
std::string_view parent) {
67+
68+
const auto self = format_name(parent, "clusterization");
69+
log(self) << "Starting clusterization" << std::endl;
70+
71+
const auto nCells = static_cast<int>(cells.size);
72+
73+
// Copy cells back to host to count non-zero entries
74+
auto h_cells = std::vector<int>(nCells);
75+
76+
co_await schedule_on(
77+
delegation_scheduler, delegate([&]() {
78+
log(self) << "Delegated copy of cells from device to host"
79+
<< std::endl;
80+
ERROR_CHECK_CUDA(cudaMemcpyAsync(h_cells.data(), cells.ptr,
81+
nCells * sizeof(int),
82+
cudaMemcpyDeviceToHost, stream));
83+
ERROR_CHECK_CUDA(cudaEventRecord(event, stream));
84+
}));
85+
86+
co_await schedule_on(delegation_scheduler, poll(event));
87+
88+
auto nClusters = 0;
89+
for (auto v : h_cells)
90+
if (v != 0)
91+
++nClusters;
92+
93+
log(self) << "Found " << nClusters << " clusters" << std::endl;
94+
95+
// Allocate clusters of appropriate size on device
96+
int* d_clusters = nullptr;
97+
98+
co_await schedule_on(
99+
delegation_scheduler, delegate([&]() {
100+
log(self) << "Delegated allocation of clusters on device"
101+
<< std::endl;
102+
ERROR_CHECK_CUDA(
103+
cudaMallocAsync(reinterpret_cast<void**>(&d_clusters),
104+
nClusters * sizeof(int), stream));
105+
106+
// Write some dummy data to the clusters buffer to simulate work
107+
ERROR_CHECK_CUDA(cudaMemsetAsync(d_clusters, 0,
108+
nClusters * sizeof(int), stream));
109+
ERROR_CHECK_CUDA(cudaMemsetAsync(
110+
d_clusters, 1, nClusters / 2 * sizeof(int), stream));
111+
}));
112+
113+
co_return DeviceBuffer<int>{d_clusters,
114+
static_cast<std::size_t>(nClusters)};
115+
}
116+
117+
subtool::Task<DeviceBuffer<int>> seeding(
118+
DeviceBuffer<int> clusters, cudaStream_t stream, cudaEvent_t event,
119+
std::function<void(std::coroutine_handle<>)> delegation_scheduler,
120+
std::string_view parent) {
121+
122+
const auto self = format_name(parent, "seeding");
123+
log(self) << "Starting seeding" << std::endl;
124+
125+
const auto nClusters = static_cast<int>(clusters.size);
126+
127+
// Copy clusters to host to count non-zero entries
128+
auto h_clusters = std::vector<int>(nClusters);
129+
130+
co_await schedule_on(
131+
delegation_scheduler, delegate([&]() {
132+
log(self) << "Delegated copy of clusters to host" << std::endl;
133+
ERROR_CHECK_CUDA(cudaMemcpyAsync(h_clusters.data(), clusters.ptr,
134+
nClusters * sizeof(int),
135+
cudaMemcpyDeviceToHost, stream));
136+
ERROR_CHECK_CUDA(cudaEventRecord(event, stream));
137+
}));
138+
139+
co_await schedule_on(delegation_scheduler, poll(event));
140+
141+
int nSeeds = 0;
142+
for (auto v : h_clusters)
143+
if (v != 0)
144+
++nSeeds;
145+
146+
log(self) << "Found " << nSeeds << " seeds" << std::endl;
147+
148+
// Allocate seeds of appropriate size on device
149+
int* d_seeds = nullptr;
150+
151+
co_await schedule_on(
152+
delegation_scheduler, delegate([&]() {
153+
log(self) << "Delegated allocation of seeds on device" << std::endl;
154+
ERROR_CHECK_CUDA(cudaMallocAsync(reinterpret_cast<void**>(&d_seeds),
155+
nSeeds * sizeof(int), stream));
156+
157+
// Write some dummy data to the seeds buffer to simulate work
158+
ERROR_CHECK_CUDA(
159+
cudaMemsetAsync(d_seeds, 0, nSeeds * sizeof(int), stream));
160+
ERROR_CHECK_CUDA(
161+
cudaMemsetAsync(d_seeds, 1, nSeeds / 2 * sizeof(int), stream));
162+
}));
163+
164+
co_return DeviceBuffer<int>{d_seeds, static_cast<std::size_t>(nSeeds)};
165+
}
166+
167+
tool::Task<tool::StatusCode> reconstruct(
168+
cudaStream_t stream, cudaEvent_t event,
169+
std::function<void(std::coroutine_handle<>)> delegation_scheduler,
170+
std::string_view parent) {
171+
const auto self = format_name(parent, "reconstruction");
172+
log(self) << "Starting reconstruction" << std::endl;
173+
174+
// Allocate some dummy input data on the device
175+
auto cells = DeviceBuffer<int>{nullptr, 1000};
176+
co_await schedule_on(
177+
delegation_scheduler, delegate([&]() {
178+
log(self) << "Delegated allocation of input data on device"
179+
<< std::endl;
180+
ERROR_CHECK_CUDA(
181+
cudaMallocAsync(reinterpret_cast<void**>(&cells.ptr),
182+
cells.size * sizeof(int), stream));
183+
ERROR_CHECK_CUDA(cudaMemsetAsync(cells.ptr, 1,
184+
cells.size * sizeof(int), stream));
185+
}));
186+
187+
// Run the clusterization and seeding steps
188+
auto clusters = co_await clusterization(cells, stream, event,
189+
delegation_scheduler, self);
190+
auto seeds =
191+
co_await seeding(clusters, stream, event, delegation_scheduler, self);
192+
193+
// Cleanup
194+
co_await schedule_on(
195+
delegation_scheduler, delegate([&]() {
196+
log(self) << "Delegated cleanup of device memory" << std::endl;
197+
ERROR_CHECK_CUDA(cudaFreeAsync(cells.ptr, stream));
198+
ERROR_CHECK_CUDA(cudaFreeAsync(clusters.ptr, stream));
199+
ERROR_CHECK_CUDA(cudaFreeAsync(seeds.ptr, stream));
200+
ERROR_CHECK_CUDA(cudaEventRecord(event, stream));
201+
}));
202+
203+
co_await schedule_on(delegation_scheduler, poll(event));
204+
205+
log(self) << "Finishing reconstruction" << std::endl;
206+
co_return tool::StatusCode::SUCCESS;
207+
}
208+
209+
int main() {
210+
int deviceCount = 0;
211+
auto error_id = cudaGetDeviceCount(&deviceCount);
212+
213+
if (error_id != cudaSuccess) {
214+
std::cout << "cudaGetDeviceCount returned "
215+
<< static_cast<int>(error_id) << "\n"
216+
<< cudaGetErrorString(error_id) << "\n";
217+
return EXIT_FAILURE;
218+
}
219+
220+
if (deviceCount == 0) {
221+
std::cout << "No CUDA devices found.\n";
222+
return EXIT_FAILURE;
223+
}
224+
225+
log() << "main Starting" << std::endl;
226+
227+
tbb::task_arena task_arena{2, 0};
228+
auto scheduler = [&task_arena](std::coroutine_handle<> handle) {
229+
task_arena.enqueue([handle]() { handle.resume(); });
230+
};
231+
232+
CoroutineTests::Threadpool delegation_threadpool(1);
233+
auto delegation_scheduler =
234+
[&delegation_threadpool](std::coroutine_handle<> handle) {
235+
delegation_threadpool.enqueue_task(handle);
236+
};
237+
238+
{
239+
cudaStream_t stream;
240+
ERROR_CHECK_CUDA(cudaStreamCreate(&stream));
241+
cudaEvent_t event;
242+
ERROR_CHECK_CUDA(
243+
cudaEventCreateWithFlags(&event, cudaEventDisableTiming));
244+
std::cout << "--- Single event, synchronous wait for completion ---\n";
245+
log() << "main Launching algorithm..." << std::endl;
246+
auto status =
247+
sync_wait(scheduler,
248+
reconstruct(stream, event, delegation_scheduler, "main"));
249+
log() << "main Final status of algorithm " << status << "" << std::endl;
250+
ERROR_CHECK_CUDA(cudaEventDestroy(event));
251+
ERROR_CHECK_CUDA(cudaStreamDestroy(stream));
252+
}
253+
254+
{
255+
std::cout << "--- Multiple events, wait for all to complete ---\n";
256+
257+
auto streams = std::vector<cudaStream_t>(2);
258+
auto events = std::vector<cudaEvent_t>(streams.size());
259+
auto status = std::vector<tool::StatusCode>(streams.size());
260+
for (std::size_t i = 0; i < streams.size(); ++i) {
261+
ERROR_CHECK_CUDA(cudaStreamCreate(&streams.at(i)));
262+
ERROR_CHECK_CUDA(cudaEventCreateWithFlags(&events.at(i),
263+
cudaEventDisableTiming));
264+
}
265+
266+
auto scope = counting_scope{};
267+
log() << "main Launching algorithms..." << std::endl;
268+
269+
auto payload = [](std::vector<cudaStream_t> streams,
270+
std::vector<cudaEvent_t> events,
271+
std::vector<tool::StatusCode>& statuses,
272+
std::function<void(std::coroutine_handle<>)>
273+
delegation_scheduler,
274+
int i) -> tool::Task<void> {
275+
const auto name = std::format("event{}:main", i);
276+
auto& stream = streams.at(i);
277+
auto& event = events.at(i);
278+
auto& status = statuses.at(i);
279+
status = co_await reconstruct(
280+
stream, event, std::move(delegation_scheduler), name);
281+
co_return;
282+
};
283+
284+
for (std::size_t i = 0; i < streams.size(); ++i) {
285+
scope.spawn(scheduler,
286+
payload(streams, events, status, delegation_scheduler,
287+
static_cast<int>(i)));
288+
}
289+
scope.join();
290+
291+
for (std::size_t i = 0; i < streams.size(); ++i) {
292+
ERROR_CHECK_CUDA(cudaEventDestroy(events.at(i)));
293+
ERROR_CHECK_CUDA(cudaStreamDestroy(streams.at(i)));
294+
}
295+
296+
log() << "main All algorithms completed. Final statuses: ";
297+
for (auto s : status) {
298+
std::cout << s << ' ';
299+
}
300+
std::cout << std::endl;
301+
}
302+
return 0;
303+
}

0 commit comments

Comments
 (0)