forked from chronolaw/annotated_nginx
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathngx_event.c
1806 lines (1388 loc) · 55.1 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
//
// * ngx_process_events_and_timers
// * ngx_event_module_init
// * ngx_event_process_init
// * ngx_handle_read_event
// * ngx_handle_write_event
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
// 默认的epoll数组长度
// 最多同时处理512个连接，太少
// 通常可以配置到32000或者更多
#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;
// 通常我们在Linux上只使用epoll
extern ngx_module_t ngx_epoll_module;
extern ngx_module_t ngx_select_module;
// 1.15.2之前只检查是否创建了配置结构体，无其他操作
// 因为event模块只有一个events指令
// 1.15.2之后增加新的代码
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里，即连接池数组的大小
// 决定了nginx同时能够处理的最大连接数量
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一个连接
// 1.11.3之前默认使用负载均衡锁，之后默认关闭
static char *ngx_event_core_init_conf(ngx_cycle_t *cycle, void *conf);
// nginx更新缓存时间的精度，如果设置了会定时发送sigalarm信号更新时间
// ngx_timer_resolution = ccf->timer_resolution;默认值是0
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_connection_counter初始指向静态变量connection_counter
// 如果是单进程，那么就使用这个静态变量
// 如果是多进程，那么就改指向共享内存里的地址
ngx_atomic_t *ngx_connection_counter = &connection_counter;
// 1.9.x如果使用了reuseport，那么就会禁用负载均衡锁
// 由linux系统内核来实现负载均衡，选择恰当的进程接受连接
// 由于锁会影响性能，所以1.11.3开始nginx默认不使用锁
// 负载均衡锁指针，初始为空指针
ngx_atomic_t *ngx_accept_mutex_ptr;
// 负载均衡锁
ngx_shmtx_t ngx_accept_mutex;
// 负载均衡锁标志量
ngx_uint_t ngx_use_accept_mutex;
// ngx_accept_events在epoll里不使用，暂不关注
ngx_uint_t ngx_accept_events;
// 是否已经持有负载均衡锁
// in ngx_event_accept.c:ngx_trylock_accept_mutex
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,那么就暂时停止接受连接
// in ngx_event_accept.c:ngx_event_accept
ngx_int_t ngx_accept_disabled;
// 统计用的共享内存变量指针
#if (NGX_STAT_STUB)
static ngx_atomic_t ngx_stat_accepted0;
ngx_atomic_t *ngx_stat_accepted = &ngx_stat_accepted0;
static ngx_atomic_t ngx_stat_handled0;
ngx_atomic_t *ngx_stat_handled = &ngx_stat_handled0;
static ngx_atomic_t ngx_stat_requests0;
ngx_atomic_t *ngx_stat_requests = &ngx_stat_requests0;
static ngx_atomic_t ngx_stat_active0;
ngx_atomic_t *ngx_stat_active = &ngx_stat_active0;
static ngx_atomic_t ngx_stat_reading0;
ngx_atomic_t *ngx_stat_reading = &ngx_stat_reading0;
static ngx_atomic_t ngx_stat_writing0;
ngx_atomic_t *ngx_stat_writing = &ngx_stat_writing0;
static 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,
// 1.15.2之前只检查是否创建了配置结构体，无其他操作
// 因为event模块只有一个events指令
// 1.15.2之后增加新的代码
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"),...}
// 决定使用哪个事件模型，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也是可以的，这样连接快但可能负载不均衡
// 1.10后支持reuseport，可以不使用此指令
// 1.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 },
// 默认负载均衡锁的等待时间是500毫秒
// 不持有锁的其他进程最多等待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函数表全是空指针
static 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一个连接
// 1.11.3之前默认使用负载均衡锁，之后默认关闭
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信号更新时间
// ngx_timer_resolution = ccf->timer_resolution;默认值是0
if (ngx_timer_resolution) {
// 要求epoll无限等待事件的发生，直至被sigalarm信号中断
timer = NGX_TIMER_INFINITE;
flags = 0;
} else {
// 没有设置时间精度，默认设置
// 在定时器红黑树里找到最小的时间，二叉树查找很快
// timer >0 红黑树里即将超时的事件的时间
// timer <0 表示红黑树为空，即无超时事件
// timer==0意味着在红黑树里已经有事件超时了，必须立即处理
// timer==0，epoll就不会等待，收集完事件立即返回
timer = ngx_event_find_timer();
// NGX_UPDATE_TIME要求epoll等待这个时间，然后主动更新时间
flags = NGX_UPDATE_TIME;
// nginx 1.9.x不再使用old threads代码
#if (NGX_WIN32)
/* handle signals from master in case of network inactivity */
if (timer == NGX_TIMER_INFINITE || timer > 500) {
timer = 500;
}
#endif
}
// 现在已经设置了合适的timer和flag
// 负载均衡锁标志量， accept_mutex on
// 1.9.x，如果使用了reuseport，那么ngx_use_accept_mutex==0
//
// 1.11.3开始，默认不使用负载均衡锁，提高性能，下面的代码直接跳过
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) {
// 如果监听失败，那么直接结束函数，不处理epoll事件
return;
}
// ngx_trylock_accept_mutex执行成功
// 使用变量ngx_accept_mutex_held检查是否成功获取了锁
// 确实已经获得了锁，接下来的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毫秒
// 如果epoll有事件发生，那么此等待时间无意义，epoll_wait立即返回
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;
}
}
}
} //ngx_use_accept_mutex
// 如果不使用负载均衡，或者没有抢到锁
// 那么就不会使用延后处理队列，即没有NGX_POST_EVENTS标志
// 1.11.3开始，默认不使用负载均衡锁，提高性能
// 省去了锁操作和队列操作
// 不管是否获得了负载均衡锁，都要处理事件和定时器
// 如果获得了负载均衡锁,事件就会多出一个accept事件
// 否则只有普通的读写事件和定时器事件
// 1.17.5新增,处理ngx_posted_next_events
if (!ngx_queue_empty(&ngx_posted_next_events)) {
ngx_event_move_posted_next(cycle);
timer = 0;
}
// 获取当前的时间，毫秒数
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获得事件，二是处理事件，加入延后队列
//
// 如果不使用负载均衡（accept_mutex off）
// 那么所有IO事件均在此函数里处理，即搜集事件并调用handler
(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
//
// 如果不使用负载均衡（accept_mutex off）或者reuseport
// 那么此处就是空操作，因为队列为空
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);
}
// 1.19.9
// 如果消耗了一点时间，那么看看是否定时器里有过期的
//if (delta) {
// ngx_event_expire_timers();
//}
// 遍历定时器红黑树，找出所有过期的事件，调用handler处理超时
// 其中可能有的socket读写超时，那么就结束请求，断开连接
ngx_event_expire_timers();
// 接下来处理延后队列里的事件，即调用事件的handler(ev)，收发数据
// in ngx_event_posted.c
// 这里因为要处理大量的事件，而且是简单的顺序调用，所以可能会阻塞
// nginx大部分的工作量都在这里
// 注意与accept的函数是相同的，但队列不同，即里面的事件不同
//
// 如果不使用负载均衡（accept_mutex off）或者reuseport
// 那么此处就是空操作，因为队列为空
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;
}
}
/* iocp */
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;
}
}
/* iocp */
return NGX_OK;
}
// 1.15.2之前只检查是否创建了配置结构体，无其他操作
// 因为event模块只有一个events指令
// 1.15.2之后增加新的代码
static char *
ngx_event_init_conf(ngx_cycle_t *cycle, void *conf)
{
// 1.15.2新增部分检查代码
#if (NGX_HAVE_REUSEPORT)
ngx_uint_t i;
ngx_listening_t *ls;
#endif
// 要求必须有events{}配置块
// 1.15.2之前只有这一段
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;
}
// 检查连接数，需要大于监听端口数量
if (cycle->connection_n < cycle->listening.nelts + 1) {
/*
* there should be at least one connection for each listening
* socket, plus an additional connection for channel
*/
ngx_log_error(NGX_LOG_EMERG, cycle->log, 0,
"%ui worker_connections are not enough "
"for %ui listening sockets",
cycle->connection_n, cycle->listening.nelts);
return NGX_CONF_ERROR;
}
#if (NGX_HAVE_REUSEPORT)
if (!ngx_test_config) {
ls = cycle->listening.elts;
for (i = 0; i < cycle->listening.nelts; i++) {
if (!ls[i].reuseport || ls[i].worker != 0) {
continue;
}
// reuseport专用的函数，1.8.x没有
// 拷贝了worker数量个的监听结构体, in ngx_connection.c
// 从ngx_stream_optimize_servers等函数处转移过来
if (ngx_clone_listening(cycle, &ls[i]) != NGX_OK) {
return NGX_CONF_ERROR;
}
/* cloning may change cycle->listening.elts */
ls = cycle->listening.elts;
}
}
#endif
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模块的配置结构体
// 实际上是一个存储void*指针的数组
cf = ngx_get_conf(cycle->conf_ctx, ngx_events_module);
// event_core模块的配置结构体
// 从数组cf里按序号查找
ecf = (*cf)[ngx_event_core_module.ctx_index];
// 上面的两行代码相当于:
// ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module)
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专用代码, 可打开的最多文件描述符
#if !(NGX_WIN32)
{
ngx_int_t limit;
struct rlimit rlmt;
// 系统调用getrlimit，Linux内核对进程的限制
// RLIMIT_NOFILE,进程可打开的最大文件描述符数量，超出将产生EMFILE错误
if (getrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
// 系统调用失败则记录alert级别日志
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"getrlimit(RLIMIT_NOFILE) failed, ignored");
} else {
// 成功获取内核参数
//
// rlmt.rlim_cur是系统的软限制
// event里配置的连接数不能超过系统内核限制
// 或者是配置的rlimit_nofile限制
if (ecf->connections > (ngx_uint_t) rlmt.rlim_cur
&& (ccf->rlimit_nofile == NGX_CONF_UNSET
|| ecf->connections > (ngx_uint_t) ccf->rlimit_nofile))
{
// 如果超过了报警告级别日志
// limit就是上限
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是一个基本长度，可以容纳原子变量
// 对齐到cache line，操作更快
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
// 创建共享内存，存放负载均衡锁和统计用的原子变量
// 因为内存很小，而且仅用做统计，比较简单
// 所以不用slab管理
shm.size = size;
ngx_str_set(&shm.name, "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;
// spin是-1则不使用信号量
// 只会自旋，不会导致进程睡眠等待
// 这样避免抢accept锁时的性能降低
ngx_accept_mutex.spin = (ngx_uint_t) -1;
// 初始化互斥锁
// spin是-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);
// 计数器置1
(void) ngx_atomic_cmp_set(ngx_connection_counter, 0, 1);
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"counter: %p, %uA",
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_next_events);
ngx_queue_init(&ngx_posted_events);
// 初始化定时器红黑树
if (ngx_event_timer_init(cycle->log) == NGX_ERROR) {
return NGX_ERROR;
}
// 遍历事件模块，但只执行实际使用的事件模块对应初始化函数
for (m = 0; cycle->modules[m]; m++) {
if (cycle->modules[m]->type != NGX_EVENT_MODULE) {
continue;
}
// 找到use指令使用的事件模型，或者是默认事件模型
if (cycle->modules[m]->ctx_index != ecf->use) {
continue;
}
// 找到事件模块
module = cycle->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);
}
// 找到一个事件模块即退出循环