[置顶] nginx源码阅读(十).ngx_event_core_module模块

前言

模块的通用接口

感兴趣的配置项

存储配置项

事件模块的通用接口

小结

前言

本小节将进入到

ngx_event_core_module

模块,它是所有事件模块中排第一位的模块,因为要负责创建连接池,还要选择I/O多路复用机制等工作。接下来我们就来看看它是如何做到的,以及在nginx中扮演了一个什么角色。

模块的通用接口

我们先来看看

ngx_event_core_module

模块实现的所有模块都需要实现的

ngx_module_t

接口。

ngx_module_t  ngx_event_core_module = {
/* NGX_MODULE_V1定义为:
* #define NGX_MODULE_V1      0, 0, 0, 0, 0, 0, 1
* 直接初始化前7个成员
*/
NGX_MODULE_V1,
/* ctx成员,事件模块的具体化接口 */
&ngx_event_core_module_ctx,            /* module context */
//ngx_command_t成员
ngx_event_core_commands,               /* module directives */
//模块类型
NGX_EVENT_MODULE,                      /* module type */
NULL,                                  /* init master */
//初始化ngx_event_core_module模块
ngx_event_module_init,                 /* init module */
//在进入到工作循环前进行初始化
ngx_event_process_init,                /* init process */
NULL,                                  /* init thread */
NULL,                                  /* exit thread */
NULL,                                  /* exit process */
NULL,                                  /* exit master */
/* 将预留空间填充为0 */
NGX_MODULE_V1_PADDING
};

关于

ngx_event_module_init

方法,在进入到master进程工作循环前由

ngx_init_cycle

调用:

//调用所有模块的init_module方法
for (i = 0; ngx_modules[i]; i++) {
if (ngx_modules[i]->init_module) {
if (ngx_modules[i]->init_module(cycle) != NGX_OK) {
/* fatal */
exit(1);
}
}
}

ngx_event_module_init

所做的工作主要是初始化了一些变量,这里不具体分析(因为跟一些用于统计的变量有关,与该模块的联系不是特别大)

关于

ngx_event_process_init

,其实上一小节我们已经分析了它是如何创建连接池、读事件、写事件,不过除此之外,它还做了其他事情,比如调用指定的I/O多路复用机制模块的init方法。

关于它的调用,它是在

ngx_worker_process_init

中被调用。从

main

函数开始的调用栈如图:



而其的源码如下:

static ngx_int_t
ngx_event_process_init(ngx_cycle_t *cycle)
{
ngx_uint_t           m, i;
ngx_event_t         *rev, *wev;
ngx_listening_t     *ls;
ngx_connection_t    *c, *next, *old;
ngx_core_conf_t     *ccf;
ngx_event_conf_t    *ecf;
ngx_event_module_t  *module;

//获取ngx_core_module的配置项结构体指针
ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);
//获取ngx_event_core_module的配置项结构体指针
ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);

/* 当配置文件中的负载均衡锁的选项启用
* 并且使用的是多进程模型(worker进程数大于1并且使用了master进程)
* 于是开启accept_mutex负载均衡锁(代表需要进行负载均衡)
*/
if (ccf->master && ccf->worker_processes > 1 && ecf->accept_mutex) {
//将全局负载均衡锁开关置为1
ngx_use_accept_mutex = 1;
//ngx_accept_mutex_held为0代表表示当前进程未获取到负载均衡锁
//为1代表获取到了
ngx_accept_mutex_held = 0;
//ngx_accept_mutex_delay的值为在配置文件中指定的最大延迟时间
//具体所代表的含义,我们在后面分析进程间负载均衡的时候再去了解
ngx_accept_mutex_delay = ecf->accept_mutex_delay;

} else {
//否则将该全局变量置为0
//表示不启用负载均衡(毕竟只有一个进程)
ngx_use_accept_mutex = 0;
}

#if (NGX_THREADS)
ngx_posted_events_mutex = ngx_mutex_init(cycle->log, 0);
if (ngx_posted_events_mutex == NULL) {
return NGX_ERROR;
}
#endif

//初始化定时器(nginx中的定时器是由红黑树实现的)
if (ngx_event_timer_init(cycle->log) == NGX_ERROR) {
return NGX_ERROR;
}

//遍历所有事件模块
for (m = 0; ngx_modules[m]; m++) {
if (ngx_modules[m]->type != NGX_EVENT_MODULE) {
continue;
}

//配置项中的use选项选择了使用的I/O多路复用机制
//找到事件模块中的该事件驱动模块
if (ngx_modules[m]->ctx_index != ecf->use) {
continue;
}

module = ngx_modules[m]->ctx;

//调用该模块对应的init方法进行初始化
if (module->actions.init(cycle, ngx_timer_resolution) != NGX_OK) {
/* fatal */
exit(2);
}

break;
}

#if !(NGX_WIN32)

/* nginx中采用了时间缓存,并不是需要获取时间时就调用gettimeofday来获取
* 因此精度方面可能需要进行控制
* 而ngx_timer_resolution(通过核心模块的配置项获取,默认是0)表示时间的精度
* 而NGX_USE_TIMER_EVENT在epoll中并没有使用
* 因此只要设置了控制时间的精度
* 就会定期进行更新
*/
if (ngx_timer_resolution && !(ngx_event_flags & NGX_USE_TIMER_EVENT)) {
/* 设置SIGALRM捕捉之后的处理函数
* ngx_timer_signal_handler
*/
struct sigaction  sa;
struct itimerval  itv;

ngx_memzero(&sa, sizeof(struct sigaction));
sa.sa_handler = ngx_timer_signal_handler;
sigemptyset(&sa.sa_mask);

if (sigaction(SIGALRM, &sa, NULL) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"sigaction(SIGALRM) failed");
return NGX_ERROR;
}

/* 设置定时时间
* it_interval成员指定间隔的时间
* it_value成员指定初始定时时间
*/
itv.it_interval.tv_sec = ngx_timer_resolution / 1000;
itv.it_interval.tv_usec = (ngx_timer_resolution % 1000) * 1000;
itv.it_value.tv_sec = ngx_timer_resolution / 1000;
itv.it_value.tv_usec = (ngx_timer_resolution % 1000 ) * 1000;

/* 调用setitimer函数
* 每过ngx_timer_resolution时间调用ngx_timer_signal_handler方法
*/
if (setitimer(ITIMER_REAL, &itv, NULL) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"setitimer() failed");
}
}

/* 使用poll事件驱动 */
if (ngx_event_flags & NGX_USE_FD_EVENT) {
struct rlimit  rlmt;

if (getrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"getrlimit(RLIMIT_NOFILE) failed");
return NGX_ERROR;
}

cycle->files_n = (ngx_uint_t) rlmt.rlim_cur;

/* 预分配运行时使用的句柄 */
cycle->files = ngx_calloc(sizeof(ngx_connection_t *) * cycle->files_n,
cycle->log);
if (cycle->files == NULL) {
return NGX_ERROR;
}
}

#endif

/* 预分配connections数组 */
cycle->connections =
ngx_alloc(sizeof(ngx_connection_t) * cycle->connection_n, cycle->log);
if (cycle->connections == NULL) {
return NGX_ERROR;
}

c = cycle->connections;

/* 预分配读事件 */
cycle->read_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
cycle->log);
if (cycle->read_events == NULL) {
return NGX_ERROR;
}

/* 初始化读事件 */
rev = cycle->read_events;
for (i = 0; i < cycle->connection_n; i++) {
rev[i].closed = 1;
rev[i].instance = 1;
#if (NGX_THREADS)
rev[i].lock = &c[i].lock;
rev[i].own_lock = &c[i].lock;
#endif
}

/* 预分配写事件 */
cycle->write_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
cycle->log);
if (cycle->write_events == NULL) {
return NGX_ERROR;
}

/* 初始化写事件 */
wev = cycle->write_events;
for (i = 0; i < cycle->connection_n; i++) {
wev[i].closed = 1;
#if (NGX_THREADS)
wev[i].lock = &c[i].lock;
wev[i].own_lock = &c[i].lock;
#endif
}

/* 初始化状态时,connections数组中的都是空闲连接
* 使用data变量将connections数组中所有元素都连接起来
* 然后将读事件和写事件绑定到连接上
* 并将fd置为-1
*/
i = cycle->connection_n;
next = NULL;

do {
i--;

c[i].data = next;
c[i].read = &cycle->read_events[i];
c[i].write = &cycle->write_events[i];
c[i].fd = (ngx_socket_t) -1;

next = &c[i];

#if (NGX_THREADS)
c[i].lock = 0;
#endif
} while (i);

//将free_connections指向第一个空闲连接
cycle->free_connections = next;
//设置空闲连接数
cycle->free_connection_n = cycle->connection_n;

/* for each listening socket */

/* 为每个监听端口分配一个连接 */
ls = cycle->listening.elts;
for (i = 0; i < cycle->listening.nelts; i++) {

/* 获取一个空闲连接 */
c = ngx_get_connection(ls[i].fd, cycle->log);

if (c == NULL) {
return NGX_ERROR;
}

c->log = &ls[i].log;

//将listening对象绑定到连接对象上
c->listening = &ls[i];
//将连接对象绑定到listening对象上
ls[i].connection = c;

//监听端口只关心读事件
rev = c->read;

rev->log = c->log;
//设置为可以接受连接
rev->accept = 1;

#if (NGX_HAVE_DEFERRED_ACCEPT)
rev->deferred_accept = ls[i].deferred_accept;
#endif

if (!(ngx_event_flags & NGX_USE_IOCP_EVENT)) {
if (ls[i].previous) {

/*
* delete the old accept events that were bound to
* the old cycle read events array
*/

old = ls[i].previous->connection;

if (ngx_del_event(old->read, NGX_READ_EVENT, NGX_CLOSE_EVENT)
== NGX_ERROR)
{
return NGX_ERROR;
}

old->fd = (ngx_socket_t) -1;
}
}

#if (NGX_WIN32)

if (ngx_event_flags & NGX_USE_IOCP_EVENT) {
ngx_iocp_conf_t  *iocpcf;

rev->handler = ngx_event_acceptex;

if (ngx_use_accept_mutex) {
continue;
}

if (ngx_add_event(rev, 0, NGX_IOCP_ACCEPT) == NGX_ERROR) {
return NGX_ERROR;
}

ls[i].log.handler = ngx_acceptex_log_error;

iocpcf = ngx_event_get_conf(cycle->conf_ctx, ngx_iocp_module);
if (ngx_event_post_acceptex(&ls[i], iocpcf->post_acceptex)
== NGX_ERROR)
{
return NGX_ERROR;
}

} else {
rev->handler = ngx_event_accept;

if (ngx_use_accept_mutex) {
continue;
}

if (ngx_add_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
}

#else
/* 设置接受连接的回调函数为ngx_event_accept
* 即监听端口上收到连接请求时的回调函数
* 在ngx_event_accept函数中从连接池里获取到连接
* 关键操作:ls->handler(c); //调用其他事件消费模块的handler(比如http模块)
*/
rev->handler = ngx_event_accept;

/* 如果使用了负载均衡(ngx_use_accept_mutex为1)
* 只有当拥有负载均衡锁的时候才添加事件到epoll中
*/
if (ngx_use_accept_mutex) {
continue;
}

if (ngx_event_flags & NGX_USE_RTSIG_EVENT) {
if (ngx_add_conn(c) == NGX_ERROR) {
return NGX_ERROR;
}

} else {
//将读事件添加到epoll监听事件中
if (ngx_add_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
}

#endif

}

return NGX_OK;
}

感兴趣的配置项

在分析事件模块的具体化通用接口

ngx_event_core_module_ctx

之前,我们先对ngx_event_core_module模块对哪些模块感兴趣做个了解,这有利于我们看懂后面的代码。

static ngx_command_t  ngx_event_core_commands[] = {

//连接池的大小,即每个worker进程中支持的tcp最大连接数
//与下面的connections配置项意义相同
{ ngx_string("worker_connections"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_connections,
0,
0,
NULL },

//连接池的大小
{ ngx_string("connections"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_connections,
0,
0,
NULL },

//确定选择的哪一个事件模块作为I/O多路复用机制
{ ngx_string("use"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_use,
0,
0,
NULL },

//对于epoll来说,该配置项意味着在接收到一个新连接事件时,调用accpet尽可能多的接受连接
{ ngx_string("multi_accept"),
NGX_EVENT_CONF|NGX_CONF_FLAG,
ngx_conf_set_flag_slot,
0,
offsetof(ngx_event_conf_t, multi_accept),
NULL },

//确定是否使用负载均衡锁,默认为开启
//但是在nginx1.11.3版本之后,默认关闭
{ ngx_string("accept_mutex"),
NGX_EVENT_CONF|NGX_CONF_FLAG,
ngx_conf_set_flag_slot,
0,
offsetof(ngx_event_conf_t, accept_mutex),
NULL },

//启用负载均衡锁后,延迟accept_mutex_delay毫秒后再试图处理新连接事件
{ ngx_string("accept_mutex_delay"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_msec_slot,
0,
offsetof(ngx_event_conf_t, accept_mutex_delay),
NULL },

//需要对来自指定ip的tcp连接打印debug级的调试日志
{ ngx_string("debug_connection"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_debug_connection,
0,
0,
NULL },

ngx_null_command
};

之前也提到过,在

ngx_init_cycle

函数中会调用

ngx_conf_parse

对

nginx.conf

文件进行解析,其中会调用

ngx_conf_handler

,用依次读取到的配置项遍历

ngx_modules

数组,如果有模块对该配置项感兴趣,则会调用其

commands

成员中的

set

指向的函数对该配置项进行处理。

这里我们稍微总结一下事件模块中配置项的解析还有存储方面。

首先是

ngx_init_cycle

函数中调用

ngx_conf_parse

函数,其中会调用

ngx_conf_handler

函数,对读取到的配置项遍历

ngx_modules

数组,当读取到”events{}”时,

ngx_events_module

就会调用

commands

成员中

set

成员指向的函数

ngx_events_block

,该函数内部申请了存储配置项的结构体空间,并与核心结构体

ngx_cycle_t

的

conf_ctx

产生了联系

存储配置项的结构体已经有了,

ngx_events_block

调用

ngx_conf_parse

,当有事件模块感兴趣的配置项时,就会调用当前事件模块设置的配置项解析方法(对应上面列出的感兴趣的那7个配置项)

最后调用每个模块都会实现的

init_conf

对对应的配置项结构体里的没设置的值设为默认值。

ngx_event_core_module模块中对应的

init_conf

方法为

ngx_event_init_conf

函数,下面我们会对该函数进行分析。解析了配置文件之后,配置项结构体中还没有被赋值的成员,将被设置为默认值,这就是

ngx_event_init_conf

的工作。

存储配置项

那么现在我们就可以来看看,事件模块中存储配置项的结构体长什么样子(event/ngx_event.h中定义):

typedef struct {
//连接池的大小
ngx_uint_t    connections;
/* 选用的事件驱动模块
* 这里存储的是事件模块中模块内部的编号
*/
ngx_uint_t    use;

/* 标识位,为1时表示在接收到一个新连接事件时,一次性建立尽可能多的连接 */
ngx_flag_t    multi_accept;
/* 标识位,为1时表示启用负载均衡锁 */
ngx_flag_t    accept_mutex;

/* 负载均衡锁会使得一些worker进程在拿不到锁时延迟建立新连接
* 而延迟的时间就由accept_mutex_delay执行(以ms为单位)
*/
ngx_msec_t    accept_mutex_delay;

//选用的事件驱动模块的名字,它与use成员相对应
u_char       *name;

#if (NGX_DEBUG)
ngx_array_t   debug_connection;
#endif
} ngx_event_conf_t;

事件模块的通用接口

......
static ngx_str_t  event_core_name = ngx_string("event_core");

......
ngx_event_module_t  ngx_event_core_module_ctx = {
//name
&event_core_name,
//create_conf
ngx_event_create_conf,                 /* create configuration */
//init_conf
ngx_event_init_conf,                   /* init configuration */

/* 关于ngx_event_actions中的方法ngx_event_core_module并不需要去实现
* 因为该模块并不需要对事件进行注册、添加、删除等操作
*/
{ NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL }
};

下面我们来看看

ngx_event_create_conf

是如何实现的:

static void *
ngx_event_create_conf(ngx_cycle_t *cycle)
{
ngx_event_conf_t  *ecf;

//申请存储配置项的结构体(ngx_event_conf_t)空间
ecf = ngx_palloc(cycle->pool, sizeof(ngx_event_conf_t));
if (ecf == NULL) {
return NULL;
}

//将配置项结构体(ngx_event_conf_t)的成员设置为unset
ecf->connections = NGX_CONF_UNSET_UINT;
ecf->use = NGX_CONF_UNSET_UINT;
ecf->multi_accept = NGX_CONF_UNSET;
ecf->accept_mutex = NGX_CONF_UNSET;
ecf->accept_mutex_delay = NGX_CONF_UNSET_MSEC;
ecf->name = (void *) NGX_CONF_UNSET;

#if (NGX_DEBUG)

if (ngx_array_init(&ecf->debug_connection, cycle->pool, 4,
sizeof(ngx_event_debug_t)) == NGX_ERROR)
{
return NULL;
}

#endif

return ecf;
}

只看了函数的源码是远远不够的,主要是要把握该函数出于什么功能被调用,何时被调用,这样才能把事件模块的逻辑理清楚,否则看懂了代码也不明白为什么这样做。

ngx_event_create_conf

函数是在

ngx_events_block

中被调用的,它会先调用所有事件模块的

create_conf

函数(即

ngx_event_create_conf

函数),作用是在解析配置文件前,先申请存储配置项结构体(

ngx_event_conf_t

)的空间,接着将结构体的成员设置为

NGX_CONF_UNSET

(这不是代表默认值,而是未设置)。

接着

ngx_events_block

会调用

ngx_conf_parse

解析配置项并存储在结构体(

ngx_event_conf_t

)中

当解析完配置文件后,调用

ngx_event_init_conf

方法将配置项结构体中没有被设置值的成员设置为默认值。

static char *
ngx_event_init_conf(ngx_cycle_t *cycle, void *conf)
{
//获取到模块的配置项指针
ngx_event_conf_t  *ecf = conf;

#if (NGX_HAVE_EPOLL) && !(NGX_TEST_BUILD_EPOLL)
int                  fd;
#endif
#if (NGX_HAVE_RTSIG)
ngx_uint_t           rtsig;
ngx_core_conf_t     *ccf;
#endif
ngx_int_t            i;
ngx_module_t        *module;
ngx_event_module_t  *event_module;

module = NULL;

#if (NGX_HAVE_EPOLL) && !(NGX_TEST_BUILD_EPOLL)
/* 调用epoll_create测试epoll机制是否可用 */
fd = epoll_create(100);

if (fd != -1) {
/* 创建成功,证明epoll可用
* 关闭测试epoll返回的fd
* 选择epoll为默认的事件驱动模块
*/
close(fd);
module = &ngx_epoll_module;

} else if (ngx_errno != NGX_ENOSYS) {
/* 虽然返回-1创建失败了
* 但是如果errno不为ENOSYS(Function not implemented)
* 证明epoll还是可用的(至少当前机器有这个机制存在)
*/
module = &ngx_epoll_module;
}

#endif

/* 若有rtsig机制并且module还未选择
* 则默认使用rtsig机制
*/
#if (NGX_HAVE_RTSIG)

if (module == NULL) {
module = &ngx_rtsig_module;
rtsig = 1;

} else {
rtsig = 0;
}

#endif

/* 如果有devpoll机制,选择devpoll机制
* 这里你可能会疑惑为什么不用判断module是否为NULL
* 因为目前一个操作系统上不会同时出现epoll、devpoll、kqueue等机制
*/
#if (NGX_HAVE_DEVPOLL)

module = &ngx_devpoll_module;

#endif

/* 若有kqueue机制,则默认选择kqueue机制 */
#if (NGX_HAVE_KQUEUE)

module = &ngx_kqueue_module;

#endif

/* 如果module还未选择
* 有select就默认选select了
*/
#if (NGX_HAVE_SELECT)

if (module == NULL) {
module = &ngx_select_module;
}

#endif

/* 走到这一步,要是module还是没有选择出来
* 那就遍历事件模块,选择第一个非ngx_event_core_module模块作为默认的事件驱动模块
*/
if (module == NULL) {
for (i = 0; ngx_modules[i]; i++) {
//过滤掉非事件模块
if (ngx_modules[i]->type != NGX_EVENT_MODULE) {
continue;
}

event_module = ngx_modules[i]->ctx;

//过滤掉ngx_event_core_module模块
if (ngx_strcmp(event_module->name->data, event_core_name.data) == 0)
{
continue;
}

//选择ngx_modules[i]作为默认的事件驱动模块
module = ngx_modules[i];
break;
}
}

//此时还没有合适的默认module
//那就出错了
if (module == NULL) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, 0, "no events module found");
return NGX_CONF_ERROR;
}

/* 若配置项中没有设置连接池的大小
* 则设置连接池的默认大小
* DEFAULT_CONNECTIONS宏定义为512
* ngx_conf_init_uint_value是一个宏函数:
* #define ngx_conf_init_uint_value(conf, default)    \
* if(conf == NGX_CONF_UNSET_UINT) {  \
*     conf = default;    \
* }
*/
ngx_conf_init_uint_value(ecf->connections, DEFAULT_CONNECTIONS);

//给cycle->connection_n赋值
cycle->connection_n = ecf->connections;

//若use配置项没有赋值,则设置默认使用的事件驱动模块
ngx_conf_init_uint_value(ecf->use, module->ctx_index);

//初始化使用的事件模块的名字
event_module = module->ctx;
ngx_conf_init_ptr_value(ecf->name, event_module->name->data);

//若配置项中没有设置multi_accept,则默认不接受多个请求,即一次只accept一个连接
ngx_conf_init_value(ecf->multi_accept, 0);
//若配置项中没有设置accept_mutex,则默认使用负载均衡锁,但是1.11.3版本之后默认关闭
ngx_conf_init_value(ecf->accept_mutex, 1);
//若配置项中没有设置accept_mutex,则默认负载均衡锁的等待时间是500ms
ngx_conf_init_msec_value(ecf->accept_mutex_delay, 500);

#if (NGX_HAVE_RTSIG)

if (!rtsig) {
return NGX_CONF_OK;
}

if (ecf->accept_mutex) {
return NGX_CONF_OK;
}

ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);

if (ccf->worker_processes == 0) {
return NGX_CONF_OK;
}

ngx_log_error(NGX_LOG_EMERG, cycle->log, 0,
"the \"rtsig\" method requires \"accept_mutex\" to be on");

return NGX_CONF_ERROR;

#else

return NGX_CONF_OK;

#endif
}

最后附上一张大致流程图来总结本小节:

小结

本小节信息量略大,与第八节关系紧密,如果感觉难以理解,建议和第八节对照着一起看,希望能做到把主框架及

ngx_events_module

、

ngx_event_core_module

联系起来。

接下来我们将分析

ngx_epoll_module

模块。