Senders and coroutines together

Mateusz Pusz · Train IT ·

13 min read Original article ↗

Part 6 ended on a pattern that sender algorithms cannot express by composition: a retry loop. A fallback chain is a fixed sequence of steps, so let_error handles it. A retry with backoff, “try up to five times, sleeping longer between attempts,” is a loop, and the answer there was to write a custom sender or to use a ready-made one like the retry algorithm in stdexec’s examples.

Here is a retry without the backoff, written as a coroutine:

task<void> retry_n(sender auto work, int n)
{
  while (true) {
    try {
      co_await work;
      co_return;
    } catch (...) {
      if (n-- == 0) throw;
    }
  }
}

And a flaky operation that fails twice before it succeeds:

int attempts = 0;
sender auto flaky = just() | then([&] {
  if (++attempts < 3) throw std::runtime_error("not yet");
  std::cout << "Success after " << attempts << " attempts\n";
});

sync_wait(retry_n(flaky, 5));   // prints: Success after 3 attempts

retry_n takes a sender and returns a task, the coroutine type C++26 adds as std::execution::task (covered at the end of this post). task implements the sender interface itself, so sync_wait runs it like any other pipeline. Inside, co_await work starts the sender and suspends the coroutine until it completes. When work completes with set_error, the error comes out of the co_await as an exception, so an ordinary try/catch in a loop does the job that otherwise takes a custom sender such as stdexec’s retry, at the price of reporting every error as an std::exception_ptr. With n set to 2 and a sender that never succeeds, retry_n makes three attempts, the first one and two retries. Then the last exception propagates out of it, and sync_wait rethrows it.

The retry uses both directions of the interop: a coroutine awaiting a sender, and a coroutine that is itself a sender.

A coroutine task is a sender#

A task<int> is a sender with three completions: set_value(int) when the body reaches co_return, set_error(std::exception_ptr) when an exception escapes the body, and set_stopped() when it is stopped (more on that below). So it goes wherever a sender goes:

task<int> answer()
{
  int base = co_await just(40);
  co_return base + 2;
}

int main()
{
  exec::static_thread_pool pool{4};
  sender auto work = starts_on(pool.get_scheduler(), answer())
                   | then([](int value) { return value * 10; });
  auto [result] = sync_wait(std::move(work)).value();
  std::cout << result << '\n';   // prints: 420
}

when_all(answer(), answer()) works the same way, and so does let_value returning a task from its callable.

Awaiting a sender#

Inside a task, every co_await operand goes through the promise’s await_transform, which turns a sender into an awaitable with as_awaitable ([task.promise]). The result of the co_await depends on how the sender completes:

The sender completes withco_await sndr
set_value()resumes, and the expression is void
set_value(a)resumes, and yields a
set_value(a, b, ...)resumes, and yields std::tuple(a, b, ...)
set_error(e)resumes, and throws
set_stopped()never resumes: the promise’s unhandled_stopped() is called instead

What gets thrown depends on the type of e. An std::exception_ptr is rethrown, an std::error_code is thrown as std::system_error, and anything else is thrown as itself ([exec.general], AS-EXCEPT-PTR):

task<void> tour()
{
  co_await just();                                  // void
  int v = co_await just(0);                         // int
  auto [i, b, c] = co_await just(0, true, 'c');     // std::tuple<int, bool, char>
  try {
    co_await just_error(std::make_error_code(std::errc::timed_out));
  } catch (const std::system_error& e) {
    std::cout << "caught: " << e.code().message() << '\n';
  }
  co_await just_stopped();                          // never resumes
  std::cout << "unreachable\n";
}

Run with sync_wait, it prints caught: Connection timed out, never prints unreachable, and sync_wait returns an empty std::optional, which is how it reports a stopped completion.

Senders a coroutine cannot await#

A sender can succeed in more than one shape (two set_value completion signatures with different arguments), and a co_await expression can have only one type. Racing a reply against a deadline is a common way to get such a sender. The exec::when_any from Part 4 completes with whatever the winning child sent, so its value completions are those of all its children together:

sender auto read_reply();   // sends an int
sender auto deadline();     // sends nothing

task<void> client()
{
  // int reply = co_await exec::when_any(read_reply(), deadline());   // does not compile

  auto result = co_await into_variant(exec::when_any(read_reply(), deadline()));
  if (auto* reply = std::get_if<std::tuple<int>>(&result))
    std::cout << "reply: " << std::get<0>(*reply) << '\n';
  else
    std::cout << "timed out\n";
}

The commented-out line fails because as_awaitable requires at most one value completion signature. For anything else, unless the sender opts in through the hook described below, it hands the sender back unchanged ([exec.as.awaitable]), and a sender is not an awaiter. That is why GCC 15 complains about a missing member await_ready instead of saying anything about completions. into_variant collapses all the value completions into one, a std::variant of tuples, and that the coroutine can await. A sender can also collapse its completions itself: P3570, adopted for C++26, gave senders an opt-in hook for it, get_await_completion_adaptor.

What a stop request does to a coroutine#

Part 4 made the case that stopping is not an error, so there is no exception to throw for it. Instead, as_awaitable does not resume the coroutine at all. It calls unhandled_stopped() on the promise, which for a task destroys the coroutine frame and completes the task with set_stopped() ([task.promise]). Locals are destroyed with the frame, so RAII still runs. If that task was itself awaited by another task, the parent receives set_stopped() and is unwound the same way, and so on up the chain until something intercepts it or the outermost receiver gets set_stopped().

Here it is on a request raced against a timeout:

struct trace {
  std::string name;
  ~trace() { std::cout << "  ~trace(" << name << ")\n"; }
};

exec::timed_thread_context timer_ctx;
scheduler auto timer = timer_ctx.get_scheduler();

task<int> slow_lookup()
{
  trace t{"slow_lookup"};
  try {
    co_await exec::schedule_after(timer, 10s);   // a slow backend
    std::cout << "  slow_lookup resumed\n";
    co_return 42;
  } catch (...) {
    std::cout << "  slow_lookup: catch (...) entered\n";
    throw;
  }
}

task<int> handle_request()
{
  trace t{"handle_request"};
  int value = co_await slow_lookup();
  std::cout << "  handle_request resumed\n";
  co_return value + 1;
}

int main()
{
  std::cout << "racing the request against a 100 ms timeout\n";
  sender auto timeout = exec::schedule_after(timer, 100ms) | then([] { return -1; });
  auto [result] = sync_wait(exec::when_any(handle_request(), timeout)).value();
  std::cout << "result: " << result << '\n';
}
racing the request against a 100 ms timeout
  ~trace(slow_lookup)
  ~trace(handle_request)
result: -1

The timeout wins, when_any asks handle_request to stop, and the request is forwarded down to the timer that slow_lookup is waiting on. The timer completes with set_stopped() early, and both coroutine frames are destroyed innermost first. Neither “resumed” line prints, and neither does the catch (...). The whole program finishes in about 100 ms instead of 10 s. The stop is cooperative in the usual way: it arrived quickly because the timer registers a stop callback. A coroutine busy in a loop that never awaits anything would not notice it.

The language has no way to cancel a coroutine, only to destroy its frame. unhandled_stopped() is the library hook that does that and reports a stopped completion the parent can see, which is how coroutines get the cancellation that Part 4 said they lack natively.

To handle a stop inside the coroutine, convert it before it reaches the co_await. stopped_as_optional, or the upon_stopped and let_stopped from Part 3’s table, turn it into a value first:

task<int> cached_value(bool canceled);   // may complete with set_stopped()

task<int> value_or_default(bool canceled)
{
  if (auto v = co_await stopped_as_optional(cached_value(canceled)))
    co_return *v;
  std::cout << "stopped, falling back\n";
  co_return -1;
}

Where a coroutine resumes#

Part 5 started from a callback that runs its continuation on whatever thread completed the previous step. A coroutine that resumes inside the receiver has the same property: the line after co_await runs wherever the awaited sender completed. In a coroutine it is harder to notice because the code reads like sequential code on one thread.

std::execution::task is scheduler-affine by default. When it starts, it takes the scheduler it was started on from its receiver’s environment and keeps it. Every sender it awaits is wrapped in affine, which moves the completion back to that scheduler with a continues_on ([exec.affine], [task.promise]). Here the coroutine starts on one single-threaded pool and awaits work that completes on another:

exec::static_thread_pool ui_pool{1};   // stands in for a UI or reactor thread
exec::static_thread_pool io_pool{1};   // where the I/O completes

sender auto fetch()
{
  return starts_on(io_pool.get_scheduler(), just() | then([] {
    std::cout << "fetch runs on the " << where() << '\n';
    return 42;
  }));
}

task<void> process()
{
  std::cout << "before co_await: " << where() << '\n';
  int data = co_await fetch();
  std::cout << "got " << data << " on the " << where() << '\n';
}

// where() returns the name of the pool the calling thread belongs to
sync_wait(starts_on(ui_pool.get_scheduler(), process()));
before co_await: ui thread
fetch runs on the io thread
got 42 on the ui thread

Affinity can be switched off through the task’s environment parameter, and the output shows what you give up:

struct no_affinity {
  using start_scheduler_type = inline_scheduler;
};

task<void, no_affinity> process_no_affinity();   // same body as process()
before co_await: ui thread
fetch runs on the io thread
got 42 on the io thread

A few details of how it behaves:

  • Affinity is to a scheduler. A pool scheduler may resume the coroutine on any of its threads, so thread_local data is not stable across a co_await. P3552 spells this out.
  • It costs a scheduling operation per co_await, unless the awaited sender is known to complete where it started. The working draft recommends, without requiring it, that just, just_error, just_stopped, read_env, and write_env be recognized as such, so that awaiting them does not reschedule ([exec.affine]/6).
  • A default task needs a scheduler to start on. sync_wait provides one (its run loop), and so does starts_on. spawn into a counting_scope from Part 7 provides none.

Spawning a task into a scope shows the last point, and one more restriction:

counting_scope scope;
// spawn(log_event(1), scope.get_token());   // does not compile
spawn(starts_on(sched, log_event(1))
        | upon_error([](std::exception_ptr) noexcept { /* log it */ }),
      scope.get_token());
sync_wait(scope.join());

The commented-out line fails for two reasons. The task finds no start scheduler in the environment spawn gives it. And spawn’s receiver accepts only set_value() and set_stopped() ([exec.spawn]), while a default task<void> can also complete with set_error(std::exception_ptr). starts_on fixes the first and upon_error the second.

The C++26 std::execution::task#

Until C++26, the standard had no task type, so code that combined coroutines with senders used one from a library or wrote its own. P3552 surveys several of them (cppcoro, libunifex, stdexec’s own exec::task) and standardizes one: std::execution::task<T, Environment>.

Apart from no_affinity, the task examples so far use its defaults: the three completion signatures, scheduler affinity, and a stop request unwinding the coroutine. The second template parameter is a struct whose members change those defaults, and a missing member means the default ([task.class]).

Controlling the error signatures#

The default says a task fails only with an std::exception_ptr. A task that can also fail with an std::error_code lists it in error_types, next to the std::exception_ptr. Its value channel is unchanged, and the body completes with the error code through co_yield with_error{...}, without throwing:

struct io_errors {
  using error_types = completion_signatures<set_error_t(std::error_code),
                                            set_error_t(std::exception_ptr)>;
};

task<int, io_errors> read_config(bool fail)
{
  if (fail) co_yield with_error{std::make_error_code(std::errc::timed_out)};
  co_return 42;
}

It is co_yield rather than co_return for a mechanical reason P3552 gives: the promise of a task<void> has return_void(), and, when P3552 was written, a promise could not have both that and a return_value(). P3950 by Robert Leahy, adopted in June 2026 as a Defect Report, lifts that restriction. The working draft still spells the error co_yield with_error{...}.

Used as a sender, the error keeps its type. An upon_error after read_config(true) is called with an std::error_code whose message is Connection timed out. Awaited from another coroutine, the same error is thrown at the co_await as an std::system_error.

A version that awaits just_error instead completes with a different error type:

task<int, io_errors> read_config_wrong()
{
  co_await just_error(std::make_error_code(std::errc::timed_out));
  co_return 0;
}

The co_await throws std::system_error, the exception escapes the body, and the task completes with set_error(std::exception_ptr). The std::error_code entry in error_types is never used, and an upon_error after it receives an std::exception_ptr. co_yield with_error{...} is the only way the standard provides for the body to complete with a typed error directly.

If error_types leaves out std::exception_ptr, the task has no way to report an exception. When one escapes the body anyway, the promise’s unhandled_exception() calls std::terminate ([task.promise]).

Allocating the frame#

The intro to this series said coroutine frames often force a heap allocation. Counting calls to the global operator new makes it concrete: a sender pipeline of two then steps run with sync_wait allocates nothing, while a task that awaits two other tasks allocates three times in stdexec, once per coroutine call. task can take its frames from an allocator you supply instead. Allocators, stop tokens, and the rest of what a receiver’s environment carries get a post of their own, which comes back to task and shows how.

Reach for senders when:

  • the code is on a hot path, or has to allocate nothing: a well-built pipeline is one operation state, and the counts above show zero allocations
  • you are writing generic library code whose callers pick the scheduler and the context
  • you want the whole computation to be a value the compiler can see and inline

The costs are deep types, long compile times, and error messages that run to pages.

Reach for coroutines when the control flow has loops and branches, like the retry above, or when a try/catch is the natural way to say what happens on failure. It is also usually the better choice for application code that runs often but not in a tight loop, where a frame allocation per call is an acceptable price for code that reads top to bottom.

In practice, you use both, with the low-level operations written as senders and the logic that strings them together written as coroutines. A coroutine task type is also the easiest way to write a new asynchronous algorithm. If the allocations and indirect calls a coroutine brings are acceptable for it, you rarely need to write a sender, a receiver, or an operation state by hand.

The next post turns to data parallelism: bulk, execution policies, and running the same pipeline on a thread pool or a GPU. How task is built, and how to teach a coroutine type of your own to await senders, get a post of their own once the series has covered receivers and operation states. Coroutines themselves, the frames, promises, and awaiters this post used without opening up, get their own series later.

Mastering std::execution covers the tradeoff between the two models and when to combine them. For teams that want the coroutine machinery as well, Structured Concurrency in C++ combines the coroutine and the senders and receivers material in one class of three to four days.