kokkos-execution 0.0.1
Loading...
Searching...
No Matches
test_when_all.cpp
Go to the documentation of this file.
2PRAGMA_DIAGNOSTIC_PUSH
4#include "exec/single_thread_context.hpp"
5PRAGMA_DIAGNOSTIC_POP
6
9
11
22
34
35using host_execution_space = Kokkos::DefaultHostExecutionSpace;
36
38
39using namespace Kokkos::utils::callbacks;
40
54
67TEST_F(WhenAllTest, single_branch) {
68 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
69
70 const context_t esc{exec};
71
72 auto sndr = stdexec::when_all(stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT(data));
73
74 static_assert(std::same_as<
75 decltype(stdexec::get_completion_domain<stdexec::set_value_t>(stdexec::get_env(sndr))),
77 >);
78
79 static_assert(std::same_as<stdexec::tag_of_t<decltype(sndr)>, stdexec::then_t>);
80
83
84 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
85
86 ASSERT_THAT(
88 testing::ElementsAre(
91
92 ASSERT_EQ(data(), 1);
93}
94
105TEST_F(WhenAllTest, schedule_sender_and_single_branch) {
106 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
107
108 const context_t esc{exec};
109
110 auto sndr = stdexec::when_all(
111 stdexec::schedule(esc.get_scheduler()), stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT(data));
112
113 static_assert(std::same_as<
114 decltype(stdexec::get_completion_domain<stdexec::set_value_t>(stdexec::get_env(sndr))),
116 >);
117
121
122 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
123
124 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(std::move(sndr));
125
128 ASSERT_THAT(recorded_events, [&]() {
130 return testing::ElementsAre(
133 MATCHER_FOR_WAIT_EVENT(recorded_events.at(1)));
134 } else {
135 return testing::ElementsAre(
137 MATCHER_FOR_BEGIN_FENCE(exec, dispatch_label(exec, "after dispatch")));
138 }
139 }());
140
141 ASSERT_EQ(data(), 1);
142}
143
155TEST_F(WhenAllTest, schedule_sender_and_single_branch_followed_by_self) {
156 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
157
158 const context_t esc{exec};
159
160 auto sndr = stdexec::when_all(
161 stdexec::schedule(esc.get_scheduler()),
162 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT(data))
163 | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
164
165 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
166
167 ASSERT_THAT(
169 testing::ElementsAre(
173
174 ASSERT_EQ(data(), 2);
175}
176
189TEST_F(WhenAllTest, schedule_sender_and_single_mixed_branch_followed_by_self) {
190 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
191
192 const context_t esc{exec};
193 experimental::execution::single_thread_context stc{};
194
195 auto sndr = stdexec::when_all(
196 stdexec::schedule(stc.get_scheduler()),
197 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT(data)
198 | stdexec::continues_on(stc.get_scheduler()) | THEN_INCREMENT(data))
199 | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
200
201 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
202
204
205 ASSERT_THAT(
207 testing::ElementsAre(
212
213 ASSERT_EQ(data(), 3);
214}
215
227TEST_F(WhenAllTest, schedule_sender_and_single_branch_followed_by_other_and_finish_on_self) {
228 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
229
230 const context_t esc{exec};
231 experimental::execution::single_thread_context stc{};
232
233 auto sndr = stdexec::when_all(
234 stdexec::schedule(esc.get_scheduler()),
235 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT(data))
236 | stdexec::continues_on(stc.get_scheduler()) | THEN_INCREMENT(data)
237 | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
238
239 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
240
242
243 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(std::move(sndr));
244
245 ASSERT_THAT(recorded_events, [&]() {
247 return testing::ElementsAre(
250 MATCHER_FOR_WAIT_EVENT(recorded_events.at(1)),
253 } else {
254 return testing::ElementsAre(
256 MATCHER_FOR_BEGIN_FENCE(exec, dispatch_label(exec, "after dispatch")),
259 }
260 }());
261
262 ASSERT_EQ(data(), 3);
263}
264
275TEST_F(WhenAllTest, two_mixed_branches_followed_by_self) {
276 const view_s_t data(Kokkos::view_alloc("data - shared space"));
277
278 const context_t esc{exec};
279 experimental::execution::single_thread_context stc{};
280
281 auto w_a = stdexec::when_all(
282 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data),
283 stdexec::schedule(stc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data)
284 | stdexec::continues_on(esc.get_scheduler()));
285
286 static_assert(
287 std::same_as<
288 decltype(stdexec::get_completion_domain<stdexec::set_value_t>(stdexec::get_env(w_a), stdexec::env<>{})),
290 >);
291
292 auto sndr = std::move(w_a) | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
293
294 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
295
297
298 ASSERT_THAT(
300 testing::ElementsAre(
304
305 ASSERT_EQ(data(), 3);
306}
307
318TEST_F(WhenAllTest, two_branches_followed_by_self) {
319 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
320
321 const auto [exec_A, exec_B] = Kokkos::Experimental::partition_space(exec, 1, 1);
322
323 const context_t esc_A{exec_A}, esc_B{exec_B};
324
325 auto sndr = stdexec::when_all(
326 stdexec::schedule(esc_A.get_scheduler()) | THEN_INCREMENT_ATOMIC(Device, data),
327 stdexec::schedule(esc_B.get_scheduler()) | THEN_INCREMENT_ATOMIC(Device, data))
328 | stdexec::continues_on(esc_A.get_scheduler()) | THEN_INCREMENT(data);
329
330 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
331
332 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(std::move(sndr));
333
334 if (Tests::Utils::are_same_instances(exec_A, exec_B)) {
335 ASSERT_THAT(
336 recorded_events,
337 testing::ElementsAre(
338 MATCHER_FOR_BEGIN_SINGLE(exec_A, dispatch_label(exec_A, "then")),
339 MATCHER_FOR_BEGIN_SINGLE(exec_B, dispatch_label(exec_B, "then")),
340 MATCHER_FOR_BEGIN_SINGLE(exec_A, dispatch_label(exec_A, "then")),
341 MATCHER_FOR_BEGIN_FENCE(exec_A, dispatch_label(exec_A, "sync_wait"))));
342 } else {
343 ASSERT_THAT(recorded_events, [&]() {
345 return testing::ElementsAre(
349 MATCHER_FOR_WAIT_EVENT(recorded_events.at(2)),
351 MATCHER_FOR_BEGIN_FENCE(exec_A, dispatch_label(exec, "sync_wait")));
352
353 } else {
354 return testing::ElementsAre(
355 MATCHER_FOR_BEGIN_SINGLE(exec_A, dispatch_label(exec_A, "then")),
356 MATCHER_FOR_BEGIN_SINGLE(exec_B, dispatch_label(exec_B, "then")),
357 MATCHER_FOR_BEGIN_FENCE(exec_B, dispatch_label(exec_B, "after dispatch")),
358 MATCHER_FOR_BEGIN_SINGLE(exec_A, dispatch_label(exec_A, "then")),
359 MATCHER_FOR_BEGIN_FENCE(exec_A, dispatch_label(exec_A, "sync_wait")));
360 }
361 }());
362 }
363
364 ASSERT_EQ(data(), 3);
365}
366
373TEST_F(WhenAllTest, two_branches_with_schedule_senders_only) {
374 const context_t esc{exec};
375
376 auto sndr = stdexec::when_all(stdexec::schedule(esc.get_scheduler()), stdexec::schedule(esc.get_scheduler()));
377
378 static_assert(std::same_as<stdexec::tag_of_t<decltype(sndr)>, stdexec::when_all_t>);
379 static_assert(std::same_as<
380 stdexec::__completion_domain_of_t<stdexec::set_value_t, decltype(sndr)>,
382 >);
383
384 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(
385 std::move(sndr)); // NOLINT(performance-move-const-arg)
386
387 ASSERT_THAT(recorded_events, testing::IsEmpty());
388}
389
401TEST_F(WhenAllTest, two_branches_host_device_followed_by_device) {
403
404 const view_s_t data(Kokkos::view_alloc("data - shared space"));
405
406 const context_t esc{exec};
407
408 const host_execution_space exec_h{};
409 const context_h_t esc_h{exec_h};
410
411 auto sndr = stdexec::when_all(
412 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data),
413 stdexec::schedule(esc_h.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data))
414 | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
415
416 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
417
418 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(std::move(sndr));
419
421 ASSERT_THAT(
422 recorded_events,
423 testing::ElementsAre(
425 MATCHER_FOR_BEGIN_SINGLE(exec_h, dispatch_label(exec_h, "then")),
428 } else {
429 ASSERT_THAT(recorded_events, [&]() {
431 return testing::ElementsAre(
433 MATCHER_FOR_BEGIN_SINGLE(exec_h, dispatch_label(exec_h, "then")),
435 MATCHER_FOR_WAIT_EVENT(recorded_events.at(2)),
438 } else {
439 return testing::ElementsAre(
441 MATCHER_FOR_BEGIN_SINGLE(exec_h, dispatch_label(exec_h, "then")),
442 MATCHER_FOR_BEGIN_FENCE(exec_h, dispatch_label(exec_h, "after dispatch")),
445 }
446 }());
447 }
448
449 ASSERT_EQ(data(), 3);
450}
451
463TEST_F(WhenAllTest, two_mixed_branches_followed_by_other_and_finish_on_self) {
464 const view_s_t data(Kokkos::view_alloc("data - shared space"));
465
466 const context_t esc{exec};
467 experimental::execution::single_thread_context stc{};
468
469 auto sndr = stdexec::when_all(
470 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data)
471 | stdexec::continues_on(stc.get_scheduler()),
472 stdexec::schedule(stc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data))
473 | stdexec::continues_on(stc.get_scheduler()) | THEN_INCREMENT(data)
474 | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
475
476 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
477
479
480 ASSERT_THAT(
482 testing::ElementsAre(
487
488 ASSERT_EQ(data(), 4);
489}
490
508TEST_F(WhenAllTest, nested_with_inner_followed_by_other) {
509 const view_s_t data(Kokkos::view_alloc("data - shared space"));
510
511 const context_t esc{exec};
512 experimental::execution::single_thread_context stc{};
513
514 auto sndr = stdexec::when_all(
515 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data),
516 stdexec::when_all(
517 stdexec::schedule(esc.get_scheduler()),
518 stdexec::schedule(esc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data))
520 | stdexec::continues_on(stc.get_scheduler()) | THEN_INCREMENT_ATOMIC(System, data)
521 | stdexec::continues_on(esc.get_scheduler()))
522 | stdexec::continues_on(esc.get_scheduler()) | THEN_INCREMENT(data);
523
524 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
525
527
528 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(std::move(sndr));
529
530 ASSERT_THAT(recorded_events, [&]() {
532 return testing::ElementsAre(
536 MATCHER_FOR_WAIT_EVENT(recorded_events.at(2)),
539 } else {
540 return testing::ElementsAre(
543 MATCHER_FOR_BEGIN_FENCE(exec, dispatch_label(exec, "after dispatch")),
546 }
547 }());
548
549 ASSERT_EQ(data(), 4);
550}
551
563TEST_F(WhenAllTest, nested_when_all_with_independent_branch) {
564 const auto [exec_A, exec_B, exec_C] = Kokkos::Experimental::partition_space(exec, 1, 1, 1);
565
566 const context_t esc_A{exec_A}, esc_B{exec_B}, esc_C{exec_C};
567
568 auto br_A = ::stdexec::schedule(esc_A.get_scheduler()) | THEN_LABELED_PFOR(TEST_EXECUTION_SPACE, 'A');
569 auto br_B = ::stdexec::schedule(esc_B.get_scheduler()) | THEN_LABELED_PFOR(TEST_EXECUTION_SPACE, 'B');
570 auto br_C = ::stdexec::schedule(esc_C.get_scheduler()) | THEN_LABELED_PFOR(TEST_EXECUTION_SPACE, 'C');
571
572 auto when_AB_then_D = ::stdexec::when_all(std::move(br_A), std::move(br_B))
573 | ::stdexec::continues_on(esc_A.get_scheduler()) | THEN_LABELED_PFOR(TEST_EXECUTION_SPACE, 'D');
574
575 auto sndr = ::stdexec::when_all(std::move(when_AB_then_D), std::move(br_C));
576
577 const auto recorded_events = Tests::Utils::record_sync_wait<recorder_listener_t>(std::move(sndr));
578
579 if (Tests::Utils::are_same_instances(exec_A, exec_B)) {
580 ASSERT_THAT(
581 recorded_events,
582 testing::ElementsAre(
583 MATCHER_FOR_BEGIN_PFOR(exec_A, "'A'"),
584 MATCHER_FOR_BEGIN_PFOR(exec_B, "'B'"),
585 MATCHER_FOR_BEGIN_PFOR(exec_A, "'D'"),
586 MATCHER_FOR_BEGIN_FENCE(exec_A, dispatch_label(exec_A, "after dispatch")),
587 MATCHER_FOR_BEGIN_PFOR(exec_C, "'C'"),
588 MATCHER_FOR_BEGIN_FENCE(exec_C, dispatch_label(exec_C, "after dispatch"))));
589 } else {
590 ASSERT_THAT(recorded_events, [&]() {
592 return testing::ElementsAre(
593 MATCHER_FOR_BEGIN_PFOR(exec_A, "'A'"),
594 MATCHER_FOR_BEGIN_PFOR(exec_B, "'B'"),
596 MATCHER_FOR_BEGIN_PFOR(exec_C, "'C'"),
598 MATCHER_FOR_WAIT_EVENT(recorded_events.at(2)),
599 MATCHER_FOR_BEGIN_PFOR(exec_A, "'D'"),
601 MATCHER_FOR_WAIT_EVENT(recorded_events.at(4)),
602 MATCHER_FOR_WAIT_EVENT(recorded_events.at(7)));
603 } else {
604 return testing::ElementsAre(
605 MATCHER_FOR_BEGIN_PFOR(exec_A, "'A'"),
606 MATCHER_FOR_BEGIN_PFOR(exec_B, "'B'"),
607 MATCHER_FOR_BEGIN_FENCE(exec_B, dispatch_label(exec_B, "after dispatch")),
608 MATCHER_FOR_BEGIN_PFOR(exec_A, "'D'"),
609 MATCHER_FOR_BEGIN_FENCE(exec_A, dispatch_label(exec_A, "after dispatch")),
610 MATCHER_FOR_BEGIN_PFOR(exec_C, "'C'"),
611 MATCHER_FOR_BEGIN_FENCE(exec_C, dispatch_label(exec_C, "after dispatch")));
612 }
613 }());
614 }
615}
616
629TEST_F(WhenAllTest, many_concurrent_branches) {
630 const view_s_t data(Kokkos::view_alloc(exec, "data - shared space"));
631
632 constexpr size_t num_branches = 6;
633
634 unsigned int counter_start = 0, counter_after_stc = 0;
635 std::array<unsigned int, num_branches> order_start{}, order_after_stc{};
636
637 const auto [exec_A, exec_B, exec_C, exec_D, exec_E, exec_F] =
638 Kokkos::Experimental::partition_space(exec, 1, 1, 1, 1, 1, 1);
639
640 using view_um_h_t = Kokkos::View<unsigned int, Kokkos::HostSpace, Kokkos::MemoryTraits<Kokkos::Unmanaged>>;
641 using increment_and_memorize_t = Tests::Utils::Functors::FetchIncrement<view_um_h_t, false>;
642
643#define DEFINE_ONE_BRANCH(_letter_, _id_) \
644 const context_t esc_##_letter_{exec_##_letter_}; \
645 experimental::execution::single_thread_context stc_##_letter_{}; \
646 auto br_##_letter_ = stdexec::just() \
647 | stdexec::then( \
648 increment_and_memorize_t{ \
649 .counter = view_um_h_t{std::addressof(counter_start)}, \
650 .value = view_um_h_t{std::addressof(order_start.at(_id_))}}) \
651 | stdexec::continues_on(stc_##_letter_.get_scheduler()) \
652 | stdexec::then( \
653 increment_and_memorize_t{ \
654 .counter = view_um_h_t{std::addressof(counter_after_stc)}, \
655 .value = view_um_h_t{std::addressof(order_after_stc.at(_id_))}}) \
656 | stdexec::continues_on(esc_##_letter_.get_scheduler()) | THEN_INCREMENT_ATOMIC(Device, data);
657
664
665 auto sndr = stdexec::when_all(
666 std::move(br_A), std::move(br_B), std::move(br_C), std::move(br_D), std::move(br_E), std::move(br_F));
667
668 ASSERT_EQ(data(), 0) << "Eager execution is not allowed.";
669
671
672 stdexec::sync_wait(std::move(sndr));
673
674 ASSERT_EQ(counter_start, num_branches);
675 ASSERT_EQ(counter_after_stc, num_branches);
676
677 ASSERT_EQ(data(), num_branches);
678
680 ASSERT_THAT(order_start, testing::ElementsAre(0, 1, 2, 3, 4, 5));
681
684 // NOLINTBEGIN(modernize-use-std-print)
685#define SHOW_ONE_BRANCH_ORDER_AFTER_STC(_letter_, _id_) \
686 SCOPED_TRACE(testing::Message() << "Branch " #_letter_ " order after 'stc' is " << order_after_stc.at(_id_));
687
694 // NOLINTEND(modernize-use-std-print)
695
696 ASSERT_THAT(order_after_stc, testing::UnorderedElementsAre(0, 1, 2, 3, 4, 5));
697}
698
699} // namespace Tests::ExecutionSpaceImpl
constexpr std::string dispatch_label(const Exec &, const std::string_view label)
Get the dispatch label from Exec and label.
#define MATCHER_FOR_BEGIN_SINGLE(_exec_, _label_)
#define MATCHER_FOR_WAIT_EVENT(_record_event_variant_)
#define MATCHER_FOR_BEGIN_PFOR(_exec_, _label_)
#define MATCHER_FOR_RECORD_EVENT(_exec_)
#define MATCHER_FOR_BEGIN_FENCE(_exec_, _label_)
RecorderListener< EventDiscardMatcher< TEST_EXECUTION_SPACE >, BeginFenceEvent, BeginParallelForEvent, BeginSingleEvent, Kokkos::Execution::Impl::RecordEvent, Kokkos::Execution::Impl::WaitEvent > recorder_listener_t
Concept for a sender whose completion scheduler is Kokkos::Execution::ExecutionSpaceImpl::Scheduler.
#define KOKKOS_EXECUTION_THREADS_THROWS_ON_SYNC_WAIT_ASSERT_AND_SKIP(_sndr_)
Definition context.hpp:76
Kokkos::DefaultHostExecutionSpace host_execution_space
#define DEFINE_ONE_BRANCH(_letter_, _id_)
#define SHOW_ONE_BRANCH_ORDER_AFTER_STC(_letter_, _id_)
#define KOKKOS_EXECUTION_STDEXEC_PRAGMA_DIAGNOSTIC_IGNORED
Basic list of ignored diagnostics when including anything from stdexec.
#define THEN_INCREMENT(_data_)
Add a then using Tests::Utils::Functors::Increment that may throw. // NOLINTNEXTLINE(cppcoreguideline...
Definition increment.hpp:64
#define THEN_INCREMENT_ATOMIC(_scope_, _data_)
Same as THEN_INCREMENT, using Tests::Utils::atomic_fetch_add. // NOLINTNEXTLINE(cppcoreguidelines-mac...
Definition increment.hpp:68
#define THEN_LABELED_PFOR(_exec_, _id_)
Add a Kokkos::Execution::parallel_for using Tests::Utils::Functors::Labeled. // NOLINTNEXTLINE(cppcor...
Definition labeled.hpp:26
constexpr check_rcvr_env_queryable_with_t< false, Queries... > check_rcvr_env_not_queryable_with
auto record_sync_wait(Sndr &&sndr)
Definition sync_wait.hpp:14
bool are_same_instances(const Exec &exec, const OtherExec &other_exec)
Definition kokkos.hpp:18
Matcher to filter out events that are just noise for tests.
Execution context using a Kokkos execution space under the hood.
auto get_scheduler() const noexcept -> ExecutionSpaceImpl::Scheduler< execution_space >
Event to be sent to Kokkos::utils::callbacks::dispatch when calling record.
Definition event.hpp:59
Event to be sent to Kokkos::utils::callbacks::dispatch when calling wait.
Definition event.hpp:80
Kokkos::Execution::ExecutionSpaceContext< Exec > context_t
Definition context.hpp:33