std::execution: Senders, Receivers, and Structured Async

// Unstructured: who owns this future? When does the computation stop?
auto f1 = std::async(std::launch::async, compute_a);
auto f2 = std::async(std::launch::async, compute_b);
int result = f1.get() + f2.get();
// What if f2 throws? f1's thread is gone, its destructor is a detach.

// A sender that produces the string "hello"
std::execution::sender auto s = std::execution::just(std::string{"hello"});
// s describes work. Nothing has run yet.

using namespace std::execution;

sender auto pipeline =
    just(42)
    | then([](int x) { return x * 2; })
    | then([](int x) { return std::to_string(x); });
// Still nothing has run.

// The system_context is a thread pool shared by the process
std::execution::system_context ctx;
std::execution::scheduler auto sch = ctx.get_scheduler();

sender auto work =
    schedule(sch)                        // start on pool thread
    | then([]{ return heavy_compute(); });

// Blocks current thread until sender completes
auto result = std::this_thread::sync_wait(pipeline);
// result is optional<tuple<values...>>
// - optional<tuple<std::string>> for just(42) | then(to_string)
// - empty optional if sender completed with set_stopped()
// - throws if sender completed with set_error()

#include <execution>
using namespace std::execution;

system_context ctx;
auto pool = ctx.get_scheduler();

// io_context (conceptually — your async I/O library provides one)
// auto io   = io_ctx.get_scheduler();

sender auto work =
    just(read_request_from_socket())        // on current thread
    | continues_on(pool)                    // switch to thread pool
    | then([](Request req) {
        return process(req);                // heavy work on pool
      })
    | continues_on(io)                      // back to I/O thread
    | then([](Response resp) {
        write_response_to_socket(resp);
      });

std::this_thread::sync_wait(std::move(work));

// starts_on: the WHOLE pipeline runs starting on sch
sender auto a = starts_on(pool, just(42) | then(compute));

// continues_on: after current completion, CONTINUE on sch
sender auto b = just(42) | then(f) | continues_on(pool) | then(g);

// on: run the wrapped sender on sch, then RETURN to original scheduler
sender auto c = just(42) | then(f) | on(pool, just(1) | then(h)) | then(k);
//              ^^^^ runs on original     ^^^^ runs on pool          ^^^^ back to original

using namespace std::execution;

system_context ctx;
auto pool = ctx.get_scheduler();

sender auto fetch_a = starts_on(pool, just() | then(fetch_from_db_a));
sender auto fetch_b = starts_on(pool, just() | then(fetch_from_db_b));

sender auto combined = when_all(std::move(fetch_a), std::move(fetch_b))
    | then([](ResultA a, ResultB b) {
        return merge(a, b);
      });

auto [result] = std::this_thread::sync_wait(std::move(combined)).value();

std::vector<float> data(1024);

sender auto fill =
    just(std::span<float>{data})
    | bulk(data.size(), [](std::size_t i, std::span<float> v) {
        v[i] = static_cast<float>(i) * 0.1f;
      });

std::this_thread::sync_wait(std::move(fill));

// load_image returns a sender<Image>
// enhance_image(img) also returns a sender<Image>
// We want: load → enhance → save, where each step is async

sender auto pipeline =
    load_image("input.png")               // sender<Image>
    | let_value([](Image img) {           // img -> sender<Image>
        return enhance_image(std::move(img));
      })
    | let_value([](Image img) {           // img -> sender<void>
        return save_image(std::move(img), "output.png");
      });

sender auto safe_pipeline =
    load_from_network()               // might complete with set_error(std::error_code)
    | upon_error([](std::error_code ec) -> std::string {
        return "default_value";       // recover: turn error into a value
      })
    | then([](std::string s) { process(s); });

std::stop_source stop;

sender auto cancellable =
    starts_on(pool, long_running_work())
    | stopped_as_error(std::runtime_error{"cancelled"});

// From another thread:
stop.request_stop();

// A minimal just_int sender (simplified from ACCU Overload 186)
struct just_int_op_state {
    using operation_state_concept = std::execution::operation_state_t;

    int value;
    auto receiver;   // holds the downstream receiver

    void start() noexcept {
        std::execution::set_value(std::move(receiver), value);
    }
};

struct just_int_sender {
    using sender_concept       = std::execution::sender_t;
    using completion_signatures = std::execution::completion_signatures<
        std::execution::set_value_t(int)
    >;

    int value;

    auto get_env() const noexcept { return std::execution::empty_env{}; }

    template<std::execution::receiver_of<completion_signatures> R>
    just_int_op_state<R> connect(R&& r) && {
        return {value, std::forward<R>(r)};
    }
};

auto just_int(int v) { return just_int_sender{v}; }

include(FetchContent)
FetchContent_Declare(stdexec
    GIT_REPOSITORY https://github.com/NVIDIA/stdexec
    GIT_TAG        main
)
FetchContent_MakeAvailable(stdexec)
target_link_libraries(myapp PRIVATE stdexec::stdexec)

#include <stdexec/execution.hpp>
namespace ex = stdexec;

int main() {
    ex::thread_pool pool{4};
    auto sch = pool.get_scheduler();

    auto [result] = ex::sync_wait(
        ex::schedule(sch)
        | ex::then([]{ return 6 * 7; })
    ).value();

    // result == 42
}

// BUG: op_state destroyed before start() completes
{
    auto op = std::execution::connect(sender, receiver);
    std::execution::start(op);
} // op destroyed here — UB if start() is async