Oracle 系统管理 - Linux 系统 - Backtrack 5 - 安全 - Juniper 技术 - Cisco 技术 - 思科模拟器 - Cisco 认证 - Cisco ios 下载

您现在的位置是:Docker > Centos > nginx源码分析 event模块

nginx源码分析 event模块

时间:2018-06-09 15:26  来源:未知  阅读次数: 复制分享 我要评论

一、简介
     nginx是一款非常受欢迎的软件,具备高性能、模块化可定制的良好特性。之前写了一篇nginx的http模块分析的文章,主要对http处理模块进行了分析讲解,同时也涉及了nginx模块化的内容。至于nginx高性能的原因,希望能够在在这篇文章中就自己对于这方面的理解给大家分享一下。
     nginx的event处理模型包含两个方面:高效的IO处理函数,事件的异步处理(可选的线程池)。
 
二、IO复用函数
     nginx中包含epoll、poll、select、devpoll、kqueue、iocp等不同的IO处理函数。下面就常见的epoll、poll、select来做简单的分析
 
select
poll
epoll
fd数量
用fd_set类型来存储fd,每个fd占一位,fd_set在linux系统中默认1024位,所以select最多支持1024个fd(可以通过修改FD_SETSIZE来增加)
采用pollfd结构体来存储fd,因此支持的最大fd数目跟系统内存有关(支持存储多少个)
采用epoll_event存储,因此也只与内存有关
通知方式
需要遍历fd_set集合中所有的fd,然后一个个检查确定是哪些fd上有event发生
与select类似
fd上有事件发生时,epoll函数会直接返回,且只会返回有事件发生的fd,不需要遍历
消息传递
将fd_set拷贝到内核空间,有event发生时再从内核空间拷贝到用户空间
与select类似
mmap申请一块共享内存来存放消息,省去内存拷贝开销
 
     另外:epoll有两种工作模式:LT(level triggered)和ET(edge triggered);LT又称水平触发,也就是说如果fd集合中还有没有处理完的event,对epoll调用会立刻返回这些未处理完的event。ET又称水平触发,调用方需要一次性把返回的所有fd都处理完成,因此下次调用epoll时这些event标识就会被清空(这些事件就会丢失)。
     由于每个IO复用函数对应的操作方式都不一样,nginx为了方便切换不同的IO复用函数,就利用nginx模块化的机制将每种函数及其相关操作都封装成一个NGX_EVENT_MODULE模块,然后用配置的方式来指明采用哪种方式(每个平台下都会有默认的IO处理函数)。下面就用代码来做些说明:
1
2
3
4
5
6
7
///////////  nginx/src/event/ngx_event.c    ////////
//这里各个模块并不像phase_handler或者filter那样用某种结构来存储,而是直接外部声明(方便直接调用)
extern ngx_module_t ngx_kqueue_module;
extern ngx_module_t ngx_eventport_module;
extern ngx_module_t ngx_devpoll_module;
extern ngx_module_t ngx_epoll_module;
extern ngx_module_t ngx_select_module;
   
  下面先介绍event模块有关的调用流程,然后再具体分析模块的各个函数的相关功能。
 
     调用fork创建worker进程之前,依据配置初始化一部分模块
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
///////////  nginx/src/core/ngx_cycle.c     ////////
//在nginx主进程启动worker之前调用,用于初始化相关的配置
ngx_cycle_t *  ngx_init_cycle(ngx_cycle_t *old_cycle)
{
     .......
     //对类型为NGX_CORE_MODULE的模块调用对应的create_conf函数,进行申请空间之类的操作
     for (i = 0; cycle->modules[i]; i++) {
        if (cycle->modules[i]->type != NGX_CORE_MODULE) {
            continue;
        }
        .....
        if (module->create_conf) {
            rv = module->create_conf(cycle);
    .......
    conf.module_type = NGX_CORE_MODULE;
    conf.cmd_type = NGX_MAIN_CONF;
    .......
    //根据上面给conf的赋值语句,ngx_conf_parse主要对配置文件NGX_MAIN_CONF这一层级的
    //配置进行解析,并只对NGX_CORE_MODULE类型的模块中的cmd进行匹配,然后执行匹配cmd对
    //应的函数
    if (ngx_conf_parse(&conf, &cycle->conf_file) != NGX_CONF_OK) {
        environ = senv;
        ngx_destroy_cycle_pools(&conf);
        return NULL;
    }
    ......
    //对类型为NGX_CORE_MODULE的模块调用对应的int_conf函数,对相关的conf结构进行初始化
    for (i = 0; cycle->modules[i]; i++) {
if (cycle->modules[i]->type != NGX_CORE_MODULE) {
            continue;
 }
 ......
       if (module->init_conf(cycle, cycle->conf_ctx[cycle->modules[i]->index])
    ......
     //调用每个模块注册的init_module函数初始化相关参数(因为这个操作在fork worker进程之前,
     //所以初始化的参数都会被继承)
     if (ngx_init_modules(cycle) != NGX_OK) {
        /* fatal */
        exit(1);
    }
     ......
}

 

     初始化worker进程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
///////////  nginx/src/os/unix/ngx_process_cycle.c    ////////
//worker进程启动初始化
static void  ngx_worker_process_init(ngx_cycle_t *cycle, ngx_int_t worker)
{
     ......
    //调用每个模块注册的init_process函数(这个操作是在每个worker进程中的,所以申请一些进程
    //自用的一些资源,如内存、变量等)
    for (i = 0; cycle->modules[i]; i++) {
        if (cycle->modules[i]->init_process) {
            if (cycle->modules[i]->init_process(cycle) == NGX_ERROR) {
                /* fatal */
                exit(2);
            }
        }
    }
    ......
}

 

上面介绍了NGX_CORE_MODULE模块的相关函数执行的相关流程,下面就对应event模块具体介绍下这个流程:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
/////////  nginx/src/event/ngx_event.c  /////////
 
//ngx_event_core_create_conf函数主要是申请空间的,比较简单这里就不讲了
 
//在执行ngx_conf_parse函数时,配置项events属于NGX_MAIN_CONF类型,且events模块类型为
//NGX_CORE_MODULE,所以events配置对应的ngx_events_block就会被调用
 
//ngx_event_block这个函数与上面的ngx_init_cycle以及ngx_http_block函数的结构类似,都是
//先调用特定模块的create_conf函数,然后解析对应level的配置项(调用配置注册的函数),接着
//调用init_conf函数实现完成模块对应配置的初始化
static char *  ngx_events_block(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
    ......
    //执行NGX_EVENT_MODULE类型的模块注册的create_conf函数
    for (i = 0; cf->cycle->modules[i]; i++) {
        if (cf->cycle->modules[i]->type != NGX_EVENT_MODULE) {
            continue;
        }
        .....
        if (m->create_conf) {
            (*ctx)[cf->cycle->modules[i]->ctx_index] =  m->create_conf(cf->cycle);
    ......
    cf->module_type = NGX_EVENT_MODULE;
    cf->cmd_type = NGX_EVENT_CONF;
    //解析配置文件中涉及NGX_EVENT_MODULE模块并执行相关配置注册的相关函数
    rv = ngx_conf_parse(cf, NULL);
    ......
    //执行NGX_EVENT_MODULE类型的模块注册的init_conf函数完成模块conf的初始化
    //在event_core_module模块的init_conf阶段,即ngx_event_core_init_conf函数
    //中完成了IO复用模型的选择
    for (i = 0; cf->cycle->modules[i]; i++) {
        if (cf->cycle->modules[i]->type != NGX_EVENT_MODULE) {
            continue;
        }
  ......
       rv = m->init_conf(cf->cycle, (*ctx)[cf->cycle->modules[i]->ctx_index]);
     ......
}
 
//该函数在init_cycle中的init_module函数中被调用。调用时worker还没有启动
static ngx_int_t  ngx_event_module_init(ngx_cycle_t *cycle)
{
    ......
    shm.size = size;
    shm.name.len = sizeof("nginx_shared_zone") - 1;
    shm.name.data = (u_char *) "nginx_shared_zone";
    shm.log = cycle->log;
    //内部调用mmap申请一块共享内存,这样fork出来的worker也能继承这块共享内存的地址
    //worker进程可以通过继承的共享内存地址来实现与master以及其他worker的通信
    if (ngx_shm_alloc(&shm) != NGX_OK) {
        return NGX_ERROR;
    }
 
    shared = shm.addr;
    ngx_accept_mutex_ptr = (ngx_atomic_t *) shared;
    ngx_accept_mutex.spin = (ngx_uint_t) -1;
    //将accept锁放到共享内存中,这样就实现了对多个进程的互斥、同步操作
    if (ngx_shmtx_create(&ngx_accept_mutex, (ngx_shmtx_sh_t *) shared,
                         cycle->lock_file.data)
        != NGX_OK)
    {
        return NGX_ERROR;
    }
     
    //将连接计数器以及其他的一些原子类型的变量也存放到共享内存,这样可以很方便统计所有整个nginx
    //的连接总数
    ngx_connection_counter = (ngx_atomic_t *) (shared + 1 * cl);
    .....
}
 
//该函数ngx_worker_process_init中的init_process函数中被调用,因为执行阶段位于
//worker进程初始化时,所以这里申请的资源都属于各个worker独有,不用担心因被继承而导致的资源浪费
static ngx_int_t  ngx_event_process_init(ngx_cycle_t *cycle)
{
     .....
    //创建保存event的队列用于保存accept、普通事件
    ngx_queue_init(&ngx_posted_accept_events);
    ngx_queue_init(&ngx_posted_events);
 
    //初始化事件定时器
    if (ngx_event_timer_init(cycle->log) == NGX_ERROR) {
        return NGX_ERROR;
    }
    .....
    //创建连接池
    cycle->connections =
        ngx_alloc(sizeof(ngx_connection_t) * cycle->connection_n, cycle->log);
    .....
}
  以上就是event这个模块的基本内容,相较http模块来讲,东西少了很多,但是都是采取的ngx模块,方便解读。
 
三、高效事件处理
       nginx的master进程初始化过程中,调用函数ngx_open_listening_sockets对配置文件中每个listen的端口创建了socket并进行了bind,然后将这些已经bind的sockfd放到一个cycle->listening数组中,这个数组在fork worker子进程时被子进程继承(子进程继承父进程的文件描述符只是给文件描述符的打开数+1,并未改变该文件描述符指向的内容)。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
////////////  nginx/src/event/ngx_event.c  /////////////
//在这个函数中完成了对从master进程继承的listen fd的初始化工作
static ngx_int_t  ngx_event_process_init(ngx_cycle_t *cycle)
{    
    ......
    //遍历所有的listen fd,然后从连接池中获取一个连接并初始化为listen fd的连接
    /* for each listening socket */
    ls = cycle->listening.elts;
    for (i = 0; i < cycle->listening.nelts; i++) {
     
#if (NGX_HAVE_REUSEPORT)
        //如果支持REUSEPORT选项,那么socket只能在一个worker上工作
        if (ls[i].reuseport && ls[i].worker != ngx_worker) {
            continue;
        }
#endif
         //从连接池中获取一个连接实例并进行部分初始化
        c = ngx_get_connection(ls[i].fd, cycle->log);
        if (c == NULL) {
            return NGX_ERROR;
        }
        //继续初始化
        c->type = ls[i].type;
        c->log = &ls[i].log;
        .....
}
     
  由于ngx_event_process_init在每个worker初始化过程中被调用,因此一个listen 端口对应的文件描述符同时存在于所有的worker进程中,如果不采取任何处理措施直接将这些socket fd直接添加到epoll等IO监听函数中,那么当有客户端向listen port发送请求时,所有的监听该fd的worker进程中epoll等函数都会返回,进而去处理该连接事件,这样就出现了多个worker去争抢一个资源的情况,这就是常见的“惊群效应”,这种方式对系统性能的影响太大。为了解决这个问题,nginx采用了accept_mutex(在ngx_event_module_init函数中,用共享内存的方式来存放accept_mutex)的方式来对listen fd加锁,使得每次有新连接到达的时候,只有一个worker进程去处理。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
////////////  nginx/src/os/unix/ngx_process_cycle.c  /////////////
//worker进程处理连接、请求的主函数
static void  ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data)
{
    .....
    //初始化worker进程
    ngx_worker_process_init(cycle, worker);
    ngx_setproctitle("worker process");
 
    for ( ;; ) {
        .....
        //处理新连接以及已有连接上的新数据
        ngx_process_events_and_timers(cycle);
        .....
    }
}
 
////////////  nginx/src/event/ngx_event.c  /////////////
//检查是否有新连接,并处理事件队列中的事件
void ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
    .....
    //检查accept锁是否启用
    if (ngx_use_accept_mutex) {
        //如果accept_disable大于0,说明当前进程负载偏高或者接受新连接异常
        //该值-1是为了尽快让该进程参与accept锁的竞争,防止一直空闲
        if (ngx_accept_disabled > 0) {
            ngx_accept_disabled--;
        } else {
            //尝试获取accept,如果获取成功ngx_accept_mutex_held会被置1,
            if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
                return;
            }
 
            if (ngx_accept_mutex_held) {
                //设置NGX_POST_EVENTS标志是为了在处理新连接时,将新的event放到
                //事件队列异步处理,而不是立刻处理(防止阻塞)
                flags |= NGX_POST_EVENTS;
 
            } else {
                //没获取到accept锁
                if (timer == NGX_TIMER_INFINITE
                    || timer > ngx_accept_mutex_delay)
                {
                    timer = ngx_accept_mutex_delay;
                }
            }
        }
    }
    delta = ngx_current_msec;
     //这里是一个函数指针,指向当前使用IO模块中的事件处理函数,等待事件到来的超时时间是timer
    (void) ngx_process_events(cycle, timer, flags);
 
    delta = ngx_current_msec - delta;
 
    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
                   "timer delta: %M", delta);
     
    //处理accept event队列中事件
    ngx_event_process_posted(cycle, &ngx_posted_accept_events);
     
    //如果持有accept锁,就释放,避免一个进程长期持有该锁。在大都是长连接的情况下,这回导致
    //各个worker负载的严重不均匀
    if (ngx_accept_mutex_held) {
        ngx_shmtx_unlock(&ngx_accept_mutex);
    }
     
    //检查是否有超时(过期)的事件,如果有进行处理
    if (delta) {
        ngx_event_expire_timers();
    }
     
    //处理普通event队列中的事件
    ngx_event_process_posted(cycle, &ngx_posted_events);
}
 
////////////  nginx/src/event/ngx_event_accept.c  /////////////
//接受新的连接
void  ngx_event_accept(ngx_event_t *ev)
{
     ......    
     //如果剩余的空闲连接数小于连接池总数的1/8,下次执行ngx_process_events_and_timers时
     //就不再竞争accept锁
     ngx_accept_disabled = ngx_cycle->connection_n / 8
                              - ngx_cycle->free_connection_n;
     ......
     //从连接池中取出一个连接并进行初始化
     c = ngx_get_connection(s, ev->log);
     ......
    //将新建的连接通过ngx_add_conn函数指针加入到IO复用函数的等待队列中
    if (ngx_add_conn && (ngx_event_flags & NGX_USE_EPOLL_EVENT) == 0) {
          if (ngx_add_conn(c) == NGX_ERROR) {
                ngx_close_accepted_connection(c);
                return;
          }
    }
 
    log->data = NULL;
    log->handler = NULL;
    //ls对应的handler是在ngx_http_add_listening函数中将ngx_http_init_connection
    //函数赋值给了handler
    ls->handler(c);
    ......
}
 
////////////  nginx/src/event/modules/ngx_epoll_module.c  /////////////
//epoll模块事件处理函数
static ngx_int_t  ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
     ......
     //等待新连接(最多阻塞timer时长)
     events = epoll_wait(ep, event_list, (int) nevents, timer);
     
     ......
     //如果设置了NGX_POST_EVENTS就说明当前持有accept锁,应当将新事件放到队列中,尽快返回
     if (flags & NGX_POST_EVENTS) {
         queue = rev->accept ? &ngx_posted_accept_events
                                    : &ngx_posted_events;
         ngx_post_event(rev, queue);
     } else {
                //没有持有accept锁,可以同步执行事件处理函数
                //rev->handler在ngx_http_init_connection函数中被赋值为ngx_http_wait_request_handler
                rev->handler(rev);
     }
     .......
}

 

     来总结一下:accept新创建的连接被添加到epoll中,然后调用ngx_http_init_connection来初始化http request,并将ngx_http_wait_request_handler赋值给rev->handler,这样无论是在epoll返回时调用rev->handler还是在处理ngx_posted_events队列时调用,都是相当于直接调用ngx_http_wait_request_handler来读取http的请求数据。至此,就将event模块和http模块连接起来,整个nginx的处理流程也就走通了。
相关资讯