Hypertable源码解读之AsyncComm目录

旧城等待， 2022-08-27 08:00 173阅读 0赞

该目录下包含了所有异步通信模块的源文件，该模块是整个Hypertable系统的基础，该模块的整体思路如下：  
工厂类类ReactorFactory中所有成员变量和函数都为静态成员，其初始化时将构造reactor\_count个Reactor对象，并将每个对象委托给ReactorRunner对象托管，然后启动一个线程，线程函数即为ReactorRunner函数对象。  
Reactor对象构造时将创建一个UPD类型的socket，（m\_interrupt\_sd，中断的socket描述符），并bind到 本地地址，然后将该socket描述符加入到监控列表中。ReactorRunner函数对象将阻塞式的监听来自对应Reactor对象中所有的socket描述符上的事件，并对监听到的事件进行处理。  
事件的处理将由IOHandler类或其子类进行。IOHandler对象将在Comm类中创建，创建时将获取在Comm类中创建的socket描述符，此时将建立IOHandler对象和socket描述符的对应关系。IOHandler类构造时，还从ReactorFactory中动态的获取一个已有的Reactor对象，由此创建了IOHandler对象和Reactor对象的对应关系。创建IOHandler对象后，Comm类将开启IOHandler对象的监听状态，此时IOHandler对象将socket描述符及其监听事件加入对应的Reactor对象的监控列表。待监听的socket产生事件时，ReactorRunner函数对象便会进行处理，具体处理过程由IOHandler子类的handle\_event函数执行。  
Comm类创建socket描述符——》Comm类创建IOHandler对象，创建时将动态关联一个Reactor对象——》Comm类将socket描述符交由IOHandler对象处理——》Comm类启动IOHandler对象的监听过程——》IOHandler对象将socket描述符放入对应的Reactor对象的监听列表——》ReactorRunner函数对象调用IOHandler对象的handle\_event函数处理监听到的socket描述符和事件。

每个IOHandler对象关联到一个DispatchHandlerPtr，当IOHandler对象产生事件时，将会调用deliver\_event函数推送事件到DispatchHandlerPtr，DispatchHandlerPtr将调用handle函数负责处理或分发该事件。

**1.class Comm **  
提供异步的网络通信服务，其为单例模式。  
**1.1.成员函数  
1.1.1.构造函数**  
如果ReactorFactory中不包含reactor对象，则错误退出。该类包含一个IOHandle对象集合，名称为m\_handler\_map，其包含了所有的socket描述符对应的IOHandle对象，该类与ReactorRunner类共享一个IOHandle对象集合。  
**1.1.2.Connect函数**  
声明：int connect(const CommAddress &addr, DispatchHandlerPtr &default\_handler)  
功能：创建一个到addr 的TCP连接，并关联default\_handler作为此连接的默认事件分发器。有两类事件（CONNECTION\_ESTABLISHED和 DISCONNECT）可以通过分发器进行分发，ERROR类型的事件不能分发，因为连接上的任何错误都会导致连接关闭，最终导致一个DISCONNECT事件。函数的返回值可用于判断分发器是否被关联，以及是否可被回调。下列返回值代码表示分发器已经成功的关联：  
\*   - <code>Error::COMM\_BROKEN\_CONNECTION</code>  
\*   - <code>Error::OK</code>  
分发器遇到了DISCONNECT事件，或者调用了addr的close\_socket方法后将与连接分离。返回值Error::COMM\_BROKEN\_CONNECTION指示连接中断，并且随后将会迅速分发一个DISCONNECT事件。下列返回值表示分发器与连接处于分离状态，当然调用不会受到任何影响：  
\*   - <code>Error::COMM\_ALREADY\_CONNECTED</code>  
\*   - <code>Error::COMM\_INVALID\_PROXY</code>  
\*   - <code>Error::COMM\_SOCKET\_ERROR</code>  
\*   - <code>Error::COMM\_BIND\_ERROR</code>  
\*   - <code>Error::COMM\_CONNECT\_ERROR</code>  
\*   - <code>Error::COMM\_POLL\_ERROR</code>  
事件分发器从不会被调用线程回调，而是被reactor线程回调，因为reactor线程用于服务多种不同socket上的I/O事件。分发器应当迅速的从回调中返回。当回调时，reactor线程不会持有任何锁，所以分发器可以安全的访问Comm对象，例如：调用其send\_response方法。  
本函数实现了重载。第二种形式比当前多一个CommAddress类型的参数，表示本地地址。两种形式的函数中都实现了socket创建、bind本地地址、connect远程地址的操作。当连接远程地址成功后，创建IOHandlerData对象开始监测socket描述符上的读写事件。  
参数：addr 需要连接的地址；default\_handler 默认的事件分发器的智能指针  
返回值：Error::OK表示成功；其它错误代码表示失败。  
**1.1.3.Listen函数**  
声明：void listen(const CommAddress &addr, ConnectionHandlerFactoryPtr &chf)

功能：创建socket描述符，bind到addr，监听此socket描述符，最后创建IOHandlerAccept对象开始监测socket描述符上的读操作。

**2.class CommBuf**  
一个持有在网络上传输数据的Buffer，该类包含一个主Buffer和一个扩展Buffer，并且使用Buffer指针可以追踪其中已经写入的数据。当Buffer被发送后，将由IOHandler来处理其指针。下面示例说明了如何使用CommBuf构建一个请求消息。  
   \*   CommHeader header(COMMAND\_FETCH\_SCANBLOCK);  
   \*   header.gid = scanner\_id;  
   \*   CommBuf \*cbuf = new CommBuf(header, 4);  
   \*   cbuf->append\_i32(scanner\_id);  
下面是一段现场使用的代码，用于响应一个读请求。  
   \*   header.initialize\_from\_request\_header(m\_event->header);  
   \*   CommBufPtr cbp(new CommBuf( header, 10, ext));  
   \*   cbp->append\_i32(Error::OK);  
   \*   cbp->append\_i16(moreflag);  
   \*   cbp->append\_i32(id);  
   \*   error = m\_comm->send\_response(m\_event->addr, cbp);  
该类在构造时必须指定一个主Buffer，并且会包含一个header，扩展Buffer可选。主Buffer指针指向header之后。整个Buffer长度为：主Buffer长度+header长度+扩展Buffer长度，其值将被记录在header中。  
  
**3.class Event**  
网络通信事件，包含五种类型事件：CONNECTION\_ESTABLISHED-连接创建事件、DISCONNECT-连接断开事件、MESSAGE-请求响应时间、ERROR-错误时间、TIMER-定时器事件。可通过分发器将其分发到应用程序，分发器介绍见DispatchHandler。  
构造一个事件时，需指定事件类型type、事件源的地址addr、事件关联的错误代码error。  
事件对象包含一个group\_id属性，其表示事件属于的线程组，可以表示针对同一目标的事件。该值在构造事件时指定，其应为发生事件的socke描述符和事件头部的gid字段的组合，如果gid字段为0，则其值也为0。  
group\_id = ((uint64\_t)sd << 32) | header->gid

每个MESSAGE类型的事件包含一个头部header和一个事件主体payload，属性payload\_len表示事件主体的长度。Header是一个CommHeader对象，其包含了Message的版本、头部长度、整体长度、超时时间、组编号、请求编号、Message标志（正常请求、忽略请求、紧急请求等）。每个事件也都有一个到达时间和过期时间。

**4.class DispatchHandler **  
一个抽象基类，只负责分发应用程序的网络通信事件，事件处理由IOHandler类进行。  
**4.1.子类  
4.1.1.Class DispatchHandlerSynchronizer**  
DispatchHandler的子类，用于同步对应于sent request messages的response messages，它包含一个事件队列和一个条件变量。  
void handle(EventPtr &event\_ptr)  将event\_ptr放入事件队列并通知条件变量  
bool wait\_for\_reply(EventPtr &event\_ptr) 阻塞等待直到条件变量非空，得到事件队列中的第一个事件。  
**4.1.2.Class ConnectManager**

创建并且维护一组TCP连接。如果任一连接中断，将会持续的重试去重建连接，当然，两次重试之间会有一定的间隔。

**5.Class ConnectionHandlerFactory**

一个抽象基类，只负责为应用程序创建分发器（DispatchHandler）对象。

**6.class RequestCache**  
该类包含了所有未完成的请求，并且类的每个对象都将关联到一个reactor。当发送一个包含有响应处理器的请求时(见Comm\#send\_request)，该请求将被插入到RequestCache。当接收到一个请求响应时，通过查找RequestCache中对应的请求ID，将会获得响应处理器。  
Cache中以双向链表的结构保存了所有的请求，链表节点如下：  
    struct CacheNode \{  
      struct CacheNode  \*prev, \*next;  //!< Doubly-linked list pointers  
      boost::xtime       expire;  //!< Absolute expiration time  
      uint32\_t           id;      //!< Request ID  
      IOHandler         \*handler; //!< IOHandler associated with this request  
      /// Callback handler to which MESSAGE, TIMEOUT, ERROR, and DISCONNECT  
      /// events are delivered  
      DispatchHandler   \*dh;  
    \};  
**6.1.成员函数  
6.1.1.get\_next\_timeout函数**  
声明：DispatchHandler \*get\_next\_timeout(boost::xtime &now, IOHandler \*&handlerp, boost::xtime \*next\_timeout);  
功能：从列表头部开始查找，找到第一个已经过期的请求，删除该请求并返回其关联的事件处理器。  
参数： Now：当前时间；Handlerp：\[output\]超时请求关联的I/O处理器；next\_timeout：\[output\]过期请求的下一个请求的过期时间，如果列表已空，则为0.  
返回值：返回超时请求关联的事件分发处理器，如果没有则为0.  
  
**7.class IOHandler **  
Socket描述符对应的I/O处理基类。当Socket创建时，将分配一个I/O处理对象去处理其上发生的事件。处理结果将会封装在Event类中，通过DispatchHandler分发到应用程序。例如：一个TCP Socket将关联一个IOHandlerData对象，该对象从Socket读出数据，然后通过DispatchHandler对象发送到应用程序。  
该类构造时必须指定关联的socket描述符和DispatchHandler对象，并且将会关联一个Reactor对象。将会采用轮询的方式从ReactorFactory中获取一个Reactor对象（ReactorFactory\#get\_reactor），这样可以尽可能的均衡每个Reactor对象的epoll句柄监听的socket描述符的数量。  
**7.1.成员函数  
7.1.1.Start\_polling**  
声明：start\_polling(int mode)  
功能：  
本函数执行前，该对象已经对应了一个Reactor对象和一个socket描述符。  
在Reactor对象的epoll句柄中注册socket描述符的mode事件，并注明事件的回调者为自身，即Reactor对象的对应线程将监听该socket描述符上的mode事件，当事件发生时，将交给本对象处理。  
参数：mode 需要监测的事件。  
**7.1.2.deliver\_event**  
声明：void deliver\_event(Event \*event, DispatchHandler \*dh=0)  
功能：  
传送事件到应用程序的一个快捷方法。如果提供了事件分发器dh，该方法将通过其分发事件到应用程序，否则，将通过默认的分发器进行。如果没有默认的事件分发器，则将事件记录到日志。  
参数：Event 指向事件的指针，将被本方法删除；Dh 事件分发器。  
**7.1.3.handle\_event**  
声明：virtual bool handle\_event(struct pollfd \*event, time\_t arrival\_time=0) = 0  
功能：纯虚函数，处理事件。子类中将实现具体的处理过程。  
**7.1.4.其它函数**  
int add\_poll\_interest(int mode)：添加需要监测的事件。  
int remove\_poll\_interest(int mode)：删除不需要监测的事件。  
void display\_event(port\_event\_t \*event)：显示事件。  
void stop\_polling()：停止事件监控模式。  
**7.2.子类  
7.2.1.Class IOHandlerAccept**  
对TCP socket上监听到事件进行I/O处理。该类功能比较简单，只实现了父类的handle\_event函数，函数中接收（accept）来自客户端的请求，然后创建IOHandlerData对象开始监测该请求的读写事件，即该类的事件处理最终将转换为IOHandlerData对象的事件处理。  
**7.2.2.class IOHandlerData**  
用于TCP socket的I/O处理。每次需要被处理的数据将封装为一个CommBuf，然后放入该类的一个链表中。   
Flush\_send\_queueh函数从链表中取出一个CommBufPtr，然后将其中的数据写入对应的socket描述符，然后删除链表中的元素。  
Send\_message函数将需要处理的CommBufPtr插入链表中，然后调用Flush\_send\_queueh函数执行写操作。  
handle\_event函数根据监测socket描述符得到的事件进行相应的处理，例如：如果是POLLOUT事件，则执行写操作，如果是POLLIN事件，则执行读操作。  
**7.2.3.class IOHandlerDatagram**

用于UDP socket的I/O处理。

**8.class Reactor**  
该类用于管理socket通信线程的状态。  
构造函数中将针对不同的操作系统和socket编程模式进行不同的初始化。如果ReactorFactory::use\_poll为true（默认为false），则使用poll模式，否则对于linux使用epoll模式，对于OSX和FreeBSD使用kqueue模式，对于Solaris使用port\_associate模式。对于polling模式，虽然创建了一个socket \#m\_interrupt\_sd，但是不支持poll\_wait接口（此接口在ReactorRunner中实现）。  
对于linux系统中的epoll模式，构造时将创建名称为poll\_fd的epoll句柄(通过函数epoll\_create)，注意:一个epoll句柄可以管理多个socket描述符。构造函数中就创建了一个名称为m\_interrupt\_sd的UPD socket，并将其交由epoll句柄进行管理（通过epoll\_ctl函数）。  
构造函数中创建的m\_interrupt\_sd，会将其bind到本地地址，然后通过其连接自身（暂不确定其意图）。

对于linux系统中的epoll模式来说，每个Reactor对象将对应一个epoll句柄，但其可以管理多个socket描述符，即通过epoll\_ctl函数注册多个socket描述符到该句柄。但是对这些描述符的监听并不是它的事，而是由ReactorRunner完成。

**9.class ReactorRunner**  
Reactor对象的执行线程类，即针对每个Reactor对象会对应一个线程，用于监听并处理Reactor对象中epoll句柄上的注册的socket描述符的事件。  
该类是一个函数对象，线程函数中将针对不同的操作系统和socket编程模式进行不同方式的监听处理。  
对于linux系统中的epoll模式，将采用epoll\_wait函数循环的监听对应的Reactor对象中epoll句柄上（m\_reactor\#poll\_fd）的事件。对监听到的事件，将交由socket描述符对应的IOHandle对象进行处理（IOHandle\#handle\_event），处理完毕后清除该IOHandle对象。  
注意：每个IOHandle对象在创建时，会指定对应的Reactor对象及socket描述符，并会通过start\_polling函数创建对象自身与Reactor对象和socket描述符的对应关系。例如：Comm\#list中，在创建了名称为sd的监听socket后，后续有两条语句：  
handler = new IOHandlerAccept(sd, default\_handler, m\_handler\_map, chf)；  
handler->start\_polling()；  
第一条语句创建了IOHandlerAccept对象handler，并指定了其与sd和Reactor对象的对应关系。当然，这些都是在IOHandlerAccept的父类IOHandle的构造函数中完成的，并且Reactor对象是从ReactorFactory中轮询挑选出来的。第二条语句确定了三者的对应关系（通过epoll\_ctl函数将三者绑定在一起）。

该类包含一个名称为handler\_map（HandlerMap类实例）的静态成员，其存放了所有的socket描述符和IOHandle对象对应关系的集合。该集合将在Comm对象、IOHandle对象和ReactorRunner对象之间共享，也就是说在一个进程内，所有的socket描述符与IOHandle的对应关系将存放在一个集合中，但是多个不同类型的对象会对该集合进行操作。

**10.class ReactorFactory**  
静态类，即所有的成员函数和变量都是静态的。主要用于创建Reactor对象和ReactorRunner线程，并指定了两者的对应关系，即一个ReactorRunner线程只能用于监听一个Reactor对象的epoll句柄上的事件。对象和线程数目可以指定，RangeServer进程中默认为CPU核数。

IOHandle对象在构造时需要关联一个Reactor对象，此时需要调用该类的get\_reactor函数。该函数将会采用轮询的方式从所有的Reactor对象中获取一个，这样可以尽可能的均衡每个Reactor对象的epoll句柄监听的socket描述符的数量。

**11.Class ApplicationHandler**  
应用程序处理器的基类。此类的对象将被添加到一个ApplicationQueue，此类在构造初始化时将使用来自Comm层的MESSAGE事件。一个需被处理的请求有两个属性Group ID和Urgency，它们可以控制该请求在ApplicationQueue中被如何处理。  
Group ID：其支持ApplicationQueue在共享资源上的串行执行，即包含相同Group ID的请求处理将被串行执行。当使用一个MESSAGE事件初始化时，Group ID将等于此MESSAGE中CommHeader的gid字段，否则将为0.  
Urgency：ApplicationQueue支持二级的请求优先级。指定为urgent的请求将优先于非urgent的请求执行，即使ApplicationQueue暂停时，urgent请求也将执行。当使用一个MESSAGE事件初始化时，如果在MESSAGE头部的flags字段中CommHeader::FLAGS\_BIT\_URGENT被设定，则m\_urgent被设为true。  
  
**12.Class ApplicationQueue: public ApplicationQueueInterface**  
这是一个服务端应用程序助手类，所谓服务端应用程序，即是由网络上接收的消息所驱动的，用于处理客户端请求的服务器端程序。该类可以联合ApplicationHandler一起来实现一个请求队列。工作线程（Worker）将从队列中拿出一个请求进行处理。该类具有以下特征：  
组：由于一组工作线程从队列中拿到请求后独立执行，所以请求执行的次序可能会和请求到达的次序不符。对于一些特定的请求序列，这可能会产生错误，例如：在DfsBroker中追加数据到一个文件的请求序列，或者使用多个预加载请求从一个扫描器中提取数据。使用分组可使应用程序能够串行处理一组请求，每个请求将会有一个组号，组号可通过ApplicationHandler\#get\_group\_id方法获取。具有相同组号的请求将按照请求到达队列的次序串行处理，不属于任何组的请求组号为0，并且会独立执行。  
优先级（Prioritization）：ApplicationQueue支持二级的请求优先级。标记为紧急级别（urgent）的请求将优先于非紧急的请求执行，即使ApplicationQueue暂停时，紧急级别的请求也将执行。Hypertable将METADATA扫描和更新操作标记为紧急级别，当RangeServer中的应用程序队列由于内存不足而暂停时，它们也会优先执行并会避免死锁情况。需要使用ApplicationHandler\#is\_urgent方法将一个请求设置为紧急级别。  
Hyperspace的下列请求事件在构造时将置为紧急级别：COMMAND\_KEEPALIVE、COMMAND\_REDIRECT、COMMAND\_HANDSHAKE、COMMAND\_STATUS。  
RangeServer的下列请求事件在构造时将置为紧急级别：COMMAND\_LOAD\_RANGE、COMMAND\_UPDATE、COMMAND\_UPDATE\_SCHEMA、COMMAND\_COMMIT\_LOG\_SYNC、COMMAND\_CREATE\_SCANNER、COMMAND\_STATUS、COMMAND\_CLOSE、COMMAND\_WAIT\_FOR\_MAINTENANCE、COMMAND\_SHUTDOWN、COMMAND\_DUMP、COMMAND\_ACKNOWLEDGE\_LOAD、COMMAND\_HEAPCHECK、COMMAND\_SET\_STATE、COMMAND\_REPLAY\_FRAGMENTS、COMMAND\_PHANTOM\_RECEIVE、COMMAND\_PHANTOM\_UPDATE、COMMAND\_GET\_STATISTICS、create\_request\_phantom\_ranges。  
该类有以下几个嵌套类：  
**1）Class GroupState**  
跟踪组状态。该类的每个对象具有一个唯一的组编号，利用组编号可以跟踪组中请求的执行状态。  
该类只包含三个属性：group\_id、running和outstanding。Running表示是否正在处理该组的一个请求；outstanding表示该组在队列中尚未处理的请求数。  
**2）Class RequestRec**  
请求队列中的单个元素，表示一个请求对象。  
每个请求对象包含两个属性：handler和group\_state，前者表示请求对应的ApplicationHandler，后者表示请求对应的GroupState。  
**3）Class ApplicationQueueState**  
跟踪应用程序请求队列状态，并与工作线程共享该状态。  
该对象包含一个正常请求队列和一个紧急级别请求队列，两个队列共享一个GroupState集合，该集合是一个以group\_id为key，GroupState指针为value的map集合。该对象还包含了已用的线程数和总线程数。  
**4）Class Worker**  
应用程序请求队列工作线程，这是一个函数对象。该对象将从请求队列中取出一个请求对象，然后调用请求对象的handler属性的run方法处理该请求。处理完毕后，将会从请求队列和GroupState集合中删除对应的数据。

该类可以通过指定线程数来实例化。实例化时将会创建指定数量的线程，多个线程共享一个ApplicationQueueState对象。工作线程将会首先从ApplicationQueueState的紧急请求队列中获取一个请求（ApplicationHandler）对象指针，如果获取不到，再从正常请求队列中获取。然后调用请求对象的run函数去执行实际的操作。

**13.TestServer文件**  
先初始化ReactorFactory，包括reactor对象创建和ReactorRunner线程启动。  
创建了数个reactor对象。每个对象中都创建了一个名称为m\_interrupt\_sd的UPD socket描述符，绑定到本地地址，然后通过该socket连接自身，最后将该socket加入到监听队列。  
针对每个reactor对象启动一个ReactorRunner线程，线程中阻塞式监听reactor对象监听队列中所有的socket描述符。  
然后创建一个Comm对象，该对象检测如果没有reactor存在则退出程序，并与ReactorRunner线程共享handler\_map。ReactorRunner对象的handler\_map为静态成员，即所有ReactorRunner线程将共享一个handler\_map集合。  
如果是TCP连接模式，先创建一个Dispatcher对象，然后使用Comm对象的listen函数完成下列工作：创建一个名称为sd的TCP监听socket，并绑定到本地地址，然后开始监听sd上的请求；创建一个IOHandlerAccept对象，负责接收sd上的请求；让Dispatcher对象负责处理sd的响应；插入IOHandlerAccept对象到handler\_map集合；开启IOHandlerAccept对象的监控状态。  
如果是UDP连接模式，先创建一个UdpDispatcher对象，然后使用Comm对象的create\_datagram\_receive\_socket函数完成下列工作：创建一个名称为sd的UDPsocket，并绑定到本地地址；创建一个IOHandlerDatagram对象，负责接收对sd的请求；让UdpDispatcher对象负责处理sd的响应；插入IOHandlerDatagram对象到handler\_map集合；开启IOHandlerDatagram对象的监控状态。  
IOHandlerAccept对象和IOHandlerDatagram对象创建时会各自关联一个reactor对象。开启它们的监控状态时，会调用其继承自父类IOHandler的start\_polling函数，该函数会把sd和需要监听的事件（默认为读事件）加入到reactor对象的监听队列中。  
至此，服务器端的初始化完成，便开始阻塞式的等待来自客户端的请求了。  
在TCP连接模式中，ReactorRunner线程监听到socket描述符上有事件产生时，将调用对应的IOHandler对象的handler\_event函数进行处理。  
如果接收到客户端的connect请求，即监听socket上有事件发生，将调用该socket对应的IOHandlerAccept对象的handler\_event函数处理事件。该函数判断事件是否读事件，如果不是则将事件从线程的handler\_map集合中移除，如果是则完成下列工作：创建接收（accept）客户端的请求的sokect sd；创建一个IOHandlerData对象，负责处理sd上的I/O请求；让Dispatcher对象负责处理sd的响应；插入IOHandlerData对象到handler\_map集合；开启IOHandlerData对象的监控状态；传递一个CONNECTION\_ESTABLISHED-连接创建事件给客户端。事件传递最终由Dispatcher对象完成，该对象的类定义于testServer文件中，继承自DispatchHandler类，对CONNECTION\_ESTABLISHED事件的处理只是简单的写入了日志。  
如果接收到客户端发送的数据，即连接该客户端的socket上有读事件发生，将调用该socket对应的IOHandlerData对象的handler\_event函数处理事件。该函数调用et\_socket\_read函数读取数据，然后响应该请求。响应信息被封装在一个Event对象中，由Dispatcher对象的Handle函数处理。该函数将Message以外的其余事件都写入了日志，对Message事件将封装成一个CommBuf包后调用Comm对象的send\_response函数发回客户端。send\_response函数中获取与客户端连接对应的IOHandlerData对象，调用该对象的send\_message函数执行发送。  
**14.smapleClient文件**  
先初始化ReactorFactory，包括reactor对象创建和ReactorRunner线程启动。  
创建了一个reactor对象。对象中创建了一个名称为m\_interrupt\_sd的UPD socket描述符，绑定到本地地址，然后通过该socket连接自身，最后将该socket加入到监听队列。  
针对这个reactor对象启动一个ReactorRunner线程，线程中阻塞式的监听reactor对象监听队列中所有的socket描述符。  
然后创建一个Comm对象，该对象检测如果没有reactor存在则退出程序，并与ReactorRunner线程共享一个handler\_map集合。  
接下来初始化TCP或者UPD连接。  
如果是TCP连接模式，先创建一个ResponseHandlerTCP对象，然后使用Comm对象的connect函数完成下列工作：创建一个名称为sd的TCP socket，并绑定到本地地址；创建一个IOHandlerData对象，负责处理sd上的I/O请求；让ResponseHandlerTCP对象负责处理sd的响应；插入IOHandlerData对象到handler\_map集合；通过sd连接服务器，连接成功后开启IOHandlerData对象的监控状态。  
如果是UDP连接模式，先创建一个ResponseHandlerUDP对象，然后使用Comm对象的create\_datagram\_receive\_socket函数完成下列工作：创建一个名称为sd的UDPsocket，并绑定到本地地址；创建一个IOHandlerDatagram对象，负责接收对sd的请求；让UdpDispatcher对象负责处理sd的响应；插入IOHandlerDatagram对象到handler\_map集合；开启IOHandlerDatagram对象的监控状态。  
IOHandlerData对象和IOHandlerDatagram对象创建时会各自关联一个reactor对象。开启它们的监控状态时，会调用其继承自父类IOHandler的start\_polling函数，该函数会把sd和需要监听的事件（默认为读事件）加入到reactor对象的监听队列中。  
最后，将从文件中每次读出一行，将其封装为一个CommBuf包，发送至服务器，直至文件结束。  
如果是TCP连接模式，将使用Comm对象的send\_request函数完成下列发送工作：从handler\_map集合中获取与该服务器连接的socket对应的IOHandlerData对象；设置CommBuf包头中的标记位和包编号，表明该包为一个请求包；调用IOHandlerData对象的send\_message函数完成发送。当ResponseHandlerTCP对象指针为空时，表示该请求无需响应，则包编号为0，并置标记位FLAGS\_BIT\_IGNORE\_RESPONSE，否则包编号不能为0。

IOHandlerData对象的send\_message函数完成下列工作：将CommBuf包放入发送队列；调用flush\_send\_queue函数将队列中的包发送到服务器；将需要返回响应信息的包编号、当前IOHandlerData对象指针、ResponseHandlerTCP对象指针一并放入reactor对象中的请求缓存队列。

**15.Linux的epoll模式下本模块主体逻辑**  
首先通过ReactorFactory\#initialize(n)创建n个Reactor对象和ReactorRunner线程，对象和线程一一对应。  
每个Reactor对象创建时，会创建一个epoll句柄。ReactorRunner线程启动后，会创建一个唯一的、全局共享的集合handle\_map，用于存放socket描述符和IOHandle对象的对应关系，并开始持续的监听注册在epoll句柄上的socket描述符的事件。  
每当新建一个socket描述符（A）时，也会创建一个IOHandle子类的对象（B）。创建B时必须提供一个socket描述符和一个事件分发器对象，并且会轮询的在已有的Reactor对象中选取一个（C）。此时，C、A和B三者之间就建立了一个对应关系。随后必须确定这个对应关系，这需要做两件事：1）在C的epoll句柄上注册A的监听事件，并注明事件的回调者为B；2）在handle\_map中添加A和B的对应关系。  
当A上有事件产生时，与C对应的ReactorRunner线程将监听到该事件。由于事件的回调者为B，所以交由B进行事件处理（IOHandle\#handle\_event）。B将捕获的事件解析后再封装为一个Event对象，然后交由事件分发器对象进行后续处理。最后从handle\_map中移除A和B的对应关系。  
事件分发器一般由本模块的上层模块提供。例如：RangeServer模块在使用本模块时，核心关联代码为：m\_comm->listen(listen\_addr, chfp)。此代码在执行时，首先会创建IOHandlerAccept对象，但是为该对象关联的事件分发器为空。这说明在创建TCP连接时，只由IOHandlerAccept\#handle\_event处理即可，无需再由事件分发器处理了。在IOHandlerAccept\#handle\_event中，还创建了IOHandlerData对象，此时事件分发器为RangeServer的ConnectionHandler对象，果然，在IOHandlerData\#handle\_event的最后，将后续的处理交给了ConnectionHandler。当然，这种做法从业务上也是完全可以理解的，因为对于connect请求只要连接创建成功即可，无需进一步操作。但是对于send请求，不但要接收send的数据（IOHandlerData\#handle\_event），还要进一步的处理（DispatchHandler\#handle）。