在libevent中服务模型

1、muduo：one loop per thread，主线程注册listen事件，通过某种负载均衡机制（round robin）将连接的事件注册到子线程的Reactor上

muduo还提到了一个runInLoop()的功能：如果用户在当前线程调用，则回调functor会同步进行，如果在其他线程调用，则IO线程会被唤醒执行这个functor。这种跨线程调用是如何实现的？因为其他线程很可能阻塞在Reactor上。传统方法是用pipe ，在muduo里面是 event_fd，将回调放入线程的任务队列，并发送一个uint64大小的消息来唤醒Reactor。

2、nginx：master + worker的工作模式，master接受连接分配给worker，事实上不是这样的，master只是通过信号管理worker进程，worker之间通过accept_mutex来决定是否将监听套接字加入到loop中。
所谓的one loop per process

3、libevent：这是在网上找的资料，libevent并不是线程安全，但不代表其不支持多线程，有一个经典的图描述了其多线程实现。

其中socketpair 与 event_fd

这种消息通知+同步层的机制，通过pipe和一个加锁的任务队列（CQ）实现。

在高性能的网络程序中，使用得最多的是非阻塞IO和多IO复用的模型，通常称为Reactor模式。还有另一种模式称为Proactor。目前大部分的实现都是基于Reactor模式的实现。
在Reactor的模式中，基本结构是事件循环，以事件驱动和事件回调的方式实现业务逻辑。其中的事件可以包含读写socket、连接的建立都可以采用非阻塞的方式进行，这种方式能够提高并发度和吞吐量，对于IO密集型的应用比较合适。但是回调的方式也导致了业务逻辑的割裂。
Reactor模式通常是一种非阻塞IO和单线程循环相结合的方式，称为non-blocking
IO + one loop per thread模式。
其中的loop是指线程的主循环，通常将Timer或IO注册到线程的loop中。对于实时性要求较高的情况，可以采用一个单一的线程完成连接操作，而将数据处理的任务分摊给一个线程池处理。实现了一种1+N的方式。非阻塞的IO和超时实现了这个循环的执行。
在面向对象的编程中都支持BlockingQueued的方式实现从一个线程往另一个线程丢数据的方式。
在linux中Socket和Pipe都是基于fd的操作，因此都支持select、poll等。在性能方面做得较好的框架中的实现方法中采用了如下的实现方式:多个单线程的进程（nginx，从接口看可支持多线程，采用主进程和多个工作进程的方式）或运行多线程的进程。通常调用fork的进程不采用多线程。
原因：调用fork的目的：1、执行其他的业务进程。2、通过父子进程之间共享文件描述符的方式进行父子间进程通信，协同完成任务。

简要的阅读了Libevent的sample，基本的实现过程如下：
首先分配一个base，后面所有的操作都会与这个base进行关联：

base = event_base_new();

然后设置一系列的事件控制块，实际上是完成对应事件的处理函数注册过程。

signal_event = evsignal_new(base, SIGINT, signal_cb, (void *)base);
event_add(signal_event, NULL);

最后进入循环中，等待对应的事件的产生和执行：

event_base_dispatch(base);

后面再详细的分析libevent的基本框架和逻辑，在Libevent中实现不同线程之间的通告采用了pipe的方式；

具体的实现在函数evthread_make_base_notifiable()中。其中使用了socketpair，该函数会创建一对socket Pipe去实现不同线程间的通告。

evutil_socketpair(LOCAL_SOCKETPAIR_AF, SOCK_STREAM, 0,base->th_notify_fd);

首先定义了一个通告函数和回调函数，通告函数用于其他线程往libevent所在的线程通告。回调函数用于libevent所在的线程进行响应。
回调函数，在使用函数中接收到另一个线程的通告

 void (*cb)(evutil_socket_t, short, void *) = evthread_notify_drain_default;

/* 设置通告函数，在其他的线程中通告 */

 int (*notify)(struct event_base *) = evthread_notify_base_default;

在Event_Base中创建了一个notify的函数，实际的执行函数为evthread_notify_base_default().通过socket pipe[1]发送数据到pipe[0]中。

 base->th_notify_fn = notify;static int  evthread_notify_base_default(struct event_base *base){char buf[1];int r;buf[0] = (char) 0;r = write(base->th_notify_fd[1], buf, 1);return (r < 0 && errno != EAGAIN) ? -1 : 0;}

因此在Libevent所在的线程中需要监听pipe[0]的接收事件，因此需要监听Pipe[0]，并将回调函数作为该事件的响应，实际就是读取发送的内容。在Libevent中添加了一个基于pipe[0]读事件的处理，设置pipe[0]用于读数据，也就是获取通告，因此pipe[1]在其他线程中调用

 event_assign(&base->th_notify, base, base->th_notify_fd[0],EV_READ|EV_PERSIST, cb, base);

其中的cb就是定义的回调函数：

static void evthread_notify_drain_default(evutil_socket_t fd, short what, void *arg)
{unsigned char buf[1024];struct event_base *base = arg;while (read(fd, (char*)buf, sizeof(buf)) > 0);EVBASE_ACQUIRE_LOCK(base, th_base_lock);base->is_notify_pending = 0;EVBASE_RELEASE_LOCK(base, th_base_lock);
}

因此通过pipe[0,1]较好的实现了跨线程的事件通告，通过Event_base将不同线程间的通告函数进行了较好的包装。当然都采用了非阻塞的IO方式。

Signal的实现方式：
在IO复用模型中经常采用了都是LoopEvent的实现方式，在初始化IO复用模型的过程中也需要考虑信号的实现，在linux中，信号和程序的正常运行实际上是两个不同的控制流，为了在两个流之间进行通信，在LibEvent中采用了跨线程的通告方式，也是采用了socketPair的方式，具体的实现如下所示:
在Select的初始化函数中初始化信号evsig_init(base);

static void * select_init(struct event_base *base)
{struct selectop *sop;...evsig_init(base);...return (sop);
}

evsig_init的实现如下：

int  evsig_init(struct event_base *base)
{/* 这个signal实际上也是采用了socket的方式实现 */evutil_socketpair( AF_UNIX, SOCK_STREAM, 0, base->sig.ev_signal_pair);...evutil_make_socket_nonblocking(base->sig.ev_signal_pair[0]);evutil_make_socket_nonblocking(base->sig.ev_signal_pair[1]);/* 添加读事件到ev_signal_pair[1]，也就是在Signal的处理函数中可以采用ev_signal_pair[0]发送数据，触发ev_signal_pair[1]的读事件，函数evsig_cb中完成读操作 */event_assign(&base->sig.ev_signal, base, base->sig.ev_signal_pair[1],EV_READ | EV_PERSIST, evsig_cb, base);.../* 这边是实际信号发生时应该处理的操作结构体eventop */base->evsigsel = &evsigops;
}

evsig_cb：

static void  evsig_cb(evutil_socket_t fd, short what, void *arg)
{...while (1) {/* 从socket中读数据，同时为非阻塞，因此多次读数据会退出，不会死循环 */n = recv(fd, signals, sizeof(signals), 0);if (n == -1) {...break;} else if (n == 0) {break;}/* 统计对应信号触发的次数 */for (i = 0; i < n; ++i) {ev_uint8_t sig = signals[i];if (sig < NSIG)ncaught[sig]++;}}EVBASE_ACQUIRE_LOCK(base, th_base_lock);for (i = 0; i < NSIG; ++i) {if (ncaught[i])/* 激活对应的信号 */evmap_signal_active(base, i, ncaught[i]);}EVBASE_RELEASE_LOCK(base, th_base_lock);
}

结构体evsigops:

static const struct eventop evsigops = {"signal",NULL,evsig_add,  /* 添加操作函数 */evsig_del,NULL,NULL,0, 0, 0
};

添加的操作函数为evsig_add：

static int
evsig_add(struct event_base *base, evutil_socket_t evsignal, short old, short events, void *p)
{...EVSIGBASE_LOCK();evsig_base = base;
/* 信号添加 */evsig_base_n_signals_added = ++sig->ev_n_signals_added;
/* 发送信号事件的socket，也就是未使用的pipe[0] */evsig_base_fd = base->sig.ev_signal_pair[0];EVSIGBASE_UNLOCK();
...
/* 设置新号的处理函数为evsig_handler */
_evsig_set_handler(base, (int)evsignal, evsig_handler);
/* 添加事件 */
event_add(&sig->ev_signal, NULL);return (0);
}

信号处理函数evsig_handler():
该函数实际就是完成了信号的发送操作：

evsig_handler(int sig) {msg = sig;send(evsig_base_fd, (char*)&msg, 1, 0);
}

libevent的使用方式是最开始调用event_init初始化一个全局的event_base指针,以后使用其中的API添加新的事件均是对这个指针进行的操作.
试想如下一种典型的场景:
主线程使用libevent处理网络IO事件,接收新连接以及接收完客户端的数据之后将该事件交给辅助线程进行处理,辅助线程处理完了,需要往客户端发送回应数据,则再通过libevent提供的API将这个事件添加到可读事件中.但是,由于libevent中这个event_base指针是全局的,如果多线程同时添加可能会造成多线程不安全问题,这个在libevent的代码注释里面作者也做了注解.
简单的说一说这个event_msgqueue的原理及使用方式:
它的结构体为:

struct  event_msgqueue {int  push_fd;int  pop_fd;int  unlock_between_callbacks;struct   event  queue_ev;sp_thread_mutex_t  lock ;void  ( * callback)( void   * ,  void   * );void   * cbarg;struct  circqueue  * queue;
};

其中的queue是保存事件的队列,callback是处理事件的回调函数, lock是线程锁.
首先,它创建了一对socketpair(也就是结构体中的push_fd和pop_fd),将其中的一个通过libevent的event_add接口添加到关注的事件中,它的事件类型的是READ|PERSIST, 也就是说可读同时永不删除,而这个事件的回调函数是msgqueue_pop,这个函数的功能是将当前queue中的数据一一取出并且调用callback函数进行回调处理.
其次,外部的多线程需要往libevent添加事件时使用这个文件提供的msgqueue_push函数进行添加,此时, 往socketpair中发送一个字节的数据, 这么做的目的是为了让第一步已经添加的socketpair得到响应,此时, 由第一步,必然会触发之前注册的回调函数msgqueue_pop,将这个队列中的事件取出来进行处理。
这里使用socketpair起到了一个通知的作用,当队列中有数据时,通过向socketpair发送数据以调用msgqueue_pop,将数据取出来进行处理.可见,这个event_msgqueue起了一个中间层的作用,辅助线程要往libevent添加事件就通过这个队列,当队列中有数据时再触发libevent的事件处理机制进行处理.设计的非常巧妙.
结束语

消息通知+同步层的机制，通过pipe和一个加锁的任务队列（CQ）实现。

在libevent中服务模型相关推荐

libevent中的bufferevent
bufferevent是libevent中处理网络事件很重要也是比较复杂的一个模块,其中包含一个读事件,一个写事件,两个缓冲区(读和写).读写水位.三个回调(读.写.出错)和函数指针组成的操作集.目前 ...
libevent中的缓冲区(一）
libevent中的缓冲区定义为evbuffer,主要在文件evbuffer-internal.h文件中,定义如下 struct evbuffer {/** The first chain in ...
libevent中的基本数据结构
libevent中文件queue.h文件包含5种数据结构:单链表,双向链表,队列,尾队列,环形队列.在处理I/O和signal中的事件时,用的就是尾队列,下面就介绍这几种数据结构 1.单链表链表 ...
libevent中事件的添加与删除
前面介绍了libevent中的hash表,在添加事件时,具体是如何操作的呢?事件操作主要是在evmap.c文件中,包含了io事件,signal事件的操作.在事件操作时,分两种情况,一种是利用ha ...
libevent中的信号处理
libevent中将信号集成到event_base_loop事件循环中,通过socketpair转换成I/O事件,本文主要介绍相关的转换. 1.将信号转成I/O 采用socket pair方式,一个s ...
libevent中的时间及相关的管理
libevent中的时间及相关的管理在介绍时间之前,先说明几个与时间相关的函数及其用法 1.基础 1.1 clock_gettime(精度比较高,ns级) #include <time.h&g ...
libevent中的hash表
libevent中的hash表的代码在ht-internal文件中,在添加io事件,signal事件时,底层是在操作 event_io_map和event_signal_map 1. hash的 ...
在libevent中使用线程池
一线程的初始化 1线程对象在进行事件驱动时,每个线程需建立自己的事件根基.由于libevent未提供线程之间通信的方式,我们采用管道来进行线程的通信.同时为方便主线程分配线程,我们还需保留各个线程 ...
libevent mysql_在 libevent 中使用 MariaDB（MySQL）
在之前我翻译的官方文档中提到了 MariaDB 提供了对异步 I/O 的支持.那篇文章是一个比较简要的介绍.不过实际适配中,官方也提供了一个完整适配 libevent 的示例代码.本文算是对我上述示例 ...
在 libevent 中使用 MariaDB（MySQL）
在之前我翻译的官方文档中提到了 MariaDB 提供了对异步 I/O 的支持.那篇文章是一个比较简要的介绍.不过实际适配中,官方也提供了一个完整适配 libevent 的示例代码.本文算是对我上述示例 ...

在libevent中服务模型

在libevent中服务模型相关推荐

最新文章

热门文章