Stop Tokens and Cancellation

std::atomic<bool> should_cancel{false};

void worker()
{
    while (!should_cancel)
    {
        do_work();
    }
}

#include <stop_token>
#include <iostream>

void example()
{
    std::stop_source source;

    // Create tokens (distribute notification capability)
    std::stop_token token1 = source.get_token();
    std::stop_token token2 = source.get_token();  // Same underlying state

    // Register callbacks (observers)
    std::stop_callback cb1(token1, []{ std::cout << "Observer 1 notified\n"; });
    std::stop_callback cb2(token2, []{ std::cout << "Observer 2 notified\n"; });

    std::cout << "Before signal\n";
    source.request_stop();  // Triggers all callbacks
    std::cout << "After signal\n";
}

std::stop_source source;
source.request_stop();  // Already signaled

// Callback runs in constructor, not later
std::stop_callback cb(source.get_token(), []{
    std::cout << "Runs immediately!\n";
});

std::stop_source source;
auto token = source.get_token();

// ... distribute token to workers ...

source.request_stop();  // Triggered, now permanently signaled

// To "reset": create new source
source = std::stop_source{};  // New shared state
// Old tokens still see the old, already-signaled state

// Must redistribute new tokens to ALL holders of the old token
auto new_token = source.get_token();

capy::task<> parent()
{
    // Our stop token is automatically passed to child
    co_await child();
}

capy::task<> child()
{
    // Receives parent's stop token via ContextAwaitable protocol
    auto token = co_await capy::this_coro::stop_token;  // Access current token
}

capy::task<> cancellable_work()
{
    auto token = co_await capy::this_coro::stop_token;

    while (!token.stop_requested())
    {
        co_await do_chunk_of_work();
    }
}

capy::task<> process_items(std::vector<Item> const& items)
{
    auto token = co_await capy::this_coro::stop_token;

    for (auto const& item : items)
    {
        if (token.stop_requested())
            co_return;  // Exit early

        co_await process(item);
    }
}

capy::task<> with_resource()
{
    auto resource = acquire_resource();  // RAII wrapper
    auto token = co_await capy::this_coro::stop_token;

    while (!token.stop_requested())
    {
        co_await use_resource(resource);
    }
    // resource destructor runs regardless of how we exit
}

capy::task<std::string> fetch_with_cancel()
{
    auto token = co_await capy::this_coro::stop_token;

    if (token.stop_requested())
    {
        throw std::system_error(
            capy::make_error_code(capy::error::canceled));
    }

    co_return co_await do_fetch();
}

// WRONG: the completion and the stop callback can both resume the
// coroutine.
struct racy_read
{
    struct cancel_fn
    {
        racy_read* self;

        void operator()() const noexcept
        {
            self->canceled_ = true;
            self->ex_.post(self->cont_);  // even if the read completed
        }
    };

    capy::continuation cont_;
    capy::executor_ref ex_;
    bool canceled_ = false;
    std::optional<std::stop_callback<cancel_fn>> stop_cb_;

    bool await_ready() const noexcept { return false; }

    std::coroutine_handle<> await_suspend(
        std::coroutine_handle<> h, capy::await_context const* env)
    {
        cont_.h = h;
        cont_.owner = env->owner;
        ex_ = env->executor;
        stop_cb_.emplace(env->stop_token, cancel_fn{this});
        start_read(this);  // the backend calls on_complete() later
        return std::noop_coroutine();
    }

    // Called by the I/O backend, on any thread, when the read finishes.
    void on_complete() noexcept
    {
        ex_.post(cont_);  // even if the stop callback posted already
    }

    void await_resume() noexcept { stop_cb_.reset(); }
};

struct fetch_channel
{
    capy::async_waker fetch_ready;
    std::atomic<bool> cancelled{false};
    std::string result;
};

// One side of the race: completes when the fetch worker thread
// wakes fetch_ready. If the deadline wins first, when_any's stop
// request cancels the wait; flag the worker so it stops early.
capy::task<capy::results<std::string>> await_fetch(fetch_channel& ch)
{
    auto [ec] = co_await ch.fetch_ready.wait();
    if (ec)
    {
        ch.cancelled.store(true);
        co_return capy::results<std::string>{ec, {}};
    }
    co_return capy::results<std::string>{std::error_code(), std::move(ch.result)};
}

// The other side of the race: completes once whatever plays the
// clock wakes the waker.
capy::task<capy::results<>> deadline(capy::async_waker& waker)
{
    auto [ec] = co_await waker.wait();
    co_return capy::results<>{ec};
}

class download_manager
{
    capy::executor_ref executor_;
    std::stop_source stop_source_;

public:
    void start_download(std::string url)
    {
        // Token propagated via await_context, not as a function argument
        capy::run_async(executor_, stop_source_.get_token())(download(url));
    }

    void cancel()
    {
        stop_source_.request_stop();
    }
};

capy::task<void> download(std::string url)
{
    // From run_async's await_context
    auto token = co_await capy::this_coro::stop_token;
    while (!token.stop_requested())
    {
        co_await fetch_next_chunk(url);
    }
}

class server
{
    std::stop_source shutdown_source_;

public:
    void shutdown()
    {
        shutdown_source_.request_stop();
        // All pending operations receive stop request
    }

    capy::task<> handle_connection(connection conn)
    {
        auto token = shutdown_source_.get_token();

        while (!token.stop_requested())
        {
            co_await process_request(conn);
        }

        // Graceful cleanup
        co_await send_goodbye(conn);
    }
};