merge main

NVIDIA · ZelboK · Aug 23, 2023 · Aug 25, 2023 · Aug 29, 2023 · Sep 1, 2023
commit 6f555416d079a64232ddc7ab5d76f07184d580e2
diff --git a/test/exec/test_any_sender.cpp b/test/exec/test_any_sender.cpp
@@ -622,5 +622,104 @@ namespace {
       start(op);
     }
   }
-  CHECK(counting_scheduler::count == 0);
+
+  TEST_CASE("Schedule Sender lifetime", "[types][any_scheduler][any_sender]") {
+    using receiver_ref =
+      any_receiver_ref<completion_signatures<set_stopped_t(), set_error_t(std::exception_ptr)>>;
+    using sender_t = receiver_ref::any_sender<>;
+    using scheduler_t = sender_t::any_scheduler<>;
+    scheduler_t scheduler = exec::inline_scheduler();
+    auto sched = schedule(scheduler);
+    scheduler = stopped_scheduler();
+    {
+      auto op = connect(schedule(scheduler), expect_stopped_receiver{});
+      start(op);
+    }
+    {
+      auto op = connect(std::move(sched), expect_void_receiver{});
+      start(op);
+    }
+  }
+
+  // A scheduler that counts how many instances are extant.
+  struct counting_scheduler {
+    using __id = counting_scheduler;
+    using __t = counting_scheduler;
+
+    static int count;
+
+    counting_scheduler() noexcept {
+      ++count;
+    }
+
+    counting_scheduler(const counting_scheduler&) noexcept {
+      ++count;
+    }
+
+    counting_scheduler(counting_scheduler&&) noexcept {
+      ++count;
+    }
+
+    ~counting_scheduler() {
+      --count;
+    }
+
+    bool operator==(const counting_scheduler&) const noexcept = default;
+
+   private:
+    template <class R>
+    struct operation : immovable {
+      R recv_;
+
+      friend void tag_invoke(ex::start_t, operation& self) noexcept {
+        ex::set_value((R&&) self.recv_);
+      }
+    };
+
+    struct sender {
+      using __id = sender;
+      using __t = sender;
+
+      using is_sender = void;
+      using completion_signatures = ex::completion_signatures<ex::set_value_t()>;
+
+      template <ex::receiver R>
+      friend operation<R> tag_invoke(ex::connect_t, sender self, R r) {
+        return {{}, (R&&) r};
+      }
+
+      friend auto tag_invoke(ex::get_completion_scheduler_t<ex::set_value_t>, sender) noexcept
+        -> counting_scheduler {
+        return {};
+      }
+
+      friend const sender& tag_invoke(ex::get_env_t, const sender& self) noexcept {
+        return self;
+      }
+    };
+
+    friend sender tag_invoke(ex::schedule_t, counting_scheduler) noexcept {
+      return {};
+    }
+  };
+
+  int counting_scheduler::count = 0;
+
+  TEST_CASE(
+    "check that any_scheduler cleans up all resources",
+    "[types][any_scheduler][any_sender]") {
+    using receiver_ref = any_receiver_ref<completion_signatures<set_value_t()>>;
+    using sender_t = receiver_ref::any_sender<>;
+    using scheduler_t = sender_t::any_scheduler<>;
+    {
+      scheduler_t scheduler = exec::inline_scheduler{};
+      scheduler = counting_scheduler{};
+      {
+        auto op = connect(schedule(scheduler), expect_value_receiver<>{});
+        start(op);
+      }
+    }
+    CHECK(counting_scheduler::count == 0);
+  }
+
 }
diff --git a/test/nvexec/upon_stopped.cpp b/test/nvexec/upon_stopped.cpp
@@ -10,42 +10,29 @@ using nvexec::is_on_gpu;
 
 namespace {
 
-  auto snd = ex::just_stopped() | ex::transfer(stream_ctx.get_scheduler())
-           | ex::upon_stopped([] { return ex::just(); });
-  STATIC_REQUIRE(ex::sender<decltype(snd)>);
-  (void) snd;
-}
-
-template <class...>
-[[deprecated]] void print(auto&&...) {
-}
-
-namespace N {
-  using stdexec::__tag_invoke::tag_invoke;
-
-  struct tag_invoke_t {
-    template <class... As>
-    auto operator()(As&&... as) const {
-      tag_invoke((As&&) as...);
-    }
-  };
-}
-
-TEST_CASE("nvexec upon_stopped executes on GPU", "[cuda][stream][adaptors][upon_stopped]") {
-  nvexec::stream_context stream_ctx{};
-
-  flags_storage_t flags_storage{};
-  auto flags = flags_storage.get();
-
-  auto snd = ex::just_stopped()                       //
-           | ex::transfer(stream_ctx.get_scheduler()) //
-           | ex::upon_stopped([=] {
-               if (is_on_gpu()) {
-                 flags.set();
-               }
-             });
-
-  stdexec::sync_wait(std::move(snd));
-
-  REQUIRE(flags_storage.all_set_once());
+  TEST_CASE("nvexec upon_stopped returns a sender", "[cuda][stream][adaptors][upon_stopped]") {
+    nvexec::stream_context stream_ctx{};
+
+    auto snd = ex::just_stopped() | ex::transfer(stream_ctx.get_scheduler())
+             | ex::upon_stopped([] { return ex::just(); });
+    STATIC_REQUIRE(ex::sender<decltype(snd)>);
+    (void) snd;
+  }
+
+  TEST_CASE("nvexec upon_stopped executes on GPU", "[cuda][stream][adaptors][upon_stopped]") {
+    nvexec::stream_context stream_ctx{};
+
+    flags_storage_t flags_storage{};
+    auto flags = flags_storage.get();
+
+    auto snd = ex::just_stopped() //
+             | ex::transfer(stream_ctx.get_scheduler()) | ex::upon_stopped([=] {
+                 if (is_on_gpu()) {
+                   flags.set();
+                 }
+               });
+    stdexec::sync_wait(std::move(snd));
+
+    REQUIRE(flags_storage.all_set_once());
+  }
 }
diff --git a/test/stdexec/algos/adaptors/test_bulk.cpp b/test/stdexec/algos/adaptors/test_bulk.cpp
@@ -275,9 +275,6 @@ namespace {
         stdexec::schedule(sch) | stdexec::bulk(n, [&](int id) { counters_1[id]++; }),
         stdexec::schedule(sch) | stdexec::bulk(n, [&](int id) { counters_2[id]++; }));
 
-      auto snd = ex::transfer_just(sch) //
-               | ex::bulk(n, [&counter](int idx) { counter[idx] = 0; })
-               | ex::bulk(n, [&counter](int idx) { counter[idx]++; });
       stdexec::sync_wait(std::move(snd));
 
       CHECK(std::count(counters_1.begin(), counters_1.end(), 1) == n);