Propagating an executor through receivers in the pushmi model

[AerialMantis]

After today's futures telecom, I thought I would put up some code which illustrates the use case that I am concerned about that could give us a starting point for figuring out a potential solution. And also explain for those who weren't on the call.

Beware that I had to reduce code from my implementation to make this easier to follow so this is slideware.

To illustrate my problem, consider the following chain of tasks:

auto ret = pipe(
    gpu_executor{},
    twoway_execute([](){
        return f();
    }),
    then_execute([](auto value){
        return g(value);
    }),
    get<int>()
);

Note that here then_execute is analogous to transform, I just changed the name as I had to change its behaviour.

If you wanted to launch an asynchronous task on a GPU for both the twoway_execute and the then_execute then you need the executor to available to the receivers of both. However, in the current pushmi design only the value can be set within the twoway_execute receiver and therefore passed to the then_execute receiver:

template<class ContinuationFn>
auto twoway(ContinuationFn cfn){
    auto adapterFunc = [cfn](auto ex) {
        auto futureFunc = [cfn, ex](auto out) {
            auto promiseFunc = [cfn](auto out, auto ex){
                // Launch an asynchronous task on the execution context of ex.
                auto res = ex.execute(cfn);
                detail::set_value(out res);
            };
            detail::submit(ex, my_promise{out, promiseFunc});
        };
        return my_future{futureFunc};
    };
    return my_adaptor{adapterFunc};
}

template<class TransformFn>
auto then_execute(TransformFn tfn){
    auto adapterFunc = [tfn = std::move(tfn)](auto in){
        auto futureFunc = [in = std::move(in), tfn = std::move(tfn)](auto out) mutable {
            auto promiseFunc = [tfn](auto out, auto v) mutable {
                // Cannot launch another asynchronous task as the executor is not available.
                detail::set_value(out, tfn(v));
            };
            detail::submit(in, my_promise{std::move(out), promiseFunc});
        };
        return my_future{futureFunc};
    };
    return my_adaptor{adapterFunc};
}

The way I solved this in my implementation is that I had the receiver of twoway_execute pass a std::pair of the executor and the resulting value, so that the receiver of then_execute had the executor available to it:

template<class ContinuationFn>
auto twoway(ContinuationFn cfn){
    auto adapterFunc = [cfn](auto ex) {
        auto futureFunc = [cfn, ex](auto out) {
            auto promiseFunc = [cfn](auto out, auto ex){
                // Launch an asynchronous task on the execution context of ex.
                auto res = ex.execute(cfn);
                detail::set_value(out std::pair{ex, res});
            };
            detail::submit(ex, my_promise{out, promiseFunc});
        };
        return my_future{futureFunc};
    };
    return my_adaptor{adapterFunc};
}

template<class TransformFn>
auto then_execute(TransformFn tfn){
    auto adapterFunc = [tfn = std::move(tfn)](auto in){
        auto futureFunc = [in = std::move(in), tfn = std::move(tfn)](auto out) mutable {
            auto promiseFunc = [tfn](auto out, auto evp) mutable {
                // Launch another asynchronous on the same exceution context using propogated executor.
                auto res = ex.execute(tfn, evp.second);
                detail::set_value(out, std::pair{evp.first, res});
            };
            detail::submit(in, my_promise{std::move(out), promiseFunc});
        };
        return my_future{futureFunc};
    };
    return my_adaptor{adapterFunc};
}

Now, this worked quite well for my purposes, but my concern is whether this is in the spirit of the sender-receiver model and whether this is the way you would expect this kind of execution to be implemented.

An alternative solution for this, that was discussed on the call, that would not require the propagation of the executor through receivers, was to use thread local storage to store the executor. Though as we discussed this does have a number of issues, one being in a potential fibres implementation where the thread can change, another being when you want to perform a require adaptation so you end up with a different executor type.

I understand that having a sender be passed to a receiver brakes the sender-receiver model, so I would propose that we in some way separate the concept of a sender from an executor so that an executor can still be passed to a receiver without allowing a sender to be passed to a receiver.

@kirkshoop in the call I believe you mentioned an approach in which a sender would have a nullary function which can return an executor. I'm not sure if this would solve the problem as you would surely still have to pass the sender to the receiver, if I understood this correctly. Perhaps you could elaborate a little further on this idea?

Another idea I had which I have been toying with is that you could instead propagate an executor through the adapters so that then the executor could be captured within a receiver that you require to have it, and have the executor replaced by the via adapter. Though I am still working on an implementation of this so I am not sure yet of how feasible this is.