2021SC@SDUSC
因为brpc使用的是epoll的边缘触发,所以将fd设置为非阻塞,然后设置socket的send,recv buffer大小,然后将当前fd加入到event_dispatcher
在GetGlobalEventDispatcher中,会只进行一次初始化dispatcher的工作,会创建FLAGS_event_dispatcher_num个dispatcher,默认为1,构造函数中会创建epoll fd,然后调用Start
在Start中会启动一个bthread,执行RunThis,RunThis会执行dispatcher的Run
Run中就是循环进行epoll_wait,
然后对判断事件,如果是EPOLLIN事件,那么执行StartInputEvent,如果是EPOLLOUT,则执行HandleEpollOut
到现在为止,dispatcher的Start就执行结束了,然后回到Socket的ResetFileDescriptor,会执行AddConsumer,将socket_id存到epoll_event的data中,然后注册EPOLLIN事件,到这里,server端初始化过程就完成了,接下来就坐等请求到来。
然后回到epoll_wait,当有新请求到来时,epoll_wait返回,遍历每一个事件,执行Socket::StartInputEvent;注册epoll事件时将socketid注册在了epoll_data的u64中,所以首先通过SocketId address到Socket,为了保证一个fd上只有一个bthread在处理,这里引入了一个atomic变量_nevent,通过_nevent判断当前socket是否有bthread正在处理,如果有的话就什么都不做,因为正在处理的线程执行完后会执行新事件,如果没有的话就使用bthread_start_urgent启动一个bthread执行ProcessEvent来处理新消息,此时epoll bthread让出当前worker的处理,worker执行新建的ProcessEvent bthread,而epoll bthread则被steal到另一个worker线程执行。这里就是官网中所说,brpc不区分io线程和worker线程,epoll bthread不负责数据的读取,IO线程的问题在于一个线程同时只能读一个fd,当多个繁忙的fd聚集在一个IO线程中时,一些读取就被延迟了;另外epoll bthread让出当前worker给ProcessEvent bthread,这样使其有更好的cache locality,可以尽快地读取fd上的数据。
ProcessEvent则是调用on_edge_triggered_events,在Socket Create的时候将该回调函数设置为了OnNewConnections。
这里会通过MoreReadEvents循环判断当前fd上是否还有新的事件,然后执行OnNewConnectionsUntilEAGAIN,这里判断是否有新事件的方法就是上文提到的_nevent。
在OnNewConnectionsUntilEAGAIN中,首先通过accept拿到一个已完成的连接,然后从监听socket中获取user,即之前的acceptor,然后创建Socket,回调函数设置为OnNewMessages,接下来的过程和之前创建监听Socket过程一样,执行AddConsumer,将socket_id存到epoll_event的data中,注册EPOLLIN事件,然后这个新连接有数据来了之后会在epoll_wait返回,执行ProcessEvent,调用到OnNewMessages。
InputMessenger负责从fd上切割和处理消息,它通过用户回调函数理解不同的格式。Parse一般是把消息从二进制流上切割下来,运行时间较固定;Process则是进一步解析消息(比如反序列化为protobuf)后调用用户回调,时间不确定。若一次从某个fd读取出n个消息(n > 1),InputMessenger会启动n-1个bthread分别处理前n-1个消息,最后一个消息则会在原地被Process。InputMessenger会逐一尝试多种协议,由于一个连接上往往只有一种消息格式,InputMessenger会记录下上次的选择,而避免每次都重复尝试。
可以看到,fd间和fd内的消息都会在brpc中获得并发,这使brpc非常擅长大消息的读取,在高负载时仍能及时处理不同来源的消息,减少长尾的存在。
这里的handler就是一开始初始化注册的所有server端协议。进入while循环,计算一次读取数据的长度,DoRead会执行_read_buf.append_from_file_descriptor,_read_buf是IOPortal类型,如上篇博客所讲,这个方法会调用readv将fd中的数据读入到iobuf的block中。
因为这一次数据读取可能会包含多个消息,因此下面会有另一个while循环,每次调用CutInputMessage尝试从iobuf中切割一条消息,如上文,server端是支持多协议的,所以这里第一次会尝试使用所有的协议进行一次parse,因为大多数情况下一个连接上只有一种协议,因此尝试一遍之后会记录下来执行成功的协议,之后将首先尝试记录的协议。
这里的parse就是上文提到的注册协议中的parse方法,这里以协议baidu_std为例简单介绍一下,baidu_std官方介绍在这里https://github.com/apache/incubator-brpc/blob/master/docs/cn/baidu_std.md,可以看到判断是否是baidu_std的方法就是判断前4 个字节是否为”prpc”,这里的copy_to为显式拷贝,因为在判断是否为baidu_std协议的过程中不能消费数据,否则可能会影响其他协议解析;如果iobuf中的数据不够4个字节且是”prpc”的前缀,那么返回PARSE_ERROR_NOT_ENOUGH_DATA错误,这个表示到目前为止不违反当前协议,但是数据不足一个消息,因此会触发重新DoRead的过程;如果前4个字节就是”prpc”,那么满足baidu_std协议,接下来将解析包体长度和包体中的元数据包长度存入到body_size和meta_size中,如果iobuf中数据长度不足body_size + 12(4 + 4 + 4),那么同样会触发重读,否则将iobuf中的meta和其他数据切割到msg中并返回,这里是零拷贝。
如果一次DoRead读入了n条消息,那么前n-1条消息会通过QueueMessage后台启动了n-1个bthread进行处理,而最后一个消息会被析构函数RunLastMessage执行,原地执行process函数,即协议中的process。
还是以baidu_std为例,首先是从iobuf中解析pb格式的meta,
然后创建req,res,Controller,然后进行流控,解析req,如果有attachment也解析出来,最后调用CallMethod,CallMethod方法在编译protobuf时生成,会调用到用户定义的Echo方法
最后在Echo中done_guard析构时会调用done的Run方法,发送response到client。