[WIP] get_completion_scheduler query for senders as of project goals

.gitignore

@@ -5,3 +5,4 @@ build/
 /Folder.DotSettings.user
 /compile_commands.json
 /.cache
+/.ccls-cache

include/unifex/completion_channels.hpp

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) Rishabh Dwivedi <[email protected]>
+ *
+ * Licensed under the Apache License Version 2.0 with LLVM Exceptions
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *   https://llvm.org/LICENSE.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <unifex/detail/prologue.hpp>
+
+namespace unifex {
+struct set_value_t {};
+struct set_error_t {};
+struct set_done_t {};
+}  // namespace unifex
+
+#include <unifex/detail/epilogue.hpp>

include/unifex/get_completion_scheduler.hpp

@@ -0,0 +1,78 @@
+/*
+ * Copyright (c) Rishabh Dwivedi <[email protected]>
+ *
+ * Licensed under the Apache License Version 2.0 with LLVM Exceptions
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *   https://llvm.org/LICENSE.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <unifex/config.hpp>
+#include <unifex/completion_channels.hpp>
+#include <unifex/scheduler_concepts.hpp>
+#include <unifex/sender_concepts.hpp>
+#include <unifex/std_concepts.hpp>
+#include <unifex/tag_invoke.hpp>
+
+#include <type_traits>
+
+#include <unifex/detail/prologue.hpp>
+
+namespace unifex {
+namespace _get_completion_scheduler {
+
+template <typename CPO>
+struct _fn;
+
+template <>
+struct _fn<set_value_t> {
+  template (typename Sender)
+  (requires sender<Sender> AND
+            is_nothrow_tag_invocable_v<_fn, Sender const&> AND
+            scheduler<tag_invoke_result_t<_fn, Sender const&>>) 
+  constexpr auto operator()(const Sender& s) const noexcept {
+    return tag_invoke(*this, s);
+  }
+};
+
+template <>
+struct _fn<set_error_t> {
+  template (typename Sender)
+  (requires sender<Sender> AND
+            is_nothrow_tag_invocable_v<_fn, Sender const&> AND
+            scheduler<tag_invoke_result_t<_fn, Sender const&>>) 
+  constexpr auto operator()(Sender const& sender) const noexcept
+      -> tag_invoke_result_t<_fn, Sender const&> {
+    return tag_invoke(*this, sender);
+  }
+};
+
+template <>
+struct _fn<set_done_t> {
+  template (typename Sender)
+  (requires sender<Sender> AND
+            is_nothrow_tag_invocable_v<_fn, Sender const&> AND
+            scheduler<tag_invoke_result_t<_fn, Sender const&>>) 
+  constexpr auto operator()(Sender const& sender) const noexcept
+      -> tag_invoke_result_t<_fn, Sender const&> {
+    return tag_invoke(*this, sender);
+  }
+};
+
+}  // namespace _get_completion_scheduler
+
+template <typename CPO>
+inline _get_completion_scheduler::_fn<CPO> get_completion_scheduler{};
+
+}  // namespace unifex
+
+#include <unifex/detail/epilogue.hpp>

include/unifex/new_thread_context.hpp

@@ -15,8 +15,11 @@
  */
 #pragma once
 
+#include <unifex/completion_channels.hpp>
+#include <unifex/get_completion_scheduler.hpp>
 #include <unifex/get_stop_token.hpp>
 #include <unifex/receiver_concepts.hpp>
+#include <unifex/tag_invoke.hpp>
 
 #include <atomic>
 #include <condition_variable>
@@ -39,18 +42,17 @@ using operation = typename _op<remove_cvref_t<Receiver>>::type;
 
 template <typename Receiver>
 class _op<Receiver>::type final {
- public:
+public:
   template <typename Receiver2>
   explicit type(context* ctx, Receiver2&& r)
-    : ctx_(ctx), receiver_((Receiver2&&)r) {}
+    : ctx_(ctx)
+    , receiver_((Receiver2 &&) r) {}
 
-  ~type() {
-      UNIFEX_ASSERT(!thread_.joinable());
-  }
+  ~type() { UNIFEX_ASSERT(!thread_.joinable()); }
 
   void start() & noexcept;
 
- private:
+private:
   void run() noexcept;
 
   context* ctx_;
@@ -65,23 +67,34 @@ class context {
   template <class Receiver>
   friend struct _op;
 
+  class scheduler;
+
   class schedule_sender {
   public:
-    template <template <typename...> class Variant,
-             template <typename...> class Tuple>
+    template <
+        template <typename...>
+        class Variant,
+        template <typename...>
+        class Tuple>
     using value_types = Variant<Tuple<>>;
 
     template <template <typename...> class Variant>
     using error_types = Variant<std::exception_ptr>;
 
     static constexpr bool sends_done = true;
 
-    explicit schedule_sender(context* ctx) noexcept
-      : context_(ctx) {}
+    explicit schedule_sender(context* ctx) noexcept : context_(ctx) {}
 
     template <typename Receiver>
     operation<Receiver> connect(Receiver&& r) const {
-      return operation<Receiver>{context_, (Receiver&&)r};
+      return operation<Receiver>{context_, (Receiver &&) r};
+    }
+
+    template <typename CPO>
+    friend auto tag_invoke(
+        _get_completion_scheduler::_fn<CPO>,
+        schedule_sender const& sender) noexcept -> scheduler {
+      return scheduler{sender.context_};
     }
 
   private:
@@ -110,9 +123,9 @@ class context {
   context() = default;
 
   ~context() {
-    // The activeThreadCount_ counter is initialised to 1 so it will never get to
-    // zero until after enter the destructor and decrement the last count here.
-    // We do this so that the retire_thread() call doesn't end up calling
+    // The activeThreadCount_ counter is initialised to 1 so it will never get
+    // to zero until after enter the destructor and decrement the last count
+    // here. We do this so that the retire_thread() call doesn't end up calling
     // into the cv_.notify_one() until we are about to start waiting on the
     // cv.
     activeThreadCount_.fetch_sub(1, std::memory_order_relaxed);
@@ -126,9 +139,7 @@ class context {
     }
   }
 
-  scheduler get_scheduler() noexcept {
-    return scheduler{this};
-  }
+  scheduler get_scheduler() noexcept { return scheduler{this}; }
 
 private:
   void retire_thread(std::thread t) noexcept {
@@ -170,7 +181,8 @@ inline void _op<Receiver>::type::start() & noexcept {
     // we ensure the count increment happens before the count decrement that
     // is performed when the thread is being retired.
     ctx_->activeThreadCount_.fetch_add(1, std::memory_order_relaxed);
-  } UNIFEX_CATCH (...) {
+  }
+  UNIFEX_CATCH(...) {
     unifex::set_error(std::move(receiver_), std::current_exception());
   }
 }
@@ -191,29 +203,28 @@ inline void _op<Receiver>::type::run() noexcept {
     // inside start().
     //
     // TODO: This can be replaced with an atomic<bool>::wait() once we have
-    // access to C++20 atomics. This would eliminate the unnecessary synchronisation
-    // performed by the unlock() at end-of-scope here.
+    // access to C++20 atomics. This would eliminate the unnecessary
+    // synchronisation performed by the unlock() at end-of-scope here.
     std::lock_guard opLock{mut_};
     thisThread = std::move(thread_);
   }
 
   if (get_stop_token(receiver_).stop_requested()) {
     unifex::set_done(std::move(receiver_));
   } else {
-    UNIFEX_TRY {
-      unifex::set_value(std::move(receiver_));
-    } UNIFEX_CATCH (...) {
+    UNIFEX_TRY { unifex::set_value(std::move(receiver_)); }
+    UNIFEX_CATCH(...) {
       unifex::set_error(std::move(receiver_), std::current_exception());
     }
   }
 
   ctx->retire_thread(std::move(thisThread));
 }
 
-} // namespace _new_thread
+}  // namespace _new_thread
 
 using new_thread_context = _new_thread::context;
 
-} // namespace unifex
+}  // namespace unifex
 
 #include <unifex/detail/epilogue.hpp>

test/get_completion_scheduler_test.cpp

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) Rishabh Dwivedi <[email protected]>
+ *
+ * Licensed under the Apache License Version 2.0 with LLVM Exceptions
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *   https://llvm.org/LICENSE.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <unifex/get_completion_scheduler.hpp>
+#include <unifex/new_thread_context.hpp>
+#include <unifex/type_traits.hpp>
+
+#include <gtest/gtest.h>
+
+TEST(GetCompletionScheduler, NewThreadScheduler) {
+  unifex::new_thread_context ctx;
+  auto sch = ctx.get_scheduler();
+  auto sender = unifex::schedule(sch);
+  static_assert(unifex::same_as<decltype(sch), decltype(unifex::get_completion_scheduler<unifex::set_value_t>(sender))>);
+  static_assert(unifex::same_as<decltype(sch), decltype(unifex::get_completion_scheduler<unifex::set_error_t>(sender))>);
+  static_assert(unifex::same_as<decltype(sch), decltype(unifex::get_completion_scheduler<unifex::set_done_t>(sender))>);
+}