std::execution — Senders & Receivers (C++26)

Scheduler → schedule() → Sender → | then()    |
                                   | let_value() | → connected to Receiver → operation.start()
                                   | when_all()  |

#include <execution>
#include <print>

namespace ex = std::execution;

int main() {
    // just() creates a sender that immediately provides a value
    auto snd = ex::just(42)
             | ex::then([](int x) { return x * 2; })
             | ex::then([](int x) { std::println("result: {}", x); });

    // sync_wait drives the sender on the current thread, blocking until done
    ex::sync_wait(std::move(snd));   // prints: result: 84
}

#include <execution>
#include <thread_pool>   // implementation-defined; e.g. std::static_thread_pool

namespace ex = std::execution;

int main() {
    std::static_thread_pool pool{4};     // 4-thread pool
    auto sched = pool.get_scheduler();

    auto result = ex::sync_wait(
        ex::schedule(sched)              // start on pool
        | ex::then([]{ return compute_heavy_thing(); })
        | ex::then([](int v){ return v + 1; })
    );

    // result is std::optional<std::tuple<int>>
    if (result) std::println("got: {}", std::get<0>(*result));
}

// Run work described by `snd` on `sched`, then return to the calling context
auto snd = ex::on(sched, ex::just(42) | ex::then(heavy_fn));

auto pipeline =
    ex::just(std::vector<int>{1, 2, 3, 4, 5})
    | ex::then([](auto v){ std::ranges::sort(v); return v; })
    | ex::then([](auto v){ return v.back(); })   // largest element
    | ex::then([](int n){ std::println("max: {}", n); });

ex::sync_wait(std::move(pipeline));

namespace ex = std::execution;

auto fetch_user  = ex::on(io_sched, fetch_from_db(user_id));
auto fetch_prefs = ex::on(io_sched, fetch_from_db(prefs_id));
auto fetch_cache = ex::on(cache_sched, read_cache(user_id));

auto combined = ex::when_all(
    std::move(fetch_user),
    std::move(fetch_prefs),
    std::move(fetch_cache)
) | ex::then([](User u, Prefs p, Cache c) {
    return merge(u, p, c);
});

auto result = ex::sync_wait(std::move(combined));

auto pipeline =
    ex::just(user_id)
    | ex::let_value([&](int id) {
        // Returns a sender — the outer pipeline waits for this
        return ex::on(db_sched, query_user(id));
      })
    | ex::let_value([&](User u) {
        return ex::on(db_sched, query_permissions(u.role));
      })
    | ex::then([](Permissions p) {
        return p.can_write;
      });

auto safe_pipeline =
    ex::just(risky_input)
    | ex::then(parse_int)               // may set_error with std::errc
    | ex::upon_error([](std::errc e) {
        std::println("parse failed: {}", std::make_error_code(e).message());
        return 0;                        // recover with default value
      })
    | ex::then(process);

auto cancellable =
    ex::schedule(sched)
    | ex::then([](ex::stop_token st) {
        // Check for stop inside long-running work
        while (!st.stop_requested()) {
            do_chunk_of_work();
        }
    });

auto parallel_sum =
    ex::just(std::span<const double>{data})
    | ex::bulk(data.size(), [](std::size_t i, std::span<const double> d) {
        // runs in parallel; no shared writes here
        partial_sums[i] = d[i] * d[i];
      })
    | ex::then([](std::span<const double>) {
        return std::ranges::fold_left(partial_sums, 0.0, std::plus{});
      });

// A receiver that prints the result
struct PrintReceiver {
    // Called on success
    friend void tag_invoke(ex::set_value_t, PrintReceiver&&, int value) {
        std::println("value: {}", value);
    }
    // Called on error (std::exception_ptr by convention)
    friend void tag_invoke(ex::set_error_t, PrintReceiver&&, std::exception_ptr ep) {
        try { std::rethrow_exception(ep); }
        catch (const std::exception& e) { std::println("error: {}", e.what()); }
    }
    // Called on cancellation
    friend void tag_invoke(ex::set_stopped_t, PrintReceiver&&) {
        std::println("stopped");
    }
};

// Connect and start manually (normally done by sync_wait or start_detached)
auto op = ex::connect(ex::just(42), PrintReceiver{});
ex::start(op);   // prints: value: 42

# vcpkg
find_package(stdexec CONFIG REQUIRED)
target_link_libraries(my_target PRIVATE stdexec::stdexec)

// With stdexec, namespace is stdexec:: not std::execution::
// but usage is identical
#include <stdexec/execution.hpp>
namespace ex = stdexec;