Composed coroutines

Question

I'm learning c++ 20 coroutines. Something is still not clear, nevethless I've seen several video about the topic.

I'm trying to get a composable coroutine. I've learned about the symmetric transfer to do this and here is my code (pardon for the long example):

#include <coroutine>
#include <iostream>


// forward declaration
template <class T> struct promise;

// simple wrapper to follow the workflow
struct MySuspendAlways
{                                  // (1)
    bool await_ready() const noexcept {
        std::cout << "\tMySuspendAlways::await_ready" << '\n';
        return false;
    }

    bool await_suspend(std::coroutine_handle<> handle) const noexcept {
        std::cout << "\tMySuspendAlways::await_suspend coro handle "  << handle.address() << '\n';
        return true;
    }

    void await_resume() const noexcept {
        std::cout << "\tMySuspendAlways::await_resume" << '\n';
    }
};


template < typename T>
struct SimpleAwaitable
{
    std::coroutine_handle<promise<T>> coro_;

    SimpleAwaitable(std::coroutine_handle<promise<T>> coro) :
        coro_{ coro }
    {
        std::cout << "\n\tSimpleAwaitable::coro handle = " << coro_.address();
    }

    bool await_ready() 
    { 
        std::cout << "\n\tSimpleAwaitable::await_ready coro handle " << coro_.address();
        if (!coro_.done())
        {
            return false;
        }

        return true; 
    }

    std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuationCoro)
    {
        std::cout << "\n\tSimpleAwaitable::await_suspend coro handle " << coro_.address();
        coro_.promise().set_continuation(continuationCoro);
        return coro_;
    }

    decltype(auto) await_resume()
    {
        std::cout << "\n\tSimpleAwaitable::await_resume coro handle " << coro_.address();
        return coro_.promise().value_;
    }
};



struct final_awaitable
{
    bool await_ready() noexcept { return false; };

    void await_resume() noexcept {}
    
    template < typename P>
    std::coroutine_handle<> await_suspend(std::coroutine_handle<P> finalizedCoro) noexcept
    {
        auto& continuation = finalizedCoro.promise().continuation_;
        return continuation ? continuation : std::noop_coroutine();
    }
};


// promise type
template <typename T>
struct promise
{
    T value_;
    std::coroutine_handle<> continuation_;

    auto get_return_object() noexcept
    {
        return std::coroutine_handle<promise>::from_promise(*this);
    }

    MySuspendAlways initial_suspend() noexcept
    {     
        std::cout << "\t|_Job prepared" << '\n';
        return {};
    }

    final_awaitable final_suspend() noexcept {  
        std::cout << "\t|_Job finished" << '\n';
        return {};
    }

    void unhandled_exception() noexcept
    {
        std::terminate();
    }

    void return_value(T t) noexcept { value_ = std::move(t); }
    
    void set_continuation(std::coroutine_handle<> continuation)  noexcept
    {
        std::cout << "\tset chain with coro hadle = " << continuation.address() << '\n';
        continuation_ = continuation;
    }

    auto continuation() noexcept
    {
        return continuation_;
    }
};


template < typename T>
class task_ 
{
public:
    
    using promise_type = promise<T>;

    task_(std::coroutine_handle<promise_type> handle) :
        handle_(handle)
        {}

    ~task_() 
    {
        if (handle_)
        {
            std::cout << "\n\t\tDestoy " << handle_.address();
            handle_.destroy();
        }
    }

    T value() noexcept { return handle_.promise().value_; }

    void resume() noexcept
    { 
        if (!handle_.done())
        {
            handle_.resume();
        }
    }
    
    bool is_ready() noexcept { return handle_.done(); }

    auto operator co_await() const noexcept
    {
        std::cout << "\n\tCO_AWAIT\n";
        return SimpleAwaitable<T>{handle_};
    }

 private:
     std::coroutine_handle<promise<T>> handle_;
};



task_<std::string> suspending_2()
{
    std::cout << "\n[, ]";
    co_await std::suspend_always{};
    std::cout << "\n[WORLD]";
    co_return "done";
}

task_<std::string> composed_2()
{
    std::cout << "\n[HELLO]";
    co_await suspending_2();
    std::cout << "\n[!!]";
    co_return "end";
}




int main() {
     task_<std::string> x = composed_2();
     while (!x.is_ready())
     {
         std::cout << "\nresuming...";
         x.resume();
     }
     return 0;
}

I was expected this flow:

HELLO -> ,_ -> WORLD -> !!

But what I get is:

HELLO -> ,_ , -> !!

Neverthless I've carefully read about the final awaiter, the continuation handle, the coro handle returned by await_suspend that resumes the continuation, it doesn't work.

Does any expert of coroutines explain me where I wrong ?

Thanks

Answer 1

You never resume the coroutine suspending_2. That's the reason why you get a memory leak with your coroutine management by the way.

When you co_await suspending_2, you set the promise::continuation_ with the composed_2 handle.

Basically, what you write is, when you suspend composed_2, take the handle of suspending_2, and resume it when suspending_2 is finished. That is no sense because it will never finish. What you must do is to resume suspending_2 just before resuming composed_2. Basically, resuming composed_2 must resume suspending_2.

To do that, what you can do is to save the handle of suspending_2 inside the promise of composed_2 and resume it if necessary.

Here is the function I modified from your code:

Your simple Awaitable and final awaitable.

template <typename T> struct SimpleAwaitable {
  std::coroutine_handle<promise<T>> coro_;

  SimpleAwaitable(std::coroutine_handle<promise<T>> coro) : coro_{coro} {
    std::cout << "\n\tSimpleAwaitable::coro handle = " << coro_.address();
  }

  bool await_ready() {
    std::cout << "\n\tSimpleAwaitable::await_ready coro handle "
              << coro_.address();
    if (!coro_.done()) {
      return false;
    }

    return true;
  }

  void await_suspend(std::coroutine_handle<promise<T>> continuationCoro) {
    std::cout << "\n\tSimpleAwaitable::await_suspend coro handle "
              << coro_.address();
    continuationCoro.promise().set_continuation(coro_);
  }

  decltype(auto) await_resume() {
    std::cout << "\n\tSimpleAwaitable::await_resume coro handle "
              << coro_.address();
    return coro_.promise().value_;
  }
};

struct final_awaitable {
  bool await_ready() noexcept { return false; };

  void await_resume() noexcept {}

  void await_suspend(std::coroutine_handle<> finalizedCoro) noexcept {}
};

And your resume function

  void resume() noexcept {
    std::cout << "resume!!:"
              << " " << handle_.address() << std::endl;
    if (handle_.promise().continuation_) {
      while (!handle_.promise().continuation_.done()) {
        std::cout << "resume!!:"
                  << " " << handle_.promise().continuation_.address()
                  << std::endl;
        handle_.promise().continuation_.resume();
      }
    }
    if (!handle_.done()) {
      handle_.resume();
    }
  }

However, there is still a memory leak that I'll let you handle by yourself. It is not to be blunt, but I encourage you to try to have a better understanding of how coroutines work before doing some complicated things like this :).

I also advise you to not manage std::coroutine_handle like that, encapsulate them into a task, and don't give them to other tasks (like is the case now). Just pass the task to other tasks to make a continuation chain :).

If you have other questions, don't hesitate.