objective-c-c++20协程详解（四）

前言

到这就是协程的最后一节了。希望能帮到大家

代码

到这里我们整合下之前二、三节的代码

#include <coroutine>
#include <functional>
#include <chrono>
#include <iostream>
#include <thread>
#include <mutex>
#include <memory>
#include <vector>

struct async_task_base
{
   virtual void completed() = 0;
   virtual void resume() = 0;
};


std::mutex m;
std::vector<std::shared_ptr<async_task_base>> g_event_loop_queue; 
std::vector<std::shared_ptr<async_task_base>> g_resume_queue; //多线程异步任务完成后后，待主线程恢复的线程
std::vector<std::shared_ptr<async_task_base>> g_work_queue; //执行耗时操作线程队列

enum class EnumAwaiterType:uint32_t{
   EnumInitial = 1, //协程initial
   EnumSchduling = 2,// 用户co_await
   EnumFinal = 3//销毁
};


template <typename ReturnType>
struct CoroutineTask;

template <typename CoTask, EnumAwaiterType AwaiterType >
struct CommonAwaiter ;

template <typename>
struct AsyncAwaiter;


template <typename ReturnType>
struct  AsyncThread
{
   using return_type = ReturnType;

   AsyncThread(std::function<return_type ()>&& func): func_(func){

   }
   std::function<return_type ()> func_;
};


template <typename ReturnType>
struct async_task: public async_task_base{
   async_task(AsyncAwaiter<ReturnType> &awaiter)
   :owner_(awaiter)
   {

   }

   void completed() override{
       ReturnType result = owner_.func_();
       owner_.value_ = result;
   }

   void resume() override{
       owner_.h_.resume();
   }
   AsyncAwaiter<ReturnType> &owner_ ;
};


template <typename ReturnType>
struct AsyncAwaiter
{
   using return_type = ReturnType;

   AsyncAwaiter(AsyncThread<ReturnType>& info){
       // std::cout<< " AsyncAwaiter(AsyncThread<ReturnType>& info)" << std::endl;
       value_ = return_type{};
       func_ = info.func_;
   }

   bool await_ready() const noexcept { 
       return false; 
   }
   
   void await_suspend(std::coroutine_handle<> h)  {
       h_ = h;
       std::lock_guard<std::mutex> g(m);
       g_work_queue.emplace_back(std::shared_ptr<async_task_base>( new async_task<uint64_t>(*this)));
   }

   return_type await_resume() const noexcept { 
       return value_;
   }

 
   std::function<return_type ()> func_;
   std::coroutine_handle<> h_; 
   return_type value_ = return_type();
};


template <typename CoTask, EnumAwaiterType AwaiterType>
struct coroutine_task: public async_task_base{
   coroutine_task(CommonAwaiter<CoTask, AwaiterType> &awaiter)
   :owner_(awaiter)
   {

   }

   void completed() override{
   }

   void resume() override{
       if(owner_.h_.done()){
           owner_.h_.destroy();
       }else{
           owner_.h_.resume();
       }
   }
   CommonAwaiter<CoTask,AwaiterType> &owner_ ;
};

template <typename CoTask, EnumAwaiterType AwaiterType = EnumAwaiterType::EnumSchduling>
struct CommonAwaiter 
{
   using return_type =  typename CoTask::return_type;
   using promise_type = typename CoTask::promise_type;
   CommonAwaiter(promise_type* promise):promise_(promise){
   }

   bool await_ready() const noexcept { 
       return false;
   }

   //也可以直接恢复 
   // std::coroutine_handle<> await_suspend(std::coroutine_handle<> h)  {
   //     return h;
   // }

   void await_suspend(std::coroutine_handle<> h)  {
       h_ = h;
       g_event_loop_queue.emplace_back(std::shared_ptr<async_task_base>( new coroutine_task<CoTask, AwaiterType>(*this)) );
   }


   return_type await_resume() const noexcept { 
       return promise_->get_value();
   }

   ~CommonAwaiter(){
   }

   bool resume_ready_= false;
   promise_type* promise_ = nullptr;
   std::coroutine_handle<> h_ = nullptr;
};


template <typename CoTask>
struct CommonAwaiter<CoTask, EnumAwaiterType::EnumInitial>
{
   CommonAwaiter(){
   }

   bool await_ready() const noexcept { 
       return true;
   }

   void await_suspend(std::coroutine_handle<>)  {
   }

   void await_resume() const noexcept { 
   }

   ~CommonAwaiter(){
   }
};



template <typename CoTask>
struct CommonAwaiter <CoTask, EnumAwaiterType::EnumFinal>
{
   CommonAwaiter(){
   }

   bool await_ready() noexcept { 
       return false;
   }

   void await_suspend(std::coroutine_handle<> h)  noexcept{
       h_ = h;
       g_event_loop_queue.emplace_back(std::shared_ptr<async_task_base>( new coroutine_task<CoTask, EnumAwaiterType::EnumFinal>(*this)));
   }

   void await_resume()  noexcept{ 
   }

   std::coroutine_handle<> h_ = nullptr;
};


template<typename CoTask>
struct Promise
{
   using return_type  = typename CoTask::return_type ;
   ~Promise(){
   }
   CommonAwaiter<CoTask, EnumAwaiterType::EnumInitial> initial_suspend() {
       return {}; 
   };
   
   CommonAwaiter<CoTask, EnumAwaiterType::EnumFinal> final_suspend() noexcept { 
       return {}; 
   }

   void unhandled_exception(){
       std::rethrow_exception(std::current_exception());
   }

   CoTask get_return_object(){ 
       return  CoTask(this);
   }

   return_type get_value() {
       return value_;
   }


   void return_value(return_type value){
       value_ = value;
   }
  
   template<typename T>
   CommonAwaiter<CoroutineTask<T>> await_transform(CoroutineTask<T> &&task){
       return CommonAwaiter<CoroutineTask<T>>(task.p_);
   }

 
   template<typename T>
   inline AsyncAwaiter<T> await_transform(AsyncThread<T>&& info)
   {
       return AsyncAwaiter(info);
   }



   return_type value_;
};

template <typename ReturnType>
struct CoroutineTask{

   using return_type  = ReturnType;
   using promise_type = Promise<CoroutineTask>;

   CoroutineTask(const CoroutineTask &other) = delete;
   CoroutineTask(const CoroutineTask &&other) = delete;
   CoroutineTask& operator=(const CoroutineTask&) {};
   CoroutineTask& operator=(const CoroutineTask&&) = delete;

   CoroutineTask(promise_type* promise) {
       p_ = promise;
       
   }

   promise_type *p_ = nullptr;

};


void do_work() {
   while (1)
   {
       std::lock_guard<std::mutex> g(m);
       for(auto task : g_work_queue){
           task->completed();
           g_resume_queue.push_back(task);
       }
       
       g_work_queue.clear();
   }   
   
}

void run_event_loop(){
   std::vector<std::shared_ptr<async_task_base>> g_raw_work_queue_tmp;
   std::vector<std::shared_ptr<async_task_base>> g_event_loop_queue_temp;
   while(1){
       g_raw_work_queue_tmp.clear();
       g_event_loop_queue_temp.clear();
       {
           g_event_loop_queue_temp.swap(g_event_loop_queue);
           std::lock_guard<std::mutex> g(m);
           g_raw_work_queue_tmp.swap(g_resume_queue);
       }
       
       for(auto &task : g_raw_work_queue_tmp){
           task->resume();
       }

       for(auto task : g_event_loop_queue_temp){
           task->resume();
       }

   }
}


// ------------------------------------------------------------------------------------------------------



template <typename ReturnType>
AsyncThread<ReturnType> do_slow_work(std::function< ReturnType () > &&func){
   return AsyncThread<ReturnType>(std::forward< std::function< ReturnType () > >(func));
}


CoroutineTask<u_int64_t> second_coroutine(){
   co_return 3;
}

CoroutineTask<float> third_coroutine(){
   co_return 3.1;
}


CoroutineTask<char> first_coroutine(){
   int a = 1;
   auto func =[&]() -> uint64_t{
       // std::cout<< "do a slow work !!!!!!!!!!!!!!!!!!!!!" << std::endl;
       return a;
   };  
   uint64_t result = co_await do_slow_work<uint64_t>(func);
   std::cout << "@@@@@@@@@ result1 is  : " << result  << std::endl;  
   a = 2;
   result = co_await do_slow_work<uint64_t>(func);
   std::cout << "@@@@@@@@@ result2 is  : " << result  << std::endl; 
   uint64_t num =  co_await second_coroutine();
   std::cout << "@@@@@@@@@ second_coroutine result is  : " << num  << std::endl; 
   a = 3;
   result = co_await do_slow_work<uint64_t>(func);
   std::cout << "@@@@@@@@@ result3 is  : " << result  << std::endl;  
   float num2 =  co_await third_coroutine();
   a = 4;
   result = co_await do_slow_work<uint64_t>(func);
   std::cout << "@@@@@@@@@ third_coroutine result is  : " << num2  << std::endl; 
   result = co_await do_slow_work<uint64_t>(func);
   std::cout << "@@@@@@@@@ result4 is  : " << result  << std::endl;  
   co_return 'b';
}


CoroutineTask<char> Coroutine(){
   int a = 1;
   auto func =[&]() -> uint64_t{
       // std::cout<< "do a slow work !!!!!!!!!!!!!!!!!!!!!" << std::endl;
       return a;
   };  
   uint64_t result = co_await do_slow_work<uint64_t>(func);
   std::cout << "@@@@@@@@@ result is  : " << result  << std::endl; 
   uint64_t num =  co_await second_coroutine();
   std::cout << "@@@@@@@@@ coroutine result is  : " << num  << std::endl; 

   co_return 'b';
}


void test_func(){
   Coroutine();
   first_coroutine();
}

int main(){
   test_func();
   std::thread work_thread(do_work);
   run_event_loop();
   return 0;
}

分析

将对两种awaiter的co_await操作规则定义到promise中

对co_await可以使用await_transform 和重载co_await运算符，但是两种用法不能同时存在。

优先恢复其他线程完成耗时任务的协程，再进行当前线程中的协程挂起、销毁、恢复调度。

协程函数相对对于协程的优势

协程每次都会在final_suspend和initial_suspend时创建awaiter，以及对awaiter挂起，在for循环中，这样显然不科学，但异步函数只有一个awaiter，在for循环中更合适。

运行结果

待扩展

异步io

如果对使用epoll实现网络io异步函数感兴趣，可以自行实现，实现方式和实现多线程异步函数一样，这里就不实现了，注意下epoll中，不能添加普通文件系统fd。

协程超时机制

可以加上定时器对async_task_base进行超时检查，以此来支持协程超时

流程图

最后附上co_await Coroutine()的流程图