cancellation

[mitsubishirgb]

Can someone just give me instructions how to cancel awaitables and what responsabilities does the caller have, i looked up the doc but it was lacking

struct CustomCancel {
    asio::cancellation_signal& signal_;

    CustomCancel(asio::cancellation_signal& signal) : signal_(signal) {}
    void operator()(asio::cancellation_type type = asio::cancellation_type::all) const {
        std::cout << "cancelled " << std::endl;
        signal_.emit(type);
    }
};

template<typename T>
struct Future {
    asio::cancellation_signal cancel_sig;
    asio::cancellation_slot cl;
    std::exception_ptr ex;

    bool await_ready() const noexcept { return done_; }

    template<typename Promise>
    void await_suspend(std::coroutine_handle<Promise> h) noexcept {
        //Check if the promise has a cancellation slot and attach a handler if available
        if constexpr (requires { h.promise().get_cancellation_slot(); })
            if ((cl = h.promise().get_cancellation_slot()).is_connected()) {
                cl.emplace<CustomCancel>(cancel_sig);
                std::cout << "Emplaced " << std::endl;

            }
        std::cout << "handle is set" << std::endl;
        handle_.reset(h.address());
    }


    auto await_resume() {
        
        if (cl.is_connected()) {
            cl.clear();
            
        }
        if (cancel_sig.slot().is_connected()) {
            throw std::runtime_error("Operation was cancelled");
        }
        if (ex) {
            std::rethrow_exception(ex);
        }

        

        return *value_;
    }

    template<typename Ty_>
    void set_value(Ty_&& value) {

        if (!done_ && handle_ && !handle_.done()) {

            value_ = value;
            done_ = true;
            auto h = std::move(handle_);
            handle_.reset();
            h.resume();
        }
        else {
            throw std::runtime_error("Either no handle or value is set");
        }
    }

    T get_value() const {

        if (done_)
            return *value_;
        else {
            throw std::runtime_error("Future value not set");
        }
    }

    bool done() const noexcept {
        return done_;
    }


    void interrupt_await()& {
        ex = cobalt::detail::detached_exception();
        handle_.release().resume();
    }
private:
    cobalt::unique_handle<void> handle_ = nullptr;
    std::optional<T> value_;
    bool done_{ false };
};

[klemens-morgenstern]

You will need to be more specific - what do you want to do?

[mitsubishirgb]

i have an async operation on the main function that wait for signals and cancels the promises , but how do i design the future to cancel its handler too, or is it handled implicitly somehow ?

void signal_handler(const system::error_code& ec, asio::signal_set* signals, std::vector<cobalt::promise<ClientResponse>>& tasks) {
    if (ec) {
        std::cerr << "Error occurred while handling signal: " << ec.message() << std::endl;
        return;
    }

    // Signal received; cancel all tasks
    std::cout << "SIGINT received! Cancelling all tasks..." << std::endl;

    // Cancel all tasks
    for (auto& task : tasks) {
        task.cancel();
    }

    //signals->cancel();

}

asio::signal_set signal(co_await this_coro::executor, SIGINT);
signal.async_wait([&signal, &tasks](const boost::system::error_code& ec, int signum) {
    if (!ec) {
        // Handle signal (SIGINT)
        signal_handler(ec, &signal, tasks);
    }
    });

[mitsubishirgb]

This is a snippet of my project if the coro is called more than the limit it queues the await up until a set_future() is called from other part of the code

cobalt::promise<Connection> connect(const ClientRequest& req, system::error_code& ec) {
    ConnectionKey conn_key = req.conn_key;
    
    if (acquired_ >= limit_) {
        Future<std::monostate> future;
        waiters_[conn_key].push(&future);
        co_await future;
    }

    // IF ANY CACHED CONNECTION IN THE POOL
    if (auto proto = get_connection(conn_key); proto != nullptr) {
        if (proto) std::cout << "Found a connection in the pool" << std::endl;
        co_return Connection(this, std::move(proto), &conn_key);

        
    } else //std::cout << "No connection in the pool. Creating new one " << std::endl;

    acquired_++;
    ProtoPtr proto = co_await create_connection(req, ec);
    Connection conn(this, std::move(proto), &conn_key);
    if (ec) {
        release_acquired();
        release_waiters();
        co_return Connection();
    }

    co_return conn;
}

[mitsubishirgb]

You will need to be more specific - what do you want to do?

.

[klemens-morgenstern]

I see, you're like me, you'd rather write 50 lines of code than two sentences in english explaining the intent :).

So let's see:

cobalt::promise<Connection> connect(const ClientRequest& req, system::error_code& ec) {
    ConnectionKey conn_key = req.conn_key;
    
    if (acquired_ >= limit_) {
        Future<std::monostate> future;
        waiters_[conn_key].push(&future);
        co_await future; // <1>
    }

The co_await future expression makes the future the awaitee, and connect the awaiter in my lingo.

So you got two cases here:

1. the awaiter (connect) gets cancelled by it's awaiter. This happens when you call .cancel() on a task/promise.

The co_await will forward this to the co_await future. So the design question for you is: what does a co_await future do when the awaiter cancels? In cobalt this usually produces an asio::error::operation_cancelled in the form of an error_code or system_error if thrown. I.e. your implementation of await_resume needs to handle this. For example, CustomCancel could just assign an error to ex.

However, you might want to forward the cancellation signal in some way.

2. The future gets cancelled.

This doesn't seems to be in the code you provided, but in this case you'll have two choices: do the same thing as above, i.e. produce an error. The awaiting coroutine can check with co_await this_coro::cancelled if it was cancelled or if the future was cancelled. Or you can just destroy the coroutine, so it will never get resumed. The latter is not really cancellation, but rather teardown.

Note that the cobalt::main co_main(int, char ***) does handle SIGINT as cancellation for you. But cancellation only gets forwarded when you co_await something.

[mitsubishirgb]

I see, you're like me, you'd rather write 50 lines of code than two sentences in english explaining the intent :).

So let's see:

cobalt::promise connect(const ClientRequest& req, system::error_code& ec) {
ConnectionKey conn_key = req.conn_key;

if (acquired_ >= limit_) {
    Future<std::monostate> future;
    waiters_[conn_key].push(&future);
    co_await future; // <1>
}

The co_await future expression makes the future the awaitee, and connect the awaiter in my lingo.

So you got two cases here:

1. the awaiter (connect) gets cancelled by it's awaiter. This happens when you call .cancel() on a task/promise.

The co_await will forward this to the co_await future. So the design question for you is: what does a co_await future do when the awaiter cancels? In cobalt this usually produces an asio::error::operation_cancelled in the form of an error_code or system_error if thrown. I.e. your implementation of await_resume needs to handle this. For example, CustomCancel could just assign an error to ex.

However, you might want to forward the cancellation signal in some way.

2. The future gets cancelled.

This doesn't seems to be in the code you provided, but in this case you'll have two choices: do the same thing as above, i.e. produce an error. The awaiting coroutine can check with co_await this_coro::cancelled if it was cancelled or if the future was cancelled. Or you can just destroy the coroutine, so it will never get resumed. The latter is not really cancellation, but rather teardown.

Note that the cobalt::main co_main(int, char ***) does handle SIGINT as cancellation for you. But cancellation only gets forwarded when you co_await something.

Your explanation was so good it clarified the vague concepts for me , thank you for spending the time, also your observation of my lack of explanation is right, it's due to my weak english and confusions with the topic as coming from python

[mitsubishirgb]

the cancellation works fine but when we turn back to the main coroutine, co_return doesn't work properly causing an abrupt deadlock. I am guessing the main loop thinks there are unfinished tasks ? output (also i don't want to handle the cancellation with exceptions) :
Application started
Coroutine cancelled
Main coroutine is cancelled
*deadlock

source code :

struct CustomCancel {
    std::coroutine_handle<> h;
    bool* canceled;
    void operator()(asio::cancellation_type) const {
        *canceled = true;
        if (h && !h.done()) h.resume();
    }
};

template<typename T>
struct Future {
    // Simplified: holds a value, cancellation state, and coroutine handle
    T* value_ = nullptr;
    bool canceled = false;
    std::coroutine_handle<> handle_;
    asio::cancellation_slot cl;

    auto await_ready() { return false; }
    auto await_resume() {
        if (canceled) return T{};
        return *value_;
    }
    template<typename Promise>
    void await_suspend(std::coroutine_handle<Promise> h) {
        handle_ = h;
        if constexpr (requires { h.promise().get_cancellation_slot(); }) {
            if ((cl = h.promise().get_cancellation_slot()).is_connected()) {
                cl.emplace<CustomCancel>(h, &canceled);
            }
        }
    }
};


cobalt::promise<void> coro() {
    Future<int> fut;
    int value = co_await fut;

    auto state = co_await cobalt::this_coro::cancellation_state;
    if (state.cancelled() == asio::cancellation_type::total) {
        std::cout << "Coroutine cancelled " << std::endl;
        co_return;
    }
    std::cout << "Future finished with result: " << value << std::endl;
    co_return;
}