微信公众号:郑尔多斯
关注可了解更多的Nginx
知识。任何问题或建议,请公众号留言;
关注公众号,有趣有内涵的文章第一时间送达!
内容回顾
前面的几篇文章中,我们介绍了nginx
的事件模块的基础知识。我们知道nginx
中包含了三个与事件相关的module
,分别为ngx_event_module
,ngx_event_core_module
,ngx_epoll_module
。我们也分别分析了这三个模块的配置文件的解析过程,从本篇文章开始,将详细分析nginx
的事件处理过程,比如如何添加事件,删除事件,处理事件等。
事件处理的概述
事件处理模块的主要解决的问题是如何收集,分发以及管理事件。在nginx
中,事件主要包含网络事件和定时器事件。当然还有其他事件,比如master
进程和worker
进程通信也是使用事件机制来完成的,但是我们的关注点主要在前两种事件,第三种事件以后会分析。前面我们分析了nginx
是如何解析配置文件中与事件有关的配置项的,我们知道,当遇到events{}
的时候,表示开始处理事件配置项了。
事件和连接
作为web服务器,每一个用户请求至少对应一个TCP
连接,每个连接都包含一个毒事件和一个写事件。这样epoll就可以根据触发的事件类型来调度相应的模块进行处理。Nginx
连接分为主动连接和被动连接。分别用ngx_connection_t
和ngx_peer_connection_t
结构体来表示。
主动连接: 客户端发起,服务器被动接受的连接
被动连接:Nginx
作为客户端,向上游服务器发起的连接
本系列文章所介绍的连接都是被动连接,即Nginx
作为服务器被动的接受客户端发起的连接。
Nginx连接池
为了提高效率,Nginx
在启动的时候,会创建好连接池和对应的读写事件,并且将每个连接都和对应的读写事件对应起来。这样的话,在后面使用到连接的时候,只需要从连接池中获取一个空闲的连接就行了。那么Nginx
是在什么时候创建了连接池和读写事件呢?
前面两篇文章我们介绍了ngx_events_module
, ngx_event_core_module
, ngx_epoll_module
的配置项解析过程,以及create_conf()
, init_conf()
方法。我们知道ngx_module_t
结构体有中有许多钩子函数,其中有一个钩子函数为init_process()
,这个方法会在Nginx
启动过程中调用。在master/worker
模式下,当创建了worker
进程之后,会在worker
进程的初始化过程中调用init_process()
。
通过源码可以发现,与事件机制相关的三个模块中,只有ngx_event_core_module
有一个init_process()
钩子函数,为ngx_event_process_init()
。其他两个模块都没有对应的钩子函数。
具体的调用过程如下:
ngx_event_process_init
下面我们就分析一下这个ngx_event_process_init()
这个方法。
1static ngx_int_t
2ngx_event_process_init(ngx_cycle_t *cycle)
3{
4 ngx_uint_t m, i;
5 ngx_event_t *rev, *wev;
6 ngx_listening_t *ls;
7 ngx_connection_t *c, *next, *old;
8 ngx_core_conf_t *ccf;
9 ngx_event_conf_t *ecf;
10 ngx_event_module_t *module;
11
12 ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);
13 ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);
14
15 if (ccf->master && ccf->worker_processes > 1 && ecf->accept_mutex) {
16 ngx_use_accept_mutex = 1;
17 ngx_accept_mutex_held = 0;
18 ngx_accept_mutex_delay = ecf->accept_mutex_delay;
19
20 } else {
21 ngx_use_accept_mutex = 0;
22 }
23
24 ngx_queue_init(&ngx_posted_accept_events);
25 ngx_queue_init(&ngx_posted_events);
26
27 if (ngx_event_timer_init(cycle->log) == NGX_ERROR) {
28 return NGX_ERROR;
29 }
30
31 for (m = 0; cycle->modules[m]; m++) {
32 if (cycle->modules[m]->type != NGX_EVENT_MODULE) {
33 continue;
34 }
35
36 if (cycle->modules[m]->ctx_index != ecf->use) {
37 continue;
38 }
39
40 module = cycle->modules[m]->ctx;
41
42 if (module->actions.init(cycle, ngx_timer_resolution) != NGX_OK) {
43 /* fatal */
44 exit(2);
45 }
46
47 break;
48 }
49
50#if !(NGX_WIN32)
51
52 if (ngx_timer_resolution && !(ngx_event_flags & NGX_USE_TIMER_EVENT)) {
53 struct sigaction sa;
54 struct itimerval itv;
55
56 ngx_memzero(&sa, sizeof(struct sigaction));
57 sa.sa_handler = ngx_timer_signal_handler;
58 sigemptyset(&sa.sa_mask);
59
60 if (sigaction(SIGALRM, &sa, NULL) == -1) {
61 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
62 "sigaction(SIGALRM) failed");
63 return NGX_ERROR;
64 }
65
66 itv.it_interval.tv_sec = ngx_timer_resolution / 1000;
67 itv.it_interval.tv_usec = (ngx_timer_resolution % 1000) * 1000;
68 itv.it_value.tv_sec = ngx_timer_resolution / 1000;
69 itv.it_value.tv_usec = (ngx_timer_resolution % 1000 ) * 1000;
70
71 if (setitimer(ITIMER_REAL, &itv, NULL) == -1) {
72 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
73 "setitimer() failed");
74 }
75 }
76
77 if (ngx_event_flags & NGX_USE_FD_EVENT) {
78 struct rlimit rlmt;
79
80 if (getrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
81 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
82 "getrlimit(RLIMIT_NOFILE) failed");
83 return NGX_ERROR;
84 }
85
86 cycle->files_n = (ngx_uint_t) rlmt.rlim_cur;
87
88 cycle->files = ngx_calloc(sizeof(ngx_connection_t *) * cycle->files_n,
89 cycle->log);
90 if (cycle->files == NULL) {
91 return NGX_ERROR;
92 }
93 }
94
95#else
96
97 if (ngx_timer_resolution && !(ngx_event_flags & NGX_USE_TIMER_EVENT)) {
98 ngx_log_error(NGX_LOG_WARN, cycle->log, 0,
99 "the \"timer_resolution\" directive is not supported "
100 "with the configured event method, ignored");
101 ngx_timer_resolution = 0;
102 }
103
104#endif
105
106 cycle->connections =
107 ngx_alloc(sizeof(ngx_connection_t) * cycle->connection_n, cycle->log);
108 if (cycle->connections == NULL) {
109 return NGX_ERROR;
110 }
111
112 c = cycle->connections;
113
114 cycle->read_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
115 cycle->log);
116 if (cycle->read_events == NULL) {
117 return NGX_ERROR;
118 }
119
120 rev = cycle->read_events;
121 for (i = 0; i < cycle->connection_n; i++) {
122 rev[i].closed = 1;
123 rev[i].instance = 1;
124 }
125
126 cycle->write_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
127 cycle->log);
128 if (cycle->write_events == NULL) {
129 return NGX_ERROR;
130 }
131
132 wev = cycle->write_events;
133 for (i = 0; i < cycle->connection_n; i++) {
134 wev[i].closed = 1;
135 }
136
137 i = cycle->connection_n;
138 next = NULL;
139
140 do {
141 i--;
142
143 c[i].data = next;
144 c[i].read = &cycle->read_events[i];
145 c[i].write = &cycle->write_events[i];
146 c[i].fd = (ngx_socket_t) -1;
147
148 next = &c[i];
149 } while (i);
150
151 cycle->free_connections = next;
152 cycle->free_connection_n = cycle->connection_n;
153
154 /* for each listening socket */
155
156 ls = cycle->listening.elts;
157 for (i = 0; i < cycle->listening.nelts; i++) {
158
159#if (NGX_HAVE_REUSEPORT)
160 if (ls[i].reuseport && ls[i].worker != ngx_worker) {
161 continue;
162 }
163#endif
164
165 c = ngx_get_connection(ls[i].fd, cycle->log);
166
167 if (c == NULL) {
168 return NGX_ERROR;
169 }
170
171 c->type = ls[i].type;
172 c->log = &ls[i].log;
173
174 c->listening = &ls[i];
175 ls[i].connection = c;
176
177 rev = c->read;
178
179 rev->log = c->log;
180 rev->accept = 1;
181
182#if (NGX_HAVE_DEFERRED_ACCEPT)
183 rev->deferred_accept = ls[i].deferred_accept;
184#endif
185
186 if (!(ngx_event_flags & NGX_USE_IOCP_EVENT)) {
187 if (ls[i].previous) {
188
189 /*
190 * delete the old accept events that were bound to
191 * the old cycle read events array
192 */
193
194 old = ls[i].previous->connection;
195
196 if (ngx_del_event(old->read, NGX_READ_EVENT, NGX_CLOSE_EVENT)
197 == NGX_ERROR)
198 {
199 return NGX_ERROR;
200 }
201
202 old->fd = (ngx_socket_t) -1;
203 }
204 }
205
206#if (NGX_WIN32)
207#else
208 rev->handler = (c->type == SOCK_STREAM) ? ngx_event_accept
209 : ngx_event_recvmsg;
210#if (NGX_HAVE_REUSEPORT)
211
212 if (ls[i].reuseport) {
213 if (ngx_add_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
214 return NGX_ERROR;
215 }
216
217 continue;
218 }
219
220#endif
221
222 if (ngx_use_accept_mutex) {
223 continue;
224 }
225
226#if (NGX_HAVE_EPOLLEXCLUSIVE)
227
228 if ((ngx_event_flags & NGX_USE_EPOLL_EVENT)
229 && ccf->worker_processes > 1)
230 {
231 if (ngx_add_event(rev, NGX_READ_EVENT, NGX_EXCLUSIVE_EVENT)
232 == NGX_ERROR)
233 {
234 return NGX_ERROR;
235 }
236
237 continue;
238 }
239
240#endif
241
242 if (ngx_add_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
243 return NGX_ERROR;
244 }
245
246#endif
247
248 }
249
250 return NGX_OK;
251}
这个函数的主要逻辑如下;
- 获取我们的全局
ngx_core_conf_t
配置结构体和事件模块的配置结构体ngx_event_conf_t
- 初始化
ngx_posted_accept_events
和ngx_posted_events
,这两个都是队列,用来保存需要滞后处理的连接和事件。- 初始化时间红黑树
- 遍历所有的
Nginx
模块,找到我们选择的事件模块,比如ngx_epoll
模块,然后调用该事件模块的init()
方法(该方法下一篇文件进行分析)- 对时间参数进行处理(之后分析这部分)
connection
,read event
,write event
进行处理- 对
cycle->listening
中的监听端口分配connection
下面我们看一下connection
,
read event,
write event`的生成逻辑,如下:
1cycle->connections =
2 ngx_alloc(sizeof(ngx_connection_t) * cycle->connection_n, cycle->log);
3 if (cycle->connections == NULL) {
4 return NGX_ERROR;
5 }
6
7 c = cycle->connections;
8
9 cycle->read_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
10 cycle->log);
11 if (cycle->read_events == NULL) {
12 return NGX_ERROR;
13 }
14
15 rev = cycle->read_events;
16 for (i = 0; i < cycle->connection_n; i++) {
17 rev[i].closed = 1;
18 rev[i].instance = 1;
19 }
20
21 cycle->write_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
22 cycle->log);
23 if (cycle->write_events == NULL) {
24 return NGX_ERROR;
25 }
26
27 wev = cycle->write_events;
28 for (i = 0; i < cycle->connection_n; i++) {
29 wev[i].closed = 1;
30 }
31
32 i = cycle->connection_n;
33 next = NULL;
34
35 do {
36 i--;
37
38 c[i].data = next;
39 c[i].read = &cycle->read_events[i];
40 c[i].write = &cycle->write_events[i];
41 c[i].fd = (ngx_socket_t) -1;
42
43 next = &c[i];
44 } while (i);
45
46 cycle->free_connections = next;
47 cycle->free_connection_n = cycle->connection_n;
上面的代码很简单,就是根据配置文件中的参数,生成对应数量的连接数,并根据下标将连接和读写事件相关联。并把空闲连接的放置在free_connection链表中
喜欢本文的朋友们,欢迎长按下图关注订阅号郑尔多斯,更多精彩内容第一时间送达
郑尔多斯