Added support for cancellable raw senders with a lambda body

Author: MaxMotovilov

doc/api_reference.md

@@ -274,7 +274,9 @@ callable is passed an rvalue reference to receiver as an argument and is expecte
 move-construct the receiver inside the returned object.
 
 The operation state returned by the factory should either define a method `void
-start() noexcept` or be a non-throwing callable.
+start() noexcept` or be a non-throwing callable. For senders wrapped into `cancellable`
+to implement cancellation, there are additional requirements on the operation state
+(see [`cancellable`](#cancellablesender-sender-stdtrue_type---sender)).
 
 `ValueTypes...` is a pack representing the value types of the resulting sender. The
 user is expected to call `unifex::set_value()` passing in the rvalue to preserved
@@ -501,6 +503,41 @@ controlling what happens when stop is requested before the operation is started:
 If the receiver does not support cancellation, no stop callback is
 registered and the operation runs unconditionally without stop overhead.
 
+#### Implementing operation state with a lambda
+
+The combination of `cancellable` and `create_raw_sender()` can be used with a
+lambda as operation state. Such a lambda must accept a generic (`auto`) event
+argument used to separate `start()` and `stop()` invocations into template
+instantiations with the alternate branch of the code optimized away with `if constexpr`:
+
+```c++
+auto makeMySender(StateArg arg) {
+  return cancellable(
+    create_raw_sender<ResultType>(
+      [arg](auto&& receiver) {
+        return [receiver = std::move(receiver), arg,
+                requestId = RequestId{}](auto event, auto* self) mutable {
+          if constexpr (event.is_start) {
+            requestId = async_request(
+              [self, &receiver](ResultType result) {
+                if (try_complete(self)) {
+                  unifex::set_value(std::move(receiver), result);
+                }
+              },
+              arg);
+          } else if constexpr (event.is_stop) {
+            if (try_complete(self)) {
+              cancel_async_request(requestId);
+            }
+          }
+        };
+      }));
+}
+```
+
+The `self` argument is what the implementation is to pass to `try_complete()`. It does
+not point directly to the lambda closure and is of a different pointer type.
+
 ### `then(Sender predecessor, Func func) -> Sender`
 
 Returns a sender that transforms the value of the `predecessor` by calling

include/unifex/create_raw_sender.hpp

@@ -19,6 +19,7 @@
 #include <unifex/sender_concepts.hpp>
 #include <unifex/type_traits.hpp>
 
+#include <unifex/detail/lambda_op.hpp>
 #include <unifex/detail/make_traits.hpp>
 #include <unifex/detail/prologue.hpp>
 
@@ -73,27 +74,6 @@ constexpr _make_traits::get_traits<_make_traits::sender_traits_literal>::type<
 
 namespace _create_raw_sndr {
 
-template <typename Receiver, typename Fn, typename... ValueTypes>
-struct _op {
-  static_assert(std::is_invocable_v<Fn, Receiver&&>);
-  using state_t = decltype(UNIFEX_DECLVAL(Fn)(UNIFEX_DECLVAL(Receiver)));
-  static_assert(std::is_nothrow_invocable_v<state_t>);
-
-  struct type {
-    explicit type(Receiver&& rec, Fn&& fn) noexcept(
-        std::is_nothrow_invocable_v<Fn, Receiver&&>)
-      : state_(fn(std::forward<Receiver>(rec))) {}
-
-    void start() noexcept { state_(); }
-
-  private:
-    UNIFEX_NO_UNIQUE_ADDRESS state_t state_;
-  };
-};
-
-template <typename Receiver, typename Fn, typename... ValueTypes>
-using _op_wrapper = typename _op<Receiver, Fn, ValueTypes...>::type;
-
 template <typename Tr, typename Fn, typename... ValueTypes>
 struct _sender : public Tr {
   template <
@@ -107,30 +87,32 @@ struct _sender : public Tr {
   explicit _sender(Fn&& fn) noexcept(std::is_nothrow_move_constructible_v<Fn>)
     : fn_(std::forward<Fn>(fn)) {}
 
+  // Overload 1: factory returns a type with start(). Returned directly.
   template <typename Receiver>
     requires receiver_of<Receiver, ValueTypes...> &&
-      (!std::is_invocable_v<
-          _start_cpo::_fn,
-          decltype(UNIFEX_DECLVAL(Fn)(UNIFEX_DECLVAL(Receiver)))&>)
+      std::is_invocable_v<
+                 _start_cpo::_fn,
+                 decltype(UNIFEX_DECLVAL(Fn)(UNIFEX_DECLVAL(Receiver)))&>
   friend auto
   tag_invoke(tag_t<connect>, _sender&& self, Receiver&& rec) noexcept(
-      std::is_nothrow_constructible_v<
-          _op_wrapper<Receiver, Fn, ValueTypes...>,
-          Receiver&&,
-          Fn&&>) {
-    return _op_wrapper<Receiver, Fn, ValueTypes...>{
-        std::forward<Receiver>(rec), std::move(self.fn_)};
+      std::is_nothrow_invocable_v<Fn, Receiver&&>) {
+    return std::move(self.fn_)(std::forward<Receiver>(rec));
   }
 
+  // Overload 2: factory returns a callable (no start()). Wrapped in
+  // _sender_op::_op which auto-selects the right specialization:
+  // plain callable → start() only; event-dispatch → start() + stop().
   template <typename Receiver>
     requires receiver_of<Receiver, ValueTypes...> &&
-      std::is_invocable_v<
-                 _start_cpo::_fn,
-                 decltype(UNIFEX_DECLVAL(Fn)(UNIFEX_DECLVAL(Receiver)))&>
+      (!std::is_invocable_v<
+          _start_cpo::_fn,
+          decltype(UNIFEX_DECLVAL(Fn)(UNIFEX_DECLVAL(Receiver)))&>)
   friend auto
   tag_invoke(tag_t<connect>, _sender&& self, Receiver&& rec) noexcept(
       std::is_nothrow_invocable_v<Fn, Receiver&&>) {
-    return std::move(self.fn_)(std::forward<Receiver>(rec));
+    using state_t = decltype(std::move(self.fn_)(std::forward<Receiver>(rec)));
+    return _lambda_op::_op<state_t>{
+        std::move(self.fn_)(std::forward<Receiver>(rec))};
   }
 
   UNIFEX_NO_UNIQUE_ADDRESS Fn fn_;

include/unifex/detail/lambda_op.hpp

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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/type_traits.hpp>
+
+#include <type_traits>
+
+namespace unifex::_lambda_op {
+
+// Probe types for detecting event-dispatch lambdas.
+// A lambda (auto event, auto* self) that branches on event.is_start /
+// event.is_stop is callable with these probes.
+struct _start_probe {
+  static constexpr bool is_start = true;
+  static constexpr bool is_stop = false;
+};
+
+struct _stop_probe {
+  static constexpr bool is_start = false;
+  static constexpr bool is_stop = true;
+};
+
+// Detection: a callable is event-dispatchable if it is not callable
+// with zero arguments but IS callable with start/stop probes.
+// The 2-arg check uses the probe types themselves as the self-pointer
+// type (not void*) to avoid try_complete<void> instantiation failures
+// when is_invocable_v triggers body instantiation for return-type
+// deduction on generic lambdas.
+template <typename T>
+inline constexpr bool _is_event_dispatchable_v = !std::is_invocable_v<T&> &&
+    (std::is_invocable_v<T&, _start_probe> ||
+     std::is_invocable_v<T&, _start_probe, _start_probe*>) &&
+    (std::is_invocable_v<T&, _stop_probe> ||
+     std::is_invocable_v<T&, _stop_probe, _stop_probe*>);
+
+// Unified operation state aggregate for callables returned by
+// create_raw_sender factories. Wraps a callable with start() (and
+// stop() for event-dispatch lambdas). Being an aggregate, supports
+// non-moveable callables (e.g. lambdas capturing std::atomic) when
+// constructed from a prvalue via guaranteed copy elision.
+//
+// The IsEventDispatch parameter is auto-deduced from the callable type.
+template <
+    typename Callable,
+    bool IsEventDispatch = _is_event_dispatchable_v<Callable>>
+struct _op;
+
+// Plain callable: operator()() → start()
+template <typename Callable>
+struct _op<Callable, false> {
+  UNIFEX_NO_UNIQUE_ADDRESS Callable callable_;
+  void start() noexcept { callable_(); }
+};
+
+// Event-dispatch lambda: operator()(event[, self*]) → start() + stop()
+template <typename Lambda>
+struct _op<Lambda, true> {
+  Lambda lambda_;
+
+  void start() noexcept {
+    if constexpr (std::is_invocable_v<Lambda&, _start_probe, _op*>) {
+      lambda_(_start_probe{}, this);
+    } else {
+      lambda_(_start_probe{});
+    }
+  }
+
+  void stop() noexcept {
+    if constexpr (std::is_invocable_v<Lambda&, _stop_probe, _op*>) {
+      lambda_(_stop_probe{}, this);
+    } else {
+      lambda_(_stop_probe{});
+    }
+  }
+};
+
+}  // namespace unifex::_lambda_op

test/cancellable_test.cpp

@@ -225,6 +225,158 @@ TEST(cancellable_test, with_query_value_stoppable_token) {
   EXPECT_FALSE(stopped);
 }
 
+TEST(cancellable_lambda_test, completes_synchronously) {
+  auto result{sync_wait(cancellable{create_raw_sender<int>([](auto&& receiver) {
+    return [receiver = std::forward<decltype(receiver)>(receiver)](
+               auto event, auto* self) mutable {
+      if constexpr (event.is_start) {
+        if (try_complete(self)) {
+          set_value(std::move(receiver), 42);
+        }
+      }
+    };
+  })})};
+  EXPECT_TRUE(result.has_value());
+  EXPECT_EQ(*result, 42);
+}
+
+// Note: stop_while_running with stop_when is tested by the struct-based
+// test above. The equivalent lambda version triggers an MSVC internal
+// compiler error due to excessive template nesting depth (stop_when +
+// cancellable + create_raw_sender + lambda). The stop mechanism with
+// event-dispatch lambdas is covered by stops_early and
+// completes_before_stop_is_forwarded below.
+
+TEST(cancellable_lambda_test, stops_early) {
+  bool started = false;
+  bool stopped = false;
+  auto result{sync_wait(let_value_with_stop_source([&](auto& stop_src) {
+    stop_src.request_stop();
+    return cancellable{
+        create_raw_sender<int>([&](auto&& receiver) {
+          return [receiver = std::forward<decltype(receiver)>(receiver),
+                  &started,
+                  &stopped](auto event, auto* self) mutable {
+            if constexpr (event.is_start) {
+              started = true;
+              if (try_complete(self)) {
+                set_value(std::move(receiver), 42);
+              }
+            } else if constexpr (event.is_stop) {
+              stopped = true;
+              if (try_complete(self)) {
+                set_done(std::move(receiver));
+              }
+            }
+          };
+        }),
+        std::true_type{}};
+  }))};
+  EXPECT_FALSE(result.has_value());
+  EXPECT_FALSE(started);
+  EXPECT_TRUE(stopped);
+}
+
+TEST(cancellable_lambda_test, completes_before_stop_is_forwarded) {
+  bool started = false;
+  bool stopped = false;
+  auto result{sync_wait(let_value_with_stop_source([&](auto& stop_src) {
+    stop_src.request_stop();
+    return cancellable{create_raw_sender<int>([&](auto&& receiver) {
+      return [receiver = std::forward<decltype(receiver)>(receiver),
+              &started,
+              &stopped](auto event, auto* self) mutable {
+        if constexpr (event.is_start) {
+          started = true;
+          if (try_complete(self)) {
+            set_value(std::move(receiver), 42);
+          }
+        } else if constexpr (event.is_stop) {
+          stopped = true;
+          if (try_complete(self)) {
+            set_done(std::move(receiver));
+          }
+        }
+      };
+    })};
+  }))};
+  EXPECT_TRUE(result.has_value());
+  EXPECT_EQ(*result, 42);
+  EXPECT_TRUE(started);
+  EXPECT_FALSE(stopped);
+}
+
+TEST(cancellable_lambda_test, without_self_pointer) {
+  // Lambda taking only (event) without self pointer.
+  // Useful when try_complete is not needed (e.g. unconditional
+  // synchronous completion with unstoppable token).
+  auto result{sync_wait(with_query_value(
+      cancellable{create_raw_sender<int>([](auto&& receiver) {
+        return [receiver = std::forward<decltype(receiver)>(receiver)](
+                   auto event) mutable {
+          if constexpr (event.is_start) {
+            set_value(std::move(receiver), 99);
+          }
+        };
+      })},
+      get_stop_token,
+      unstoppable_token{}))};
+  EXPECT_TRUE(result.has_value());
+  EXPECT_EQ(*result, 99);
+}
+
+TEST(cancellable_lambda_test, with_non_moveable_state) {
+  // Verifies that event-dispatch lambdas capturing non-moveable types
+  // (std::atomic) work via aggregate init from prvalue in _event_op.
+  auto result{sync_wait(with_query_value(
+      cancellable{create_raw_sender<int>([](auto&& receiver) {
+        return
+            [receiver = std::forward<decltype(receiver)>(receiver),
+             call_count = std::atomic<int>{0}](auto event, auto* self) mutable {
+              if constexpr (event.is_start) {
+                call_count.fetch_add(1, std::memory_order_relaxed);
+                if (try_complete(self)) {
+                  set_value(
+                      std::move(receiver),
+                      call_count.load(std::memory_order_relaxed));
+                }
+              }
+            };
+      })},
+      get_stop_token,
+      unstoppable_token{}))};
+  EXPECT_TRUE(result.has_value());
+  EXPECT_EQ(*result, 1);
+}
+
+TEST(cancellable_lambda_test, with_unstoppable_token) {
+  auto result{sync_wait(with_query_value(
+      cancellable{create_raw_sender<int>([](auto&& receiver) {
+        return [receiver = std::forward<decltype(receiver)>(receiver)](
+                   auto event, auto* self) mutable {
+          if constexpr (event.is_start) {
+            if (try_complete(self)) {
+              set_value(std::move(receiver), 42);
+            }
+          }
+        };
+      })},
+      get_stop_token,
+      unstoppable_token{}))};
+  EXPECT_TRUE(result.has_value());
+  EXPECT_EQ(*result, 42);
+}
+
+TEST(cancellable_lambda_test, event_probe_fields) {
+  // Verify probe types have the expected boolean fields.
+  // The lambda's if-constexpr dispatch depends on them.
+  using namespace unifex::_lambda_op;
+  static_assert(_start_probe::is_start);
+  static_assert(!_start_probe::is_stop);
+  static_assert(!_stop_probe::is_start);
+  static_assert(_stop_probe::is_stop);
+}
+
 #endif
 
 TEST(cancellable_test, constructs_sender_in_place) {