AsyncMessenger 简析 II

在IO栈的优化中，同步改异步是很常用的一种方法，往往可以提高系统的吞吐率，Ceph Messenger就是利用Linux的阻塞异步IO机制，以及自身设计的队列结构来实现对Event的异步处理，进而获取了比Simple Messenger优越很多的性能。

Ceph的异步机制根据EventCenter::EventDriver的不同实例可以使用几种模型，本文讨论的是基于epoll机制的Ceph Async + RDMAStack下的IO路径

EventCenter

Ceph Async机制的核心即是EventCenter，下图简述了EventCenter对这一机制的实现

在Async机制中，Ceph Msg模块将所有的异步事件分为两类:

1. handler固定、”定点”执行的file_event，通常用于频繁执行的routine，
2. handler形式多样但”顺便”执行的external_event

file events（与fd绑定）

file events的核心结构是EventCenter::vector<> file_events，该vector以fd为索引封装了一组FileEvents实例，每个FileEvents实例都包含两个handler:

1)readcb用于处理fd可读，

2)writecb用于fd可写。

系统通过create_file_event()将fd及其handler注册到file_events。最终，所有的file_events注册的fd都会通过epoll_ctl()将其注册到内核(epoll)，并通过在Worker中调用epoll_wait()来实现异步阻塞IO。

external events （与fd无关）--->绑定pipe

相比于file_events与特定的fd绑定，external events要自由的多，external_events就是一个EventCallback的deque，使用方式与fd无关， 所以只要EventCallback及其子类的实例，都可以在需要的时候注册到其中以此让Worker线程执行之。值得注意的是，在Ceph的设计中，file event和external event并不是割裂的，事实上，external event依赖于在file_events中设置代理才能工作，通过将管道的读端注册为file event，dispatch_event_external()首先会将相应的EventCallback实例入队，再向管道的写端notify_send_fd写入一个字符，如此一来管道的读端notify_receive_fd就会变的可读，进而通过的epoll机制wakeup相应的Worker工作线程，Worker工作线程首先会回调发生事件的notify_receive_fd的readcb，该cb只是将在写端写入的字符读出，之后，Worker就会将external event全部回调一遍，即将其从队列清空。

Ceph Async机制

Ceph Async机制的核心即是EventCenter，下图简述了EventCenter对这一机制的实现, 在Async机制中，Ceph Msg模块将所有的异步事件分为两类:

1. handler固定、”定点”执行的file_event，通常用于频繁执行的routine，
2. handler形式多样但”顺便”执行的external_event

 

file events（与fd绑定）

file events的核心结构是EventCenter::vector<> file_events，该vector以fd为索引封装了一组FileEvents实例，每个FileEvents实例都包含两个handler:

1)readcb用于当fd可读。

2)writecb用于fd可写。

系统通过create_file_event()将(fd,handler)成对注册到file_events。最终，所有的file_events注册的fd都会通过epoll_ctl()将其注册到内核(epoll)，并通过在Worker中调用epoll_wait()阻塞在epoll等待请求到来。

external events （与fd无关）--->绑定pipe

相比于file_events中的handler与特定的fd绑定， external_events就是一个EventCallback的deque，EventCallback与fd无关， 所以只要EventCallback及其子类的实例，都可以在需要的时候注册到其中，然后唤醒worker来执行external_events里的EventCallback。

值得注意的是，在Ceph的设计中，file event和external event并不是割裂的，事实上external event依赖于在file_events中设置代理才能工作，通过创建一个管道并将读端注册到file events，当dispatch_event_external()将相应的EventCallback实例入external event队列，再wakeup相应的Worker工作线程(Worker阻塞在epoll_wait,向管道的写端notify_send_fd写入一个字符，如此一来管道的读端notify_receive_fd就会变的可读，监听notify_receive_fd的epoll的epoll_wait返回，worker往下执行)，Worker工作线程首先会回调发生事件的notify_receive_fd的readcb，该cb只是将在写端写入的字符读出，之后，Worker就会将external event中的EventCallbacks全部回调一遍，即将其从队列清空。

AsyncMessenger 简析 III

RDMAStack IO路径

下图是在使用RDMAStack时的IO路径，经过三层（Async—>NetworkStackàRDMAStack）,异步读写流程都是EventCenter在承担主要工作。

Send IO

消息发送过程可以分为几个关键流程(注意图中的序号s1……s7）：

1. AsyncMessenger::send_message(Message m) 是Msg模块提供给上层提交IO请求的接口，IO请求封装在Message中。
2. AsyncMessenger::submit_message() 根据目标地址获取AsyncConnection对象，将Message传入该AsyncConnection对象中继续处理
3. AsyncConnection::send_message(m) 是IO请求在AsyncConnection实例中的入口，该函数主要将Message放在优先级队列AsyncConnection::out_q，并调用dispatch_event_external()将write_handler入deque<EventCallbackRef> external_events队列，并wakeup() 阻塞在wait的Worker线程。
4. 当Worker线程被唤醒,从external_events取出write_handler执行时，就会调用AsyncConnection::handle_write()，该函数主要是从out_q中依照优先级将一个Message出队，并通过AsyncConnection::write_message()把Mesage中的payload、middle、data都拷贝到AsyncConnection::outcoming_bl中，并调用AsyncConnection::_try_send()。
5. AsyncConnection::_try_send()作为Send IO在Async层的最后一个接口，主要是将outcoming_bl向下层传递，最终会到达RDMAStack，调用其中的RDMAConnectedSocketImpl::send()。
6. RDMAStack内部会维持一个pending_bl，将RDMAConnectedSocketImpl::send()上层传入的outcoming_bl”接”在其后，调用的Chunk::write()启动发送。
7. ibv_post_send()是verbs中发送数据的接口，发送过程至此结束。

Receive IO

消息接收过程可以分为几个关键流程(注意图中的序号s1……s7）

在RDMA协议中，收发完成(!=成功)后，首先会在CQ(Completion Queue)中放入CQE(CQ Entry)来通知上层有事件完成，verbs标准支持轮询和通知两种机制，Ceph RDMAStack使用的是轮询机制

1. RDMADispatcher::polling线程不断的调用ibv_poll_cq()来轮询事件是否完成。
2. 如果发现一个底层读事件完成，就会通过conn = get_conn_lockless(response->qp_num)来获取其Connection对象，进而找到其关联的Worker和EventCenter，写EventCenter的notify_fd，由于该notify_fd已经注册在file_events中，此举将唤醒Worker线程。
3. EventCenter::Worker被唤醒后回调notify_fd的readcb，其中核心函数是AsyncConnection::process();
4. AsyncConnection::process()线程会申请一块buffer用于接收收到的数据，这个buffer最终会被封装到bufferlist中并进一步被封装为Message供上层使用。
5. AsyncConnection本身有一个读缓冲区：recv_buf，该buffer只给AsyncConnection::read_util()使用，until，顾名思义，就是该接口一定会读到想读取的长度，该接口首先试图从recv_buf中获取请求的数据，当recv_buf已有数据不能满足请求时，就要依情况从底层读取，对于recv_buf可以承载的数据长度，会调用底层接口先将recv_buf填满，再将从中读取所需;对于recv_buf不足的数据长度，直接从底层获取，从底层读取的接口是read_bulk()。
6. AsyncConnection::read_bulk()最终会带着传入的buffer，层层调用，直到RDMAStack中RDMAConnectedSocketImpl::read()–>Chunk::read()–>memcpy()将获取的数据填充到为Message准备号的buffer中。
7. AsyncConnection::process()经过层层调用已为Message的构造封装好了数据，接下来就只是构造一个Message，入dispatch_queue，交给上层处理。