facebookexperimental / facebookexperimental/libunifex

Investigate changing customisations of algorithms to be based on connect instead of invocation of algorithm itself

Open
#402 2 comments 0 reactions 0 assignees View on GitHub
documentation
Dominant language
C++
Stars
1.7k
Forks
210
PR merge metrics
No merged PRs in 30d

Description

The current implementation of scheduler affinity for the coroutine `task` type relies on special behaviour of being able to identify when the coroutine awaits a sender returned by `schedule(some_scheduler)` so that it can treat that as a change of the current scheduler to some other scheduler.

The current solution complicates the implementation of `schedule()` a lot and makes it difficult to extend to other kinds of expressions as well.

One option could be to turn invocation of sender algorithms into generic algorithms that just curry their arguments into a `sender` type which then customises `connect()` to then pass the arguments through to `connect(receiver, CPO, args...)`.

As all senders would then be effectively just an argument currying mechanism we could customise `await_transform()` when the awaited object is of type `sender, SomeScheduler>` to handle this.

This would also allow simpler customisation of entire sender expressions when, e.g. the expression is passed to `on(sched, some_sender_expression)`. The sender expression can just be an expression-template of a well-known structure/type that the scheduler can use to customise `on()`. See below for an example of how this might work.

## Implementation Sketch

It might also help if you think of `connect()` instead as `async_invoke()`.

For example: We can define a generic `sender` type, parameterised on a CPO which is just responsible for currying arguments.
```c++
template
struct sender {
constexpr sender(const sender& other) = default;
constexpr sender(sender&& other) = default;

template
requires sizeof...(Args) == sizeof...(Args2) && (std::constructible && ...)
constexpr sender(Args2&&... args2) : args(static_cast(args2)...)

// Allow further currying args to produce another sender.
template
requires (std::copy_constructible && ...) && (std::constructible, ExtraArgs> && ...)
sender...> operator()(ExtraArgs&&... extraArgs) & const {
return std::apply([&](const Args&... args) {
return sender...>{args..., static_cast(extraArgs)...};
}, this->args);
}

// Allow pipe operator to prepend an argument.
template
friend sender, Args...> operator|(Arg&& arg, const sender& s) {
return std::apply([&](const Args&... args) {
return sender, Args...>{static_cast(arg), args...};
}, s.args);
}

// etc.. for other value categories

// Customise async_invoke() to forward to the CPO
template
requires std::invocable, Receiver, CPO, const Args&..., ExtraArgs...>
friend decltype(auto) tag_invoke(tag_t, Receiver r, const sender& s, ExtraArgs&&... extraArgs) {
return std::apply([&](const Args&... args) {
return async_invoke(move(r), CPO{}, args..., static_cast(extraArgs)...);
}, s.args);
}

std::tuple args;
};
```

Then we can define CPOs to have a default implementation of async_invoke() that has the default implementation.
We can use a helper here to allow invocation of the CPO to return the sender.
```c++
template
struct sender_cpo_base {
template
sender...> operator()(Args&&... args) const {
return sender...>{static_cast(args)...};
}
};

struct just_t : sender_cpo_base {
template
struct default_op_state {
Receiver r;
std::tuple...> args;

friend void tag_invoke(tag_t, default_op_state& op) noexcept {
std::apply([&](Args&&... args) noexcept {
set_value(move(op.r), move(args)...);
}, move(op.args));
}
};

template
requires /* ... */
default_op_state...> tag_invoke(tag_t, Receiver r, just_t, Args&&... args) {
return {move(r), {static_cast(args)...}};
}
};

struct transform_t : sender_cpo_base {
template
struct default_op_state {
struct receiver {
op_state& op;

template
friend void tag_invoke(tag_t, receiver&& r, Values&&... values) {
try {
std::apply([&](Funcs&&... funcs) {
if constexpr (sizeof...(Funcs) == 1) {
set_value(move(r.op.r), move(funcs)(static_cast(values)...)...);
} else {
set_value(move(r.op.r), move(funcs)(values...)...);
}
}, move(r.op.funcs));
} catch (...) {
set_error(move(r.op.r), std::current_exception());
}
}
};

friend void tag_invoke(tag_t, default_op_state& op) noexcept {
start(op.childOp);
}

template
default_op_state(Receiver r, Sender&& s, Fs2&&... fs)
: r(move(r))
, funcs(static_cast(fs)...)
, childOp(async_invoke(receiver{*this}, static_cast(s)))
{}

Receiver r;
std::tuple funcs;
async_invoke_result_t childOp;
};
```

Then a customisation of the algorithm could be implemented as follows:
```
struct my_scheduler {
template
struct schedule_op { ... };

template
friend auto tag_invoke(tag_t, Receiver r, schedule_t, my_scheduler s) {
return schedule_op{ ... };
}
};
```

This doesn't yet handle things like sender-queries or sender-traits which also need to be considered, however.

## Sketch of Scheduler Customisation

```c++
struct my_scheduler {
// ...
};

// customise: on(my_scheduler{}, bulk(src, count, f))
template
auto tag_invoke(tag_t, Receiver r, tag_t, my_scheduler s, sender, Src, Count, Func> bulkOp) {
// customisation goes here...
}
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.