when_all: implementation missing mechanism for explicit resource transition?

[maartenarnst]

In a when_all with child senders that complete on different execution contexts, it appears the resource transitions are not explicit. This appears to be at odds with the p2300 proposal:

We propose that, for senders advertising their completion scheduler, all execution resource transitions must be explicit; running user code anywhere but where they defined it to run must be considered a bug.

For example, if we build a sender like,

TEST(WhenAll, different_domain_children) {
    experimental::execution::single_thread_context cpu{};
    nvexec::stream_context gpu{};

    auto when_all = stdexec::when_all(
        stdexec::schedule(cpu.get_scheduler()) | stdexec::then([] { }),
        stdexec::schedule(gpu.get_scheduler()) | stdexec::then([] __device__ { }));

    auto sndr = std::move(when_all) | stdexec::continues_on(cpu.get_scheduler()) | stdexec::then([] { });

    stdexec::sync_wait(sndr);
}

there is a resource transition from the gpu to the cpu, but it appears the current implementation of when_all does not make it explicit. In particular, the implementation of when_all does not insert a schedule_from to transition away from the gpu.

Note that it appears that the continues_on may not be considered to be responsible for the transition because the completion domain of the when_all is the default_domain. Also note that this snippet does not compile because of a host device error, but this compilation problem might be separate from the conceptual problem of how to think of resource transitions in when_all.

We therefore wanted to ask whether it is a deliberate design decision that the current implementation of when_all does not end branches with an explicit resource transition? If so, how should we think of such resource transitions?

Joint work with @romintomasetti.

[RobertLeahy]

If the design of when_all included a resource transition that transition would be "implicit" (as it would be hidden inside when_all) which is at odds with your quote from P2300.

[maartenarnst]

If the design of when_all included a resource transition that transition would be "implicit" (as it would be hidden inside when_all) which is at odds with your quote from P2300.

Hi @RobertLeahy. Thank you for your answer. It is a fair point. But we are still not entirely sure to fully understand, and so we wanted to continue the discussion a bit further.

This quote from P3826R3 may also be helpful:

In order to transition between two execution contexts, each of which may know nothing about the other, it is necessary to do it in two steps: a transition from a context to the default domain, and a transition from the default domain to another context. A scheduler can customize either or both of these two steps by customizing the continues_on and schedule_from algorithms.

When we consider a linear chain, we can clearly see this in action. E.g., for

schedule(gpu) | then --> continues_on(cpu) | then

as in

experimental::execution::single_thread_context cpu{};
nvexec::stream_context gpu{};

auto sndr = stdexec::schedule(gpu.get_scheduler()) | stdexec::then(Functor{})
          | stdexec::continues_on(cpu.get_scheduler()) | stdexec::then(Functor{});

we can look at the sender that stdexec builds up:

static_assert(
    std::same_as<
        stdexec::__demangle_t<decltype(sndr)>,
        stdexec::__basic_sender<
            stdexec::then_t,
            Functor,
            stdexec::__basic_sender<
                stdexec::continues_on_t,
                stdexec::run_loop::scheduler,
                stdexec::__basic_sender<
                    stdexec::schedule_from_t,
                    stdexec::__,
                    stdexec::__basic_sender<
                        stdexec::then_t,
                        Functor,
                        decltype(::stdexec::schedule(std::declval<nvexec::stream_context>().get_scheduler()))
                    >::type
                >::type
            >::type
        >::type
    >);

Here:

• stdexec builds up a sender in which the transition is represented explicitly in terms of a schedule_from away from the gpu, followed by a continues_on onto the cpu.
• the schedule_from provides a customisation point to insert work useful when transitioning away from the execution resource.

We can contrast this with the current behavior of when_all with heterogeneous branches. E.g., for

schedule(gpu) | then -- \
                         when_all --> continues_on(cpu) | then
schedule(cpu) | then -- /

as in

experimental::execution::single_thread_context cpu{};
nvexec::stream_context gpu{};

auto when_all = stdexec::when_all(
    stdexec::schedule(cpu.get_scheduler()) | stdexec::then(Functor{}),
    stdexec::schedule(gpu.get_scheduler()) | stdexec::then(Functor{}));

auto sndr = std::move(when_all) | stdexec::continues_on(cpu.get_scheduler()) | stdexec::then(Functor{});

we can again look at the sender that stdexec builds up:

static_assert(
    std::same_as<
        stdexec::__demangle_t<decltype(sndr)>,
        stdexec::__basic_sender<
            stdexec::then_t,
            Functor,
            stdexec::__basic_sender<
                stdexec::continues_on_t,
                stdexec::run_loop::scheduler,
                stdexec::__basic_sender<
                    stdexec::schedule_from_t,
                    stdexec::__,
                    stdexec::__basic_sender<
                        stdexec::when_all_t,
                        stdexec::__,
                        stdexec::__basic_sender<
                            stdexec::then_t,
                            Functor,
                            decltype(stdexec::schedule(
                                std::declval<experimental::execution::single_thread_context>().get_scheduler()))
                        >::type,
                        stdexec::__basic_sender<
                            stdexec::then_t,
                            Functor,
                            decltype(::stdexec::schedule(std::declval<::nvexec::stream_context>().get_scheduler()))
                        >::type
                    >::type
                >::type
            >::type
        >::type
    >);

This time:

• the transition away from a branch's execution resource to the common domain is not represented explicitly
• there is no customisation point to insert work useful when transitioning away from a branch's execution resource.

We note again that this snippet does not compile when connecting with stdexec::sync_wait(sndr);. Compilation fails with a host device error.

To us, this raises several questions:

• Semantics and need to represent transitions explicitly: Thinking again of the original quote from p2300 and the quote "In order to transition between two execution contexts, ... it is necessary to do it in two steps", it seems it is OK not to represent the second step within when_all because the destination is not known. However, it does seem worth wondering whether the first step of transitioning from each branch's execution resource to the/a default/common domain should be represented, because when_all does seem to possess all needed information. It seems it would not be implicit.
• Absence of customisation point for work when transitioning away from branches: How should a user handle branches that require work when transitioning off them? Is the recommended approach to customise when_all entirely? Terminate such branches manually with a continues_on? Customise schedule_from for indeterminate domains? Should the default implementation disallow usage with branches on execution contexts that require work when transitioning away from them? If so, how could this be achieved?
• transfer_when_all: Is transfer_when_all an algorithm that may lend itself to representing transitions explicitly? Indeed, this algorithm knows both the sources and the destination.

As a concrete example, we have implemented a case of a scheduler that encrypts a string during schedule. The corresponding schedule_from decrypts the string. Without explicit transition, the string remains encrypted.

example

#include <iostream>
#include <ranges>

#include <exec/completion_signatures.hpp>
#include <exec/single_thread_context.hpp>
#include <stdexec/execution.hpp>

//! Call @ref rot13 once to encrypt a string, and a second time to decipher it.
inline std::string rot13(std::string text) noexcept {
    for (auto& chk: text)
        if (std::isalpha(chk))
            chk = std::isupper(chk) ? 'A' + (chk - 'A' + 13) % 26 : 'a' + (chk - 'a' + 13) % 26;
    return text;
}

template <stdexec::receiver Rcvr>
struct CryptoReceiver {
    using receiver_concept = stdexec::receiver_t;

    Rcvr rcvr;

    void set_value(std::string value) && noexcept {
        stdexec::set_value(std::move(rcvr), rot13(std::move(value)));
    }

    auto get_env() const noexcept -> stdexec::env_of_t<Rcvr> {
        return stdexec::get_env(rcvr);
    }
};

template <stdexec::sender Sndr>
struct CryptoSender {
    using sender_concept = stdexec::sender_t;

    Sndr sndr;

    template <stdexec::__decays_to<CryptoSender> Self, typename... Env>
    static consteval auto get_completion_signatures() {
        return experimental::execution::get_child_completion_signatures<Self, Sndr, Env...>();
    }

    template <stdexec::receiver Rcvr>
    auto connect(Rcvr rcvr) && {
        return stdexec::connect(std::move(sndr), CryptoReceiver<Rcvr>{std::move(rcvr)});
    }

    auto get_env() const noexcept -> stdexec::env_of_t<Sndr> {
        return stdexec::get_env(sndr);
    }
};

//! Forward declaration.
struct CryptoScheduler;

//! Custom domain.
struct CryptoSchedulerDomain {
    //! For @c stdexec::schedule_from_t.
    template <stdexec::sender Sndr, typename Env>
    requires(std::same_as<stdexec::tag_of_t<Sndr>, stdexec::schedule_from_t>)
    static auto transform_sender(stdexec::set_value_t, Sndr&& sndr, const Env&) -> stdexec::sender auto {
        auto [tag, data, inner] = std::forward<Sndr>(sndr);
        return CryptoSender<decltype(inner)>{std::move(inner)};
    }

    //! For @c stdexec::continues_on_t.
    template <stdexec::sender Sndr, typename Env>
    requires(std::same_as<stdexec::tag_of_t<Sndr>, stdexec::continues_on_t>)
    static auto transform_sender(stdexec::set_value_t, Sndr&& sndr, const Env&) -> stdexec::sender auto {
        auto [tag, data, inner] = std::forward<Sndr>(sndr);
        return CryptoSender<decltype(inner)>{std::move(inner)};
    }
};

//! Scheduler.
struct CryptoScheduler {
    struct Sender {
        using sender_concept = stdexec::sender_t;

        using completion_signatures = stdexec::completion_signatures<stdexec::set_value_t()>;

        template <stdexec::receiver Rcvr>
        struct OpState {
            Rcvr rcvr;

            void start() & noexcept {
                stdexec::set_value(std::move(rcvr));
            }
        };

        template <stdexec::receiver Rcvr>
        OpState<Rcvr> connect(Rcvr rcvr) {
            return {std::move(rcvr)};
        }

        struct attrs {
            constexpr auto
                query(stdexec::get_completion_scheduler_t<stdexec::set_value_t>) const noexcept -> CryptoScheduler {
                return {};
            }
            constexpr auto
                query(stdexec::get_completion_domain_t<stdexec::set_value_t>) const noexcept -> CryptoSchedulerDomain {
                return {};
            }
        };

        constexpr auto get_env() const noexcept -> attrs {
            return {};
        }
    };

    Sender schedule() const {
        return {};
    }

    constexpr auto
        query(stdexec::get_completion_domain_t<stdexec::set_value_t>) const noexcept -> CryptoSchedulerDomain {
        return {};
    }

    bool operator==(const CryptoScheduler&) const = default;
};

//! Shout if any forbidden word is found.
void moderate_message_content(const std::string& message) {
    static constexpr std::array forbidden = {"bad", "evil", "wrong"};
    std::ranges::for_each(forbidden, [&](const auto& word) {
        if (message.find(word) != std::string::npos)
            throw std::runtime_error(
                std::string("Your message is rude because it contains '") + word + "'. Be nicer please.");
    });
}

int main() {
    exec::single_thread_context stc{};
    const CryptoScheduler scheduler{};

    const std::string message("This is my message to you! It is not rude but 'bad' is forbidden.");

    /// @c continues_on and @c schedule_from ensure that the value channel gets encrypted when entering the domain, and
    /// deciphered when exiting the domain.
    /// The @ref moderate_message_content thus properly warns and saved you from harsh words.
    try {
        const auto [good] = stdexec::sync_wait(
                                stdexec::just(message) | stdexec::continues_on(scheduler)
                                | stdexec::then([](std::string value) noexcept {
                                      /* Do something meaningful with the encrypted message. */
                                      return value;
                                  })
                                | stdexec::continues_on(stc.get_scheduler()) | stdexec::then([](std::string value) {
                                      moderate_message_content(value);
                                      return value;
                                  }))
                                .value();
        throw std::runtime_error("It should not arrive here, because your message is rude.");
    } catch (const std::runtime_error& e) {
        std::cout << "The message was properly detected as aggressive:\n\t" << e.what() << '\n';
    }

    //! The @ref moderate_message_content will let some rude words go through because @c schedule_from was not called.
    const auto [bad] = stdexec::sync_wait(
        stdexec::when_all(
            stdexec::schedule(stc.get_scheduler())
                | stdexec::then([]() { std::printf("Do whatever concurrently here\n"); }),
            stdexec::just(message) | stdexec::continues_on(scheduler) | stdexec::then([](std::string value) noexcept {
                /* Do something meaningful with the encrypted message. */
                return value;
            }))
        | stdexec::continues_on(stc.get_scheduler()) | stdexec::then([](std::string value) {
              moderate_message_content(value);
              return value;
          })).value();
    
    if (bad != rot13(message))
        throw std::runtime_error("The message should still be encrypted.");

    std::cout << "This is a rude message that is still encrypted and so the moderation did not catch it:\n\t" << bad << '\n';
}