forked from chronolaw/annotated_nginx
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathngx_event.c
1637 lines (1246 loc) · 47.8 KB
/
ngx_event.c
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// annotated by chrono since 2016
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
// 默认的epoll数组长度
#define DEFAULT_CONNECTIONS 512
// 各内置事件模块
// rtsig在1.9.x里已经删除
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_rtsig_module;
extern ngx_module_t ngx_select_module;
// 只检查是否创建了配置结构体，无其他操作
// 因为event模块只有一个events指令
static char *ngx_event_init_conf(ngx_cycle_t *cycle, void *conf);
// 在ngx_init_cycle里调用，fork子进程之前
// 创建共享内存，存放负载均衡锁和统计用的原子变量
static ngx_int_t ngx_event_module_init(ngx_cycle_t *cycle);
// 重要！
// fork之后，worker进程初始化时调用，即每个worker里都会执行
// 初始化两个延后处理的事件队列,初始化定时器红黑树
// 发送定时信号，更新时间用
// 初始化cycle里的连接和事件数组
// 设置接受连接的回调函数为ngx_event_accept，可以接受连接
static ngx_int_t ngx_event_process_init(ngx_cycle_t *cycle);
// 解析events配置块
// 设置事件模块的ctx_index
static char *ngx_events_block(ngx_conf_t *cf, ngx_command_t *cmd, void *conf);
// 解析worker_connections指令
// 取得指令字符串,转换为数字
// 再设置到cycle里，即连接池数组的大小
static char *ngx_event_connections(ngx_conf_t *cf, ngx_command_t *cmd,
void *conf);
// 解析use指令
// 决定使用哪个事件模型，linux上通常是epoll
static char *ngx_event_use(ngx_conf_t *cf, ngx_command_t *cmd, void *conf);
static char *ngx_event_debug_connection(ngx_conf_t *cf, ngx_command_t *cmd,
void *conf);
// 创建event_core模块的配置结构体，成员初始化为unset
static void *ngx_event_core_create_conf(ngx_cycle_t *cycle);
// 所有模块配置解析完毕后，对配置进行初始化
// 如果有的指令没有写，就要给正确的默认值
// 模块默认使用epoll
// 默认不接受多个请求，也就是一次只accept一个连接
// 默认使用负载均衡锁
static char *ngx_event_core_init_conf(ngx_cycle_t *cycle, void *conf);
// nginx更新缓存时间的精度，如果设置了会定时发送sigalarm信号更新时间
static ngx_uint_t ngx_timer_resolution;
// 在epoll的ngx_epoll_process_events里检查，更新时间的标志
sig_atomic_t ngx_event_timer_alarm;
// 事件模块计数器
static ngx_uint_t ngx_event_max_module;
// 事件模型的基本标志位
// 在ngx_epoll_init里设置为et模式，边缘触发
// NGX_USE_CLEAR_EVENT|NGX_USE_GREEDY_EVENT|NGX_USE_EPOLL_EVENT
// 在ngx_recv.c:ngx_unix_recv里使用，尽量多读数据
ngx_uint_t ngx_event_flags;
// 全局的事件模块访问接口，是一个函数表
// 定义了若干宏简化对它的操作
// 常用的有ngx_add_event/ngx_del_event
ngx_event_actions_t ngx_event_actions;
// 连接计数器，使用共享内存，所有worker公用
static ngx_atomic_t connection_counter = 1;
ngx_atomic_t *ngx_connection_counter = &connection_counter;
// 负载均衡锁指针，初始为空指针
ngx_atomic_t *ngx_accept_mutex_ptr;
// 负载均衡锁
ngx_shmtx_t ngx_accept_mutex;
// 负载均衡锁标志量
ngx_uint_t ngx_use_accept_mutex;
ngx_uint_t ngx_accept_events;
// 是否已经持有负载均衡锁
ngx_uint_t ngx_accept_mutex_held;
// 等待多少时间再次尝试获取负载均衡锁
// ngx_accept_mutex_delay = ecf->accept_mutex_delay;
ngx_msec_t ngx_accept_mutex_delay;
// ngx_accept_disabled是总连接数的1/8-空闲连接数
// 也就是说空闲连接数小于总数的1/8,那么就暂时停止接受连接
ngx_int_t ngx_accept_disabled;
#if (NGX_STAT_STUB)
ngx_atomic_t ngx_stat_accepted0;
ngx_atomic_t *ngx_stat_accepted = &ngx_stat_accepted0;
ngx_atomic_t ngx_stat_handled0;
ngx_atomic_t *ngx_stat_handled = &ngx_stat_handled0;
ngx_atomic_t ngx_stat_requests0;
ngx_atomic_t *ngx_stat_requests = &ngx_stat_requests0;
ngx_atomic_t ngx_stat_active0;
ngx_atomic_t *ngx_stat_active = &ngx_stat_active0;
ngx_atomic_t ngx_stat_reading0;
ngx_atomic_t *ngx_stat_reading = &ngx_stat_reading0;
ngx_atomic_t ngx_stat_writing0;
ngx_atomic_t *ngx_stat_writing = &ngx_stat_writing0;
ngx_atomic_t ngx_stat_waiting0;
ngx_atomic_t *ngx_stat_waiting = &ngx_stat_waiting0;
#endif
// events模块仅支持一个指令，即events块
static ngx_command_t ngx_events_commands[] = {
{ ngx_string("events"),
NGX_MAIN_CONF|NGX_CONF_BLOCK|NGX_CONF_NOARGS,
ngx_events_block,
0,
0,
NULL },
ngx_null_command
};
static ngx_core_module_t ngx_events_module_ctx = {
ngx_string("events"),
NULL,
// 只检查是否创建了配置结构体，无其他操作
// 因为event模块只有一个events指令
ngx_event_init_conf
};
// ngx_events_module只是组织各具体的事件模块，本身无功能
ngx_module_t ngx_events_module = {
NGX_MODULE_V1,
&ngx_events_module_ctx, /* module context */
ngx_events_commands, /* module directives */
NGX_CORE_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
// event_core模块的名字："event_core"
static ngx_str_t event_core_name = ngx_string("event_core");
static ngx_command_t ngx_event_core_commands[] = {
// nginx每个worker进程里的连接池数量，决定了nginx的服务能力
{ ngx_string("worker_connections"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_connections,
0,
0,
NULL },
// 功能同worker_connections，但已经被废弃，不要使用
{ ngx_string("connections"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_connections,
0,
0,
NULL },
// 决定使用哪个事件模型，linux上通常是epoll
{ ngx_string("use"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_use,
0,
0,
NULL },
// 默认不接受多个请求，也就是一次只accept一个连接
{ ngx_string("multi_accept"),
NGX_EVENT_CONF|NGX_CONF_FLAG,
ngx_conf_set_flag_slot,
0,
offsetof(ngx_event_conf_t, multi_accept),
NULL },
// 默认使用负载均衡锁
// accept_mutex off也是可以的，这样连接快但可能负载不均衡
{ ngx_string("accept_mutex"),
NGX_EVENT_CONF|NGX_CONF_FLAG,
ngx_conf_set_flag_slot,
0,
offsetof(ngx_event_conf_t, accept_mutex),
NULL },
// 默认负载均衡锁的等待时间是500毫秒
{ 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 },
{ ngx_string("debug_connection"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_event_debug_connection,
0,
0,
NULL },
ngx_null_command
};
// event_core模块是event模块，不是core模块
// 但它不实现具体的事件模型，所以actions函数表全是空指针
ngx_event_module_t ngx_event_core_module_ctx = {
// event_core模块的名字："event_core"
&event_core_name,
// 创建event_core模块的配置结构体，成员初始化为unset
ngx_event_core_create_conf, /* create configuration */
// 所有模块配置解析完毕后，对配置进行初始化
// 如果有的指令没有写，就要给正确的默认值
// 模块默认使用epoll
// 默认不接受多个请求，也就是一次只accept一个连接
// 默认使用负载均衡锁
ngx_event_core_init_conf, /* init configuration */
// 不实现具体的事件模型，所以actions函数表全是空指针
{ NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL }
};
ngx_module_t ngx_event_core_module = {
NGX_MODULE_V1,
&ngx_event_core_module_ctx, /* module context */
ngx_event_core_commands, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
// 在ngx_init_cycle里调用，fork子进程之前
// 创建共享内存，存放负载均衡锁和统计用的原子变量
ngx_event_module_init, /* init module */
// 初始化cycle里的连接和事件数组
// fork之后，worker进程初始化时调用，即每个worker里都会执行
// 初始化两个延后处理的事件队列,初始化定时器红黑树
// 发送定时信号，更新时间用
// 初始化cycle里的连接和事件数组
// 设置接受连接的回调函数为ngx_event_accept，可以接受连接
ngx_event_process_init, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
// 重要！！
// 在ngx_process_cycle.c:ngx_single_process_cycle/ngx_worker_process_cycle里调用
// 处理socket读写事件和定时器事件
// 获取负载均衡锁，监听端口接受连接
// 调用epoll模块的ngx_epoll_process_events获取发生的事件
// 然后处理超时事件和在延后队列里的所有事件
void
ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
ngx_uint_t flags;
ngx_msec_t timer, delta;
// ccf->timer_resolution
// nginx更新缓存时间的精度，如果设置了会定时发送sigalarm信号更新时间
if (ngx_timer_resolution) {
// 要求epoll无限等待事件的发生，直至被sigalarm信号中断
timer = NGX_TIMER_INFINITE;
flags = 0;
} else {
// 没有设置时间精度
// 在定时器红黑树里找到最小的时间，二叉树查找很快
// timer==0意味着在红黑树里已经有事件超时了，必须立即处理
timer = ngx_event_find_timer();
// NGX_UPDATE_TIME要求epoll等待这个时间，然后主动更新时间
flags = NGX_UPDATE_TIME;
// nginx 1.9.x不再使用old threads代码
#if (NGX_OLD_THREADS)
if (timer == NGX_TIMER_INFINITE || timer > 500) {
timer = 500;
}
#endif
}
// 负载均衡锁标志量， accept_mutex on
if (ngx_use_accept_mutex) {
// ngx_accept_disabled = ngx_cycle->connection_n / 8
// - ngx_cycle->free_connection_n;
// ngx_accept_disabled是总连接数的1/8-空闲连接数
// 也就是说空闲连接数小于总数的1/8,那么就暂时停止接受连接
if (ngx_accept_disabled > 0) {
// 但也不能永远不接受连接，毕竟还是有空闲连接的，所以每次要减一
ngx_accept_disabled--;
} else {
// 尝试获取负载均衡锁，开始监听端口
// 如未获取则不监听端口
// 内部调用ngx_enable_accept_events/ngx_disable_accept_events
if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
return;
}
// 确实已经获得了锁，接下来的epoll的事件需要加入延后队列处理
// 这样可以尽快释放锁给其他进程，提高运行效率
if (ngx_accept_mutex_held) {
// 加上NGX_POST_EVENTS标志
// epoll获得的所有事件都会加入到ngx_posted_events
// 待释放锁后再逐个处理，尽量避免过长时间持有锁
flags |= NGX_POST_EVENTS;
} else {
// 未获取到锁
// 要求epoll无限等待，或者等待时间超过配置的ngx_accept_mutex_delay
// 也就是说nginx的epoll不会等待超过ngx_accept_mutex_delay的500毫秒
if (timer == NGX_TIMER_INFINITE
|| timer > ngx_accept_mutex_delay)
{
// epoll的超时时间最大就是ngx_accept_mutex_delay
// ngx_accept_mutex_delay = ecf->accept_mutex_delay;
// 如果时间精度设置的太粗，那么就使用这个时间,500毫秒
timer = ngx_accept_mutex_delay;
}
}
}
}
// 不管是否获得了负载均衡锁，都要处理事件和定时器
// 如果获得了负载均衡锁,事件就会多出一个accept事件
// 否则只有普通的读写事件和定时器事件
// 获取当前的时间，毫秒数
delta = ngx_current_msec;
// #define ngx_process_events ngx_event_actions.process_events
// 实际上就是ngx_epoll_process_events
//
// epoll模块核心功能，调用epoll_wait处理发生的事件
// 使用event_list和nevents获取内核返回的事件
// timer是无事件发生时最多等待的时间，即超时时间
// 如果ngx_event_find_timer返回timer==0，那么epoll不会等待，立即返回
// 函数可以分为两部分，一是用epoll获得事件，二是处理事件，加入延后队列
(void) ngx_process_events(cycle, timer, flags);
// 在ngx_process_events里缓存的时间肯定已经更新
// 计算得到epoll一次调用消耗的毫秒数
delta = ngx_current_msec - delta;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"timer delta: %M", delta);
// 先处理连接事件，通常只有一个accept的连接
// in ngx_event_posted.c
// 实际上调用的就是ngx_event_accept
// 在http模块里是http.c:ngx_http_init_connection
ngx_event_process_posted(cycle, &ngx_posted_accept_events);
// 释放锁，其他进程可以获取，再监听端口
// 这里只处理accept事件，工作量小，可以尽快释放锁，供其他进程使用
if (ngx_accept_mutex_held) {
// 释放负载均衡锁
// 其他进程等待ngx_accept_mutex_delay毫秒后
// 再走ngx_trylock_accept_mutex决定端口的监听权
ngx_shmtx_unlock(&ngx_accept_mutex);
}
// 如果消耗了一点时间，那么看看是否定时器里有过期的
if (delta) {
// 遍历定时器红黑树，找出所有过期的事件，调用handler处理超时
ngx_event_expire_timers();
}
// 接下来处理延后队列里的事件，即调用事件的handler(ev)，收发数据
// in ngx_event_posted.c
// 这里因为要处理大量的事件，而且是简单的顺序调用，所以可能会阻塞
// nginx大部分的工作量都在这里
ngx_event_process_posted(cycle, &ngx_posted_events);
}
// 添加读事件的便捷接口，适合epoll/kqueue/select等各种事件模型
// 内部还是调用ngx_add_event
ngx_int_t
ngx_handle_read_event(ngx_event_t *rev, ngx_uint_t flags)
{
// 使用et模式，epoll/kqueue
if (ngx_event_flags & NGX_USE_CLEAR_EVENT) {
/* kqueue, epoll */
if (!rev->active && !rev->ready) {
if (ngx_add_event(rev, NGX_READ_EVENT, NGX_CLEAR_EVENT)
== NGX_ERROR)
{
return NGX_ERROR;
}
}
return NGX_OK;
} else if (ngx_event_flags & NGX_USE_LEVEL_EVENT) {
/* select, poll, /dev/poll */
if (!rev->active && !rev->ready) {
if (ngx_add_event(rev, NGX_READ_EVENT, NGX_LEVEL_EVENT)
== NGX_ERROR)
{
return NGX_ERROR;
}
return NGX_OK;
}
if (rev->active && (rev->ready || (flags & NGX_CLOSE_EVENT))) {
if (ngx_del_event(rev, NGX_READ_EVENT, NGX_LEVEL_EVENT | flags)
== NGX_ERROR)
{
return NGX_ERROR;
}
return NGX_OK;
}
} else if (ngx_event_flags & NGX_USE_EVENTPORT_EVENT) {
/* event ports */
if (!rev->active && !rev->ready) {
if (ngx_add_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
return NGX_OK;
}
if (rev->oneshot && !rev->ready) {
if (ngx_del_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
return NGX_OK;
}
}
/* aio, iocp, rtsig */
return NGX_OK;
}
// 添加写事件的便捷接口，适合epoll/kqueue/select等各种事件模型
// 内部还是调用ngx_add_event,多了个send_lowat操作
// linux不支持send_lowat指令，send_lowat总是0
ngx_int_t
ngx_handle_write_event(ngx_event_t *wev, size_t lowat)
{
ngx_connection_t *c;
if (lowat) {
c = wev->data;
// 设置发送数据时epoll的响应阈值
// 当系统空闲缓冲超过lowat时触发epoll可写事件
// linux不支持send_lowat指令，send_lowat总是0
if (ngx_send_lowat(c, lowat) == NGX_ERROR) {
return NGX_ERROR;
}
}
if (ngx_event_flags & NGX_USE_CLEAR_EVENT) {
/* kqueue, epoll */
if (!wev->active && !wev->ready) {
if (ngx_add_event(wev, NGX_WRITE_EVENT,
NGX_CLEAR_EVENT | (lowat ? NGX_LOWAT_EVENT : 0))
== NGX_ERROR)
{
return NGX_ERROR;
}
}
return NGX_OK;
} else if (ngx_event_flags & NGX_USE_LEVEL_EVENT) {
/* select, poll, /dev/poll */
if (!wev->active && !wev->ready) {
if (ngx_add_event(wev, NGX_WRITE_EVENT, NGX_LEVEL_EVENT)
== NGX_ERROR)
{
return NGX_ERROR;
}
return NGX_OK;
}
if (wev->active && wev->ready) {
if (ngx_del_event(wev, NGX_WRITE_EVENT, NGX_LEVEL_EVENT)
== NGX_ERROR)
{
return NGX_ERROR;
}
return NGX_OK;
}
} else if (ngx_event_flags & NGX_USE_EVENTPORT_EVENT) {
/* event ports */
if (!wev->active && !wev->ready) {
if (ngx_add_event(wev, NGX_WRITE_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
return NGX_OK;
}
if (wev->oneshot && wev->ready) {
if (ngx_del_event(wev, NGX_WRITE_EVENT, 0) == NGX_ERROR) {
return NGX_ERROR;
}
return NGX_OK;
}
}
/* aio, iocp, rtsig */
return NGX_OK;
}
// 只检查是否创建了配置结构体，无其他操作
// 因为event模块只有一个events指令
static char *
ngx_event_init_conf(ngx_cycle_t *cycle, void *conf)
{
if (ngx_get_conf(cycle->conf_ctx, ngx_events_module) == NULL) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, 0,
"no \"events\" section in configuration");
return NGX_CONF_ERROR;
}
return NGX_CONF_OK;
}
// 在ngx_init_cycle里调用，fork子进程之前
// 创建共享内存，存放负载均衡锁和统计用的原子变量
static ngx_int_t
ngx_event_module_init(ngx_cycle_t *cycle)
{
void ***cf;
u_char *shared;
size_t size, cl;
ngx_shm_t shm;
ngx_time_t *tp;
ngx_core_conf_t *ccf;
ngx_event_conf_t *ecf;
// events模块的配置结构体
cf = ngx_get_conf(cycle->conf_ctx, ngx_events_module);
// event_core模块的配置结构体
ecf = (*cf)[ngx_event_core_module.ctx_index];
if (!ngx_test_config && ngx_process <= NGX_PROCESS_MASTER) {
ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0,
"using the \"%s\" event method", ecf->name);
}
// core模块的配置结构体
ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);
// 获取核心配置的时间精度，用在epoll里更新缓存时间
ngx_timer_resolution = ccf->timer_resolution;
// unix专用代码, core dump相关
#if !(NGX_WIN32)
{
ngx_int_t limit;
struct rlimit rlmt;
if (getrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"getrlimit(RLIMIT_NOFILE) failed, ignored");
} else {
if (ecf->connections > (ngx_uint_t) rlmt.rlim_cur
&& (ccf->rlimit_nofile == NGX_CONF_UNSET
|| ecf->connections > (ngx_uint_t) ccf->rlimit_nofile))
{
limit = (ccf->rlimit_nofile == NGX_CONF_UNSET) ?
(ngx_int_t) rlmt.rlim_cur : ccf->rlimit_nofile;
ngx_log_error(NGX_LOG_WARN, cycle->log, 0,
"%ui worker_connections exceed "
"open file resource limit: %i",
ecf->connections, limit);
}
}
}
#endif /* !(NGX_WIN32) */
// 如果非master/worker进程，即只启动一个进程，那么就没必要使用负载均衡锁
if (ccf->master == 0) {
return NGX_OK;
}
// 已经有了负载均衡锁，已经初始化过了，就没必要再做操作
if (ngx_accept_mutex_ptr) {
return NGX_OK;
}
/* cl should be equal to or greater than cache line size */
// cl是一个基本长度，可以容纳原子变量
cl = 128;
// 最基本的三个：负载均衡锁，连接计数器,
size = cl /* ngx_accept_mutex */
+ cl /* ngx_connection_counter */
+ cl; /* ngx_temp_number */
// 其他统计用的原子变量
#if (NGX_STAT_STUB)
size += cl /* ngx_stat_accepted */
+ cl /* ngx_stat_handled */
+ cl /* ngx_stat_requests */
+ cl /* ngx_stat_active */
+ cl /* ngx_stat_reading */
+ cl /* ngx_stat_writing */
+ cl; /* ngx_stat_waiting */
#endif
// 创建共享内存，存放负载均衡锁和统计用的原子变量
shm.size = size;
shm.name.len = sizeof("nginx_shared_zone");
shm.name.data = (u_char *) "nginx_shared_zone";
shm.log = cycle->log;
if (ngx_shm_alloc(&shm) != NGX_OK) {
return NGX_ERROR;
}
// shared是共享内存的地址指针
shared = shm.addr;
// 第一个就是负载均衡锁
ngx_accept_mutex_ptr = (ngx_atomic_t *) shared;
ngx_accept_mutex.spin = (ngx_uint_t) -1;
if (ngx_shmtx_create(&ngx_accept_mutex, (ngx_shmtx_sh_t *) shared,
cycle->lock_file.data)
!= NGX_OK)
{
return NGX_ERROR;
}
// 连接计数器
ngx_connection_counter = (ngx_atomic_t *) (shared + 1 * cl);
(void) ngx_atomic_cmp_set(ngx_connection_counter, 0, 1);
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"counter: %p, %d",
ngx_connection_counter, *ngx_connection_counter);
// 临时文件用
ngx_temp_number = (ngx_atomic_t *) (shared + 2 * cl);
tp = ngx_timeofday();
// 随机数
ngx_random_number = (tp->msec << 16) + ngx_pid;
#if (NGX_STAT_STUB)
ngx_stat_accepted = (ngx_atomic_t *) (shared + 3 * cl);
ngx_stat_handled = (ngx_atomic_t *) (shared + 4 * cl);
ngx_stat_requests = (ngx_atomic_t *) (shared + 5 * cl);
ngx_stat_active = (ngx_atomic_t *) (shared + 6 * cl);
ngx_stat_reading = (ngx_atomic_t *) (shared + 7 * cl);
ngx_stat_writing = (ngx_atomic_t *) (shared + 8 * cl);
ngx_stat_waiting = (ngx_atomic_t *) (shared + 9 * cl);
#endif
return NGX_OK;
}
#if !(NGX_WIN32)
// sigalarm信号的处理函数，只设置ngx_event_timer_alarm变量
// 在epoll的ngx_epoll_process_events里检查，更新时间的标志
// 信号处理函数应该尽量简单，避免阻塞进程
static void
ngx_timer_signal_handler(int signo)
{
ngx_event_timer_alarm = 1;
#if 1
ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ngx_cycle->log, 0, "timer signal");
#endif
}
#endif
// fork之后，worker进程初始化时调用，即每个worker里都会执行
// 初始化两个延后处理的事件队列,初始化定时器红黑树
// 发送定时信号，更新时间用
// 初始化cycle里的连接和事件数组
// 设置接受连接的回调函数为ngx_event_accept，可以接受连接
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;
// core模块的配置结构体
ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);
// event_core模块的配置结构体
ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);
// 使用master/worker多进程，使用负载均衡
if (ccf->master && ccf->worker_processes > 1 && ecf->accept_mutex) {
// 设置全局变量
// 使用负载均衡，刚开始未持有锁，设置抢锁的等待事件
ngx_use_accept_mutex = 1;
ngx_accept_mutex_held = 0;
ngx_accept_mutex_delay = ecf->accept_mutex_delay;
} else {
// 单进程、未明确指定负载均衡，就不使用负载均衡
ngx_use_accept_mutex = 0;
}
#if (NGX_WIN32)
/*
* disable accept mutex on win32 as it may cause deadlock if
* grabbed by a process which can't accept connections
*/
ngx_use_accept_mutex = 0;
#endif
// 初始化两个延后处理的事件队列
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;
}
// 遍历事件模块，但只执行实际使用的事件模块对应初始化函数
for (m = 0; ngx_modules[m]; m++) {
if (ngx_modules[m]->type != NGX_EVENT_MODULE) {
continue;
}
// 找到use指令使用的事件模型，或者是默认事件模型
if (ngx_modules[m]->ctx_index != ecf->use) {
continue;
}
module = ngx_modules[m]->ctx;
// 调用事件模块的事件初始化函数
//
// 调用epoll_create初始化epoll机制
// 参数size=cycle->connection_n / 2，但并无实际意义
// 设置全局变量，操作系统提供的底层数据收发接口
// 初始化全局的事件模块访问接口，指向epoll的函数
// 默认使用et模式，边缘触发，高速
if (module->actions.init(cycle, ngx_timer_resolution) != NGX_OK) {
/* fatal */
exit(2);
}
break;
}
// unix代码, 发送定时信号，更新时间用
#if !(NGX_WIN32)
if (ngx_timer_resolution && !(ngx_event_flags & NGX_USE_TIMER_EVENT)) {
struct sigaction sa;
struct itimerval itv;
// 设置信号掩码，sigalarm
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;
}
// 设置信号发送的时间间隔，也就是nginx的时间精度
// 收到信号会设置设置ngx_event_timer_alarm变量
// 在epoll的ngx_epoll_process_events里检查，更新时间的标志
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;
if (setitimer(ITIMER_REAL, &itv, NULL) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"setitimer() failed");
}
}
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
// 创建连接池数组，大小是cycle->connection_n
// 直接使用malloc分配内存，没有使用内存池
cycle->connections =
ngx_alloc(sizeof(ngx_connection_t) * cycle->connection_n, cycle->log);
if (cycle->connections == NULL) {
return NGX_ERROR;
}
c = cycle->connections;
// 创建读事件池数组，大小是cycle->connection_n
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;
}
// 创建写事件池数组，大小是cycle->connection_n
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;
}
i = cycle->connection_n;
next = NULL;
// 把连接对象与读写事件关联起来
// 注意i是数组的末尾，从最后遍历
do {
i--;
// 使用data成员，把连接对象串成链表
c[i].data = next;
// 读写事件
c[i].read = &cycle->read_events[i];
c[i].write = &cycle->write_events[i];
// 连接的描述符是-1，表示无效
c[i].fd = (ngx_socket_t) -1;
// next指针指向数组的前一个元素
next = &c[i];
} while (i);
// 连接对象已经串成链表，现在设置空闲链表指针
// 此时next指向连接对象数组的第一个元素
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;
c->listening = &ls[i];
ls[i].connection = c;
rev = c->read;
rev->log = c->log;
// 设置accept标志，接受连接
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) {