组合协程

Question

我正在学习 c++ 20 协程。有些事情还不清楚，尽管我已经看过几个关于这个话题的视频。

我正在尝试获得一个可组合的协程。我已经了解了这样做的对称传输，这是我的代码（请原谅长示例）：

#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;
}

我期望这个流程：

你好 -> ,_ -> 世界 -> !!

但我得到的是：

你好 -> ,_ , -> !!

尽管我仔细阅读了有关最终等待者、继续句柄、await_suspend 返回的恢复继续的 coro 句柄，但它不起作用。

有没有协程专家解释我哪里错了？

谢谢

Answer 1

你永远不会恢复协程suspending_2。这就是为什么你的协程管理会出现内存泄漏的原因。

当您co_await suspending_2 时，您将promise::continuation_ 设置为composed_2 句柄。

基本上，你写的是，当你暂停compose_2时，获取suspending_2的句柄，并在suspending_2完成时恢复它。这是没有意义的，因为它永远不会结束。您必须做的是在恢复composed_2 之前恢复suspending_2。基本上，恢复composed_2必须恢复suspending_2。

为此，您可以将suspending_2 的句柄保存在composed_2 的promise 中，并在必要时恢复它。

这是我从您的代码中修改的函数：

你的简单 Awaitable 和 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 {}
};

还有你的简历功能

  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();
    }
  }

不过，还是有内存泄漏，我让你自己处理。 这不是直截了当，但我鼓励您在做一些复杂的事情之前尝试更好地了解协程的工作原理:)。

我也劝你不要这样管理std::coroutine_handle，把它们封装成一个任务，不要交给其他任务（就像现在这样）。只需将任务传递给其他任务以形成延续链:)。

如果您还有其他问题，请不要犹豫。