Fuchsia 中的异步事件库 — async

事件是重要的概念，是指系统硬件或软件的状态出现任何重大改变，程序时刻都在触发和接收着各种事件：鼠标点击事件，键盘事件，socket 数据收发等等

我们常见的事件触发机制模型有两种：

一种是轮询机制模型，一种是事件驱动模型

在 Linux 下，事件驱动模型主要是由 poll、epoll 实现，Fuchsia 也提供了类似的事件驱动机制，其提供 async 库使用

Epoll 的使用

先来回忆一下 Linux 下 Epoll + socket 的典型 C/S 架构的具体使用

  1. // 创建Epoll实例
  3. int epfd = epoll_crete(100);
  5. // 添加需要监听事件和监听对象
  7. epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &listen_event);
  11. while(1)
  13. {
  15.     // 阻塞中
  17.     int active_cnt = epoll_wait(epfd, events, 1000 ，-1);
  19.     // 对监听到的事件处理
  21.     for(i=0; i< active_cnt; i++)
  23.     {
  25.         // 新到连接
  27.         if(events[i].data.fd == listen_fd)
  29.         {
  31.             connfd = accept(listen_fd,(sockaddr *)&clientaddr, &clilen);             ev.data.fd=connfd;
  33.             ev.events=EPOLLIN|EPOLLET;
  35.             epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);//将新的fd添加到epoll的监听队列中
  37.         }
  39.         else if( events[i].events & EPOLLIN ) // 接收到数据，读socket
  41.         {
  43.             n = read(sockfd, line, MAXLINE)) < 0;
  45.             ev.data.ptr = md;     // 添加数据
  47.             ev.events=EPOLLOUT|EPOLLET;
  49.             epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);//修改标识符，等待下一个循环时发送数据，异步处理的精髓
  51.         }
  53.         else if(events[i].events & EPOLLOUT) // 有数据待发送，写socket
  55.         {
  57.             struct myepoll_data* md = (myepoll_data*)events[i].data.ptr;    //取数据
  59.             sockfd = md->fd;
  61.             send( sockfd, md->ptr, strlen((char*)md->ptr), 0 );        //发送数据
  63.             ev.data.fd=sockfd;
  65.             ev.events=EPOLLIN|EPOLLET;
  67.             epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
  69.         }
  71.         else
  73.         {
  75.             //其他的处理
  77.         }
  79.     }
  81. }

实例 — echo

在 fuchsia/garnet/examples/fidl/echo_server_cpp 中我们看到 echo_server.cc 源码和编译配置文件

  1. #include "echo_server_app.h"
  5. #include <lib/async-loop/cpp/loop.h>
  7. #include <lib/async-loop/default.h>
  9. #include <string>
  13. int main(int argc, const char** argv) {
  15.   async::Loop loop(&kAsyncLoopConfigAttachToCurrentThread);
  17.   bool quiet = (argc >= 2) && std::string("-q") == argv[1];
  19.   echo::EchoServerApp app(quiet);
  21.   loop.Run();
  23.   return 0;
  25. }
  26. executable("bin") {
  28.   output_name = "echo_server_cpp"
  32.   sources = [ "echo_server.cc" ]
  36.   deps = [
  38.     ":lib",
  40.     "//zircon/system/ulib/async-default",
  42.     "//zircon/system/ulib/async-loop:async-loop-cpp",
  44.     "//zircon/system/ulib/async-loop:async-loop-default",
  46.   ]
  48. }

编译配置文件中引入 async 异步事件库作为 component 依赖，在主程序头文件引入相关声明

在 echo_server_cpp.cc 中先创建 async 对象，配置好相关信息后调用 loop.Run(); 进入死循环一直等待事件的发生（除非使用 loop.quit 主动退出监听循环）

相对于 epoll，async 库的 loop 使用屏蔽了许多细节，接下来我们深入源码层级，来看看 async 库中 loop 做了什么。

Dive into ansyc source code

async-loop-cpp 和 async-loop-defaut 库均来自 fuchsia/zircon/system/ulib/async-loop 的编译

  1. // 核心数据结构 -- async_loop_t对象定义
  3. typedef struct async_loop {
  5.   async_dispatcher_t dispatcher;  // must be first (the loop inherits from async_dispatcher_t)
  7.   async_loop_config_t config;     // immutable
  9.   zx_handle_t port;               // immutable
  11.   zx_handle_t timer;              // immutable
  15.   _Atomic async_loop_state_t state;
  17.   atomic_uint active_threads;  // number of active dispatch threads
  21.   mtx_t lock;                  // guards the lists and the dispatching tasks flag
  23.   bool dispatching_tasks;      // true while the loop is busy dispatching tasks
  25.   list_node_t wait_list;       // most recently added first
  27.   list_node_t task_list;       // pending tasks, earliest deadline first
  29.   list_node_t due_list;        // due tasks, earliest deadline first
  31.   list_node_t thread_list;     // earliest created thread first
  33.   list_node_t irq_list;        // list of IRQs
  35.   list_node_t paged_vmo_list;  // most recently added first
  37.   bool timer_armed;            // true if timer has been set and has not fired yet
  39. } async_loop_t;
  40. // loop对象
  42. class Loop {
  44.  public:
  46.   explicit Loop(const async_loop_config_t* config);
  50.   Loop(const Loop&) = delete;
  52.   Loop(Loop&&) = delete;
  54.   Loop& operator=(const Loop&) = delete;
  56.   Loop& operator=(Loop&&) = delete;
  60.   ~Loop();
  64.   async_loop_t* loop() const { return loop_; }
  68.   async_dispatcher_t* dispatcher() const { return async_loop_get_dispatcher(loop_); }
  72.   void Shutdown();
  76.   zx_status_t Run(zx::time deadline = zx::time::infinite(), bool once = false);
  80.   zx_status_t RunUntilIdle();
  84.   void Quit();
  88.   zx_status_t ResetQuit();
  92.   async_loop_state_t GetState() const;
  96.   zx_status_t StartThread(const char* name = nullptr, thrd_t* out_thread = nullptr);
  100.   void JoinThreads();
  104.  private:
  106.   async_loop_t* loop_;
  108. };
  109. // 在loop的构造函数里 使用async_loop_create创建zx_status_t实例
  111. Loop::Loop(const async_loop_config_t* config) {
  113.   zx_status_t status = async_loop_create(config, &loop_);
  115.   ZX_ASSERT_MSG(status == ZX_OK, "status=%d", status);
  117. }
  118. // Run方法调用async_loop_run
  120. zx_status_t Loop::Run(zx::time deadline, bool once) {
  122.   return async_loop_run(loop_, deadline.get(), once);
  124. }
  128. zx_status_t async_loop_run(async_loop_t* loop, zx_time_t deadline, bool once) {
  130.   ZX_DEBUG_ASSERT(loop);
  134.   zx_status_t status;
  136.   atomic_fetch_add_explicit(&loop->active_threads, 1u, memory_order_acq_rel);
  138.   do {
  140.     status = async_loop_run_once(loop, deadline);
  142.   } while (status == ZX_OK && !once);
  144.   atomic_fetch_sub_explicit(&loop->active_threads, 1u, memory_order_acq_rel);
  146.   return status;
  148. }

创建 async_loop_t 对象

接下来，我们深入理解创建 async_loop_t 对象

  1. zx_status_t async_loop_create(const async_loop_config_t* config, async_loop_t** out_loop) {
  3.   ZX_DEBUG_ASSERT(out_loop);
  5.   ZX_DEBUG_ASSERT(config != NULL);
  7.   // If a setter was given, a getter should have been, too.
  9.   ZX_ASSERT((config->default_accessors.setter != NULL) ==
  11.             (config->default_accessors.getter != NULL));
  15.   async_loop_t* loop = calloc(1u, sizeof(async_loop_t));
  17.   if (!loop)
  19.     return ZX_ERR_NO_MEMORY;
  21.   atomic_init(&loop->state, ASYNC_LOOP_RUNNABLE);
  23.   atomic_init(&loop->active_threads, 0u);
  27.   loop->dispatcher.ops = (const async_ops_t*)&async_loop_ops;
  29.   loop->config = *config;
  31.   mtx_init(&loop->lock, mtx_plain);
  33.   list_initialize(&loop->wait_list);
  35.   list_initialize(&loop->irq_list);
  37.   list_initialize(&loop->task_list);
  39.   list_initialize(&loop->due_list);
  41.   list_initialize(&loop->thread_list);
  43.   list_initialize(&loop->paged_vmo_list);
  47.   zx_status_t status =
  49.       zx_port_create(config->irq_support ? ZX_PORT_BIND_TO_INTERRUPT : 0, &loop->port);
  51.   if (status == ZX_OK)
  53.     status = zx_timer_create(ZX_TIMER_SLACK_LATE, ZX_CLOCK_MONOTONIC, &loop->timer);
  55.   if (status == ZX_OK) {
  57.     *out_loop = loop;
  59.     if (loop->config.make_default_for_current_thread) {
  61.       ZX_DEBUG_ASSERT(loop->config.default_accessors.getter() == NULL);
  63.       loop->config.default_accessors.setter(&loop->dispatcher);
  65.     }
  67.   } else {
  69.     // Adjust this flag so we don't trip an assert trying to clear a default dispatcher we never
  71.     // installed.
  73.     loop->config.make_default_for_current_thread = false;
  75.     async_loop_destroy(loop);
  77.   }
  79.   return status;
  81. }

可以看到，我们先是为 async_loop_t 分配空间，接着初始化字段与监听对象、执行的需要异步任务、需要执行的超时的异步任务、中断请求等相关链表。接着创建监听端口，定时器等配置。最后一步返回创建结果。

事件监听循环与分发

Loop::Run 方法调用 async_loop_run 函数，这个函数关键部分在这里

  1.   do {
  3.     status = async_loop_run_once(loop, deadline);
  5.   } while (status == ZX_OK && !once);

这个循环读取底层上报事件循环体，async_loop_run_once 每次阻塞地读取一个事件，接着解析事件包，分发给对应函数进行处理

  1. static zx_status_t async_loop_run_once(async_loop_t* loop, zx_time_t deadline) {
  3.   async_loop_state_t state = atomic_load_explicit(&loop->state, memory_order_acquire);
  5.   if (state == ASYNC_LOOP_SHUTDOWN)
  7.     return ZX_ERR_BAD_STATE;
  9.   if (state != ASYNC_LOOP_RUNNABLE)
  11.     return ZX_ERR_CANCELED;
  15.   zx_port_packet_t packet;
  17.   zx_status_t status = zx_port_wait(loop->port, deadline, &packet);
  19.   if (status != ZX_OK)
  21.     return status;
  25.   if (packet.key == KEY_CONTROL) {
  27.     // Handle wake-up packets.
  29.     if (packet.type == ZX_PKT_TYPE_USER)
  31.       return ZX_OK;
  35.     // Handle task timer expirations.
  37.     if (packet.type == ZX_PKT_TYPE_SIGNAL_ONE && packet.signal.observed & ZX_TIMER_SIGNALED) {
  39.       return async_loop_dispatch_tasks(loop);
  41.     }
  43.   } else {
  45.     // Handle wait completion packets.
  47.     if (packet.type == ZX_PKT_TYPE_SIGNAL_ONE) {
  49.       async_wait_t* wait = (void*)(uintptr_t)packet.key;
  51.       mtx_lock(&loop->lock);
  53.       list_delete(wait_to_node(wait));
  55.       mtx_unlock(&loop->lock);
  57.       return async_loop_dispatch_wait(loop, wait, packet.status, &packet.signal);
  59.     }
  63.     // Handle queued user packets.
  65.     if (packet.type == ZX_PKT_TYPE_USER) {
  67.       async_receiver_t* receiver = (void*)(uintptr_t)packet.key;
  69.       return async_loop_dispatch_packet(loop, receiver, packet.status, &packet.user);
  71.     }
  75.     // Handle guest bell trap packets.
  77.     if (packet.type == ZX_PKT_TYPE_GUEST_BELL) {
  79.       async_guest_bell_trap_t* trap = (void*)(uintptr_t)packet.key;
  81.       return async_loop_dispatch_guest_bell_trap(loop, trap, packet.status, &packet.guest_bell);
  83.     }
  87.     // Handle interrupt packets.
  89.     if (packet.type == ZX_PKT_TYPE_INTERRUPT) {
  91.       async_irq_t* irq = (void*)(uintptr_t)packet.key;
  93.       return async_loop_dispatch_irq(loop, irq, packet.status, &packet.interrupt);
  95.     }
  97.     // Handle pager packets.
  99.     if (packet.type == ZX_PKT_TYPE_PAGE_REQUEST) {
  101.       async_paged_vmo_t* paged_vmo = (void*)(uintptr_t)packet.key;
  103.       return async_loop_dispatch_paged_vmo(loop, paged_vmo, packet.status, &packet.page_request);
  105.     }
  107.   }
  109.   ZX_DEBUG_ASSERT(false);
  111.   return ZX_ERR_INTERNAL;
  113. }

从 packet.type 我们可以判断出 async 库支持的事件类型有定时器任务、等待 wait、data packet 事件、bell trap 事件、中断请求事件、内存共享事件

对事件的添加与取消这里以定时器为例，详细函数实现在

static zx_status_t async_loop_post_task(async_dispatcher_t* async, async_task_t* task) — 添加定时器任务

static zx_status_t async_loop_cancel_task(async_dispatcher_t* async, async_task_t* task) — 取消定时器任务

这里不过多赘述

handler

以定时器事件为例，handler 为 async_loop_dispatch_tasks，这个 handler 是由调用者在添加事件监听时，构造 async_task_t 传入

  1. static void async_loop_dispatch_task(async_loop_t* loop, async_task_t* task, zx_status_t status) {
  3.   // Invoke the handler.  Note that it might destroy itself.
  5.   async_loop_invoke_prologue(loop);
  7.   task->handler((async_dispatcher_t*)loop, task, status);
  9.   async_loop_invoke_epilogue(loop);
  11. }