Implement execution::on adaptor

Author: miscco

stl/inc/execution

@@ -5341,6 +5341,25 @@ namespace execution {
     template <class _Sender>
     struct sender_traits : decltype(_Get_sender_traits_base<_Sender>()) {};
 
+    template <class _Sender>
+    _NODISCARD _CONSTEVAL auto _Get_sender_traits_derived() noexcept {
+        if constexpr (_Has_sender_types<sender_traits<_Sender>>) {
+            return _Typed_sender<sender_traits<_Sender>>{};
+        } else if constexpr (derived_from<_Sender, sender_base>) {
+            return _Sender_traits_base{};
+#if 0 // TRANSITION: coroutine support
+            } else if constexpr (_Is_awaitable<_Sender>) {
+                if constexpr (is_void_v<_Await_result_t<_Sender>>) {
+                    return _Void_sender<false>{};
+                } else {
+                    return _Sender_of<false, _Await_result_t<_Sender>>{};
+                }
+#endif // 0 TRANSITION: coroutine support
+        } else {
+            return _No_sender_traits{};
+        }
+    }
+
     template <class _Sender>
     concept sender = move_constructible<remove_cvref_t<_Sender>> && !requires {
         typename sender_traits<remove_cvref_t<_Sender>>::_Unspecialized;
@@ -5543,6 +5562,134 @@ namespace execution {
     _NODISCARD constexpr _Just_done_sender just_done() noexcept {
         return {};
     };
+
+    template <class _Tag, class... _Args>
+    concept _Tag_invocable_to_sender = tag_invocable<_Tag, _Args...> && sender<tag_invoke_result_t<_Tag, _Args...>>;
+
+    namespace _On {
+        template <scheduler _Scheduler, receiver _InputReceiver>
+        struct _WrappedReceiver {
+            _Scheduler _Sched;
+            _InputReceiver _Input_receiver;
+
+            friend constexpr void tag_invoke(set_done_t, _WrappedReceiver _Wrapped_receiver) noexcept {
+                _EXEC set_done(_STD move(_Wrapped_receiver._Input_receiver));
+            }
+            template <class _Error>
+            friend constexpr void tag_invoke(set_error_t, _WrappedReceiver _Wrapped_receiver, _Error&& _Err) noexcept {
+                _EXEC set_error(_STD move(_Wrapped_receiver._Input_receiver), _STD forward<_Error>(_Err));
+            }
+            template <class... _ArgTys>
+            friend constexpr void tag_invoke(
+                set_value_t, _WrappedReceiver _Wrapped_receiver, _ArgTys&&... _Args) noexcept {
+                _EXEC set_value(_STD move(_Wrapped_receiver._Input_receiver), _STD forward<_ArgTys>(_Args)...);
+            }
+            _NODISCARD friend constexpr auto tag_invoke(
+                get_scheduler_t, const _WrappedReceiver& _Wrapped_receiver) noexcept {
+                return _Wrapped_receiver._Sched;
+            }
+        };
+
+        template <scheduler _Scheduler, sender _InputSender, receiver _InputReceiver>
+        struct _InnerReceiver {
+            using _WrappedReceiver_t = _WrappedReceiver<_Scheduler, _InputReceiver>;
+
+            // Clang complains about incomplete type, so for now no _OutputSenderState*
+            _WrappedReceiver_t* _Wrapped_receiver;
+            _InputSender* _Input_sender;
+            invoke_result_t<connect_t, _InputSender, _WrappedReceiver_t>* _Op_state3;
+
+            friend constexpr void tag_invoke(set_done_t, _InnerReceiver _Inner_receiver) noexcept {
+                _EXEC set_done(_STD move(_Inner_receiver._Wrapped_receiver->_Input_receiver));
+            }
+            template <class _Error>
+            friend constexpr void tag_invoke(set_error_t, _InnerReceiver _Inner_receiver, _Error&& _Err) noexcept {
+                _EXEC set_error(
+                    _STD move(_Inner_receiver._Wrapped_receiver->_Input_receiver), _STD forward<_Error>(_Err));
+            }
+            friend constexpr void tag_invoke(set_value_t, _InnerReceiver _Inner_receiver) noexcept {
+                auto* _Wrapped_receiver = _Inner_receiver._Wrapped_receiver;
+                _TRY_BEGIN
+                // *this is stored inside of `_Op_state2` so the call below will invalidate it. store the members
+                auto* _Input_sender = _Inner_receiver._Input_sender;
+                auto* _Op_state3    = _Inner_receiver._Op_state3;
+                _STD construct_at(_Op_state3, _EXEC connect(_STD move(*_Input_sender), _STD move(*_Wrapped_receiver)));
+                _EXEC start(*_Op_state3);
+                _CATCH_ALL
+                _EXEC set_error(_STD move(_Wrapped_receiver->_Input_receiver), _STD current_exception());
+                _CATCH_END
+            }
+        };
+
+        template <scheduler _Scheduler, sender _InputSender, receiver _InputReceiver>
+        struct _OutputSenderState {
+            using _WrappedReceiver_t = _WrappedReceiver<_Scheduler, _InputReceiver>;
+            using _InnerReceiver_t   = _InnerReceiver<_Scheduler, _InputSender, _InputReceiver>;
+
+            _WrappedReceiver_t _Wrapped_receiver;
+            _InputSender _Input_sender;
+            union {
+                invoke_result_t<connect_t, invoke_result_t<schedule_t, _Scheduler>, _InnerReceiver_t> _Op_state2;
+                invoke_result_t<connect_t, _InputSender, _WrappedReceiver_t> _Op_state3;
+            };
+
+            explicit constexpr _OutputSenderState(_WrappedReceiver_t _Wrapped_receiver_, _InputSender _Input_sender_)
+                : _Wrapped_receiver(_STD move(_Wrapped_receiver_)), _Input_sender(_STD move(_Input_sender_)),
+                  _Op_state2(_EXEC connect(_EXEC schedule(_Wrapped_receiver._Sched),
+                      _InnerReceiver_t{_STD addressof(_Wrapped_receiver), _STD addressof(_Input_sender),
+                          _STD addressof(_Op_state3)})) {}
+
+            friend constexpr void tag_invoke(start_t, _OutputSenderState& _State) noexcept {
+                _EXEC start(_State._Op_state2);
+            }
+        };
+
+        template <scheduler _Scheduler, sender _InputSender>
+        struct _OutputSender {
+            _Scheduler _Sched;
+            _InputSender _Input_sender;
+
+            template <receiver _InputReceiver>
+            _NODISCARD friend constexpr auto
+                tag_invoke(connect_t, _OutputSender _Output_sender, _InputReceiver&& _Input_receiver) noexcept(
+                    is_nothrow_move_constructible_v<_Scheduler>&& is_nothrow_move_constructible_v<_InputSender>&&
+                        is_nothrow_constructible_v<remove_cvref_t<_InputReceiver>, _InputReceiver>) {
+                return _OutputSenderState<_Scheduler, _InputSender, _InputReceiver>{
+                    _WrappedReceiver<_Scheduler, _InputReceiver>{
+                        _STD move(_Output_sender._Sched), _STD forward<_InputReceiver>(_Input_receiver)},
+                    _STD move(_Output_sender._Input_sender)};
+            }
+
+            template <_One_of<set_value_t, set_error_t, set_done_t> _Target>
+            _NODISCARD friend constexpr auto tag_invoke(
+                get_completion_scheduler_t<_Target>, const _OutputSender& _Output_sender) noexcept requires
+                tag_observable<get_completion_scheduler_t<_Target>, _InputSender> {
+                return tag_invoke(get_completion_scheduler<_Target>, _Output_sender._Input_sender);
+            }
+        };
+
+        struct _Cpo {
+            template <scheduler _Scheduler, sender _InputSender>
+            _NODISCARD constexpr auto operator()(_Scheduler&& _Sched, _InputSender&& _Input_sender) const noexcept {
+                if constexpr (_Tag_invocable_to_sender<_Cpo, _Scheduler, _InputSender>) {
+                    return tag_invoke(
+                        *this, _STD forward<_Scheduler>(_Sched), _STD forward<_InputSender>(_Input_sender));
+                } else {
+                    return _OutputSender<_Scheduler, _InputSender>{
+                        _STD forward<_Scheduler>(_Sched), _STD forward<_InputSender>(_Input_sender)};
+                }
+            }
+        };
+    } // namespace _On
+    using on_t = _On::_Cpo;
+
+    inline namespace _Cpos {
+        inline constexpr _On::_Cpo on;
+    }
+
+    template <scheduler _Scheduler, sender _InputSender>
+    struct sender_traits<_On::_OutputSender<_Scheduler, _InputSender>>
+        : decltype(_Get_sender_traits_derived<_InputSender>()) {};
 } // namespace execution
 #endif // __cpp_lib_executors
 _STD_END

stl/inc/ranges

@@ -60,119 +60,6 @@ namespace ranges {
     template <bool _IsWrapped, class _Ty>
     using _Maybe_wrapped = conditional_t<_IsWrapped, _Ty, _Unwrapped_t<_Ty>>;
 
-    namespace _Pipe {
-        // clang-format off
-        template <class _Left, class _Right>
-        concept _Can_pipe = requires(_Left&& __l, _Right&& __r) {
-            static_cast<_Right&&>(__r)(static_cast<_Left&&>(__l));
-        };
-
-        template <class _Left, class _Right>
-        concept _Can_compose = constructible_from<remove_cvref_t<_Left>, _Left>
-            && constructible_from<remove_cvref_t<_Right>, _Right>;
-        // clang-format on
-
-        template <class, class>
-        struct _Pipeline;
-
-        template <class _Derived>
-        struct _Base {
-            // clang-format off
-            template <class _Other>
-                requires _Can_compose<_Derived, _Other>
-            constexpr auto operator|(_Base<_Other>&& __r) && noexcept(
-                noexcept(_Pipeline{static_cast<_Derived&&>(*this), static_cast<_Other&&>(__r)})) {
-                // clang-format on
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Derived, _Base<_Derived>>);
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Other, _Base<_Other>>);
-                return _Pipeline{static_cast<_Derived&&>(*this), static_cast<_Other&&>(__r)};
-            }
-
-            // clang-format off
-            template <class _Other>
-                requires _Can_compose<_Derived, const _Other&>
-            constexpr auto operator|(const _Base<_Other>& __r) && noexcept(
-                noexcept(_Pipeline{static_cast<_Derived&&>(*this), static_cast<const _Other&>(__r)})) {
-                // clang-format on
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Derived, _Base<_Derived>>);
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Other, _Base<_Other>>);
-                return _Pipeline{static_cast<_Derived&&>(*this), static_cast<const _Other&>(__r)};
-            }
-
-            // clang-format off
-            template <class _Other>
-                requires _Can_compose<const _Derived&, _Other>
-            constexpr auto operator|(_Base<_Other>&& __r) const& noexcept(
-                noexcept(_Pipeline{static_cast<const _Derived&>(*this), static_cast<_Other&&>(__r)})) {
-                // clang-format on
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Derived, _Base<_Derived>>);
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Other, _Base<_Other>>);
-                return _Pipeline{static_cast<const _Derived&>(*this), static_cast<_Other&&>(__r)};
-            }
-
-            // clang-format off
-            template <class _Other>
-                requires _Can_compose<const _Derived&, const _Other&>
-            constexpr auto operator|(const _Base<_Other>& __r) const& noexcept(
-                noexcept(_Pipeline{static_cast<const _Derived&>(*this), static_cast<const _Other&>(__r)})) {
-                // clang-format on
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Derived, _Base<_Derived>>);
-                _STL_INTERNAL_STATIC_ASSERT(derived_from<_Other, _Base<_Other>>);
-                return _Pipeline{static_cast<const _Derived&>(*this), static_cast<const _Other&>(__r)};
-            }
-
-            template <_Can_pipe<const _Derived&> _Left>
-            friend constexpr auto operator|(_Left&& __l, const _Base& __r)
-#ifdef __EDG__ // TRANSITION, VSO-1222776
-                noexcept(noexcept(_STD declval<const _Derived&>()(_STD forward<_Left>(__l))))
-#else // ^^^ workaround / no workaround vvv
-                noexcept(noexcept(static_cast<const _Derived&>(__r)(_STD forward<_Left>(__l))))
-#endif // TRANSITION, VSO-1222776
-            {
-                return static_cast<const _Derived&>(__r)(_STD forward<_Left>(__l));
-            }
-
-            template <_Can_pipe<_Derived> _Left>
-            friend constexpr auto operator|(_Left&& __l, _Base&& __r)
-#ifdef __EDG__ // TRANSITION, VSO-1222776
-                noexcept(noexcept(_STD declval<_Derived>()(_STD forward<_Left>(__l))))
-#else // ^^^ workaround / no workaround vvv
-                noexcept(noexcept(static_cast<_Derived&&>(__r)(_STD forward<_Left>(__l))))
-#endif // TRANSITION, VSO-1222776
-            {
-                return static_cast<_Derived&&>(__r)(_STD forward<_Left>(__l));
-            }
-        };
-
-        template <class _Left, class _Right>
-        struct _Pipeline : _Base<_Pipeline<_Left, _Right>> {
-            /* [[no_unique_address]] */ _Left __l;
-            /* [[no_unique_address]] */ _Right __r;
-
-            template <class _Ty1, class _Ty2>
-            constexpr explicit _Pipeline(_Ty1&& _Val1, _Ty2&& _Val2) noexcept(
-                is_nothrow_convertible_v<_Ty1, _Left>&& is_nothrow_convertible_v<_Ty2, _Right>)
-                : __l(_STD forward<_Ty1>(_Val1)), __r(_STD forward<_Ty2>(_Val2)) {}
-
-            template <class _Ty>
-            _NODISCARD constexpr auto operator()(_Ty&& _Val) noexcept(
-                noexcept(__r(__l(_STD forward<_Ty>(_Val))))) requires requires {
-                __r(__l(static_cast<_Ty&&>(_Val)));
-            }
-            { return __r(__l(_STD forward<_Ty>(_Val))); }
-
-            template <class _Ty>
-            _NODISCARD constexpr auto operator()(_Ty&& _Val) const
-                noexcept(noexcept(__r(__l(_STD forward<_Ty>(_Val))))) requires requires {
-                __r(__l(static_cast<_Ty&&>(_Val)));
-            }
-            { return __r(__l(_STD forward<_Ty>(_Val))); }
-        };
-
-        template <class _Ty1, class _Ty2>
-        _Pipeline(_Ty1, _Ty2) -> _Pipeline<_Ty1, _Ty2>;
-    } // namespace _Pipe
-
     template <range _Rng, class _Derived>
     class _Cached_position : public view_interface<_Derived> {
         static_assert(_Always_false<_Rng>, "A range must be at least forward for position caching to be worthwhile.");