1. nginx 连接结构 ngx_connection_t 这个连接表示是客户端主动发起的、Nginx服务器被动接受的TCP连接，我们可以简单称其为被动连接。同时，在有些请求的处理过程中，Nginx会试图主动向其他上游服务器建立连接，并以此连接与上游服务器通信，因此，这样的连接与ngx_connection_t又是不同的，Nginx定义了ngx_peer_connection_t结构体来表示主动连接，当然，ngx_peer_connection_t主动连接是以ngx_connection-t结构体为基础实现的
/*一个ngx_connection_s对应一个ngx_event_s read和一个ngx_event_s write,其中事件的fd是从ngx_connection_s->fd获取，他们 在ngx_worker_process_init->ngx_event_process_init中关联起来 ngx_event_t事件和ngx_connection_t连接是处理TCP连接的基础数据结构, 通过ngx_get_connection从连接池中获取一个ngx_connection_s结构， 被动连接(客户端连接nginx)对应的数据结构是ngx_connection_s，主动连接(nginx连接后端服务器)对应的数据结构是ngx_peer_connection_s */ struct ngx_connection_s { /* // 连接未使用时，充当连接池空闲链表中的 next 指针 连接未使用时，data成员用于充当连接池中空闲连接链表中的next指针(ngx_event_process_init)。当连接被使用时，data的意义由使用它的Nginx模块而定， 如在HTTP框架中，data指向ngx_http_request_t请求 在服务器端accept客户端连接成功(ngx_event_accept)后，会通过ngx_get_connection从连接池获取一个ngx_connection_t结构，也就是每个客户端连接对于一个ngx_connection_t结构， 并且为其分配一个ngx_http_connection_t结构，ngx_connection_t->data = ngx_http_connection_t，见ngx_http_init_connection */ /* 在子请求处理过程中，上层父请求r的data指向第一个r下层的子请求，例如第二层的r->connection->data指向其第三层的第一个 创建的子请求r，c->data = sr见ngx_http_subrequest,在subrequest往客户端发送数据的时候，只有data指向的节点可以先发送出去 listen过程中，指向原始请求ngx_http_connection_t(ngx_http_init_connection ngx_http_ssl_handshake),接收到客户端数据后指向ngx_http_request_t(ngx_http_wait_request_handler) http2协议的过程中，在ngx_http_v2_connection_t(ngx_http_v2_init) */ void *data; /* 如果是subrequest，则data最终指向最下层子请求r,见ngx_http_subrequest */ //如果是文件异步i/o中的ngx_event_aio_t，则它来自ngx_event_aio_t->ngx_event_t(只有读),如果是网络事件中的event,则为ngx_connection_s中的event(包括读和写) ngx_event_t *read;//连接对应的读事件 赋值在ngx_event_process_init，空间是从ngx_cycle_t->read_event池子中获取的 ngx_event_t *write; //连接对应的写事件 赋值在ngx_event_process_init 一般在ngx_handle_write_event中添加些事件，空间是从ngx_cycle_t->read_event池子中获取的 ngx_socket_t fd;// 连接 fd 套接字句柄 /* 如果启用了ssl,则发送和接收数据在ngx_ssl_recv ngx_ssl_write ngx_ssl_recv_chain ngx_ssl_send_chain */ //服务端通过ngx_http_wait_request_handler读取数据 ngx_recv_pt recv; //直接接收网络字符流的方法 见ngx_event_accept或者ngx_http_upstream_connect 赋值为ngx_os_io 在接收到客户端连接或者向上游服务器发起连接后赋值 ngx_send_pt send; //直接发送网络字符流的方法 见ngx_event_accept或者ngx_http_upstream_connect 赋值为ngx_os_io 在接收到客户端连接或者向上游服务器发起连接后赋值 /* 如果启用了ssl,则发送和接收数据在ngx_ssl_recv ngx_ssl_write ngx_ssl_recv_chain ngx_ssl_send_chain */ //以ngx_chain_t链表为参数来接收网络字符流的方法 ngx_recv_chain ngx_recv_chain_pt recv_chain; //赋值见ngx_event_accept ngx_event_pipe_read_upstream中执行 //以ngx_chain_t链表为参数来发送网络字符流的方法 ngx_send_chain //当http2头部帧发送的时候，会在ngx_http_v2_header_filter把ngx_http_v2_send_chain.send_chain=ngx_http_v2_send_chain ngx_send_chain_pt send_chain; //赋值见ngx_event_accept ngx_http_write_filter和ngx_chain_writer中执行 //这个连接对应的ngx_listening_t监听对象,通过listen配置项配置，此连接由listening监听端口的事件建立,赋值在ngx_event_process_init //接收到客户端连接后会冲连接池分配一个ngx_connection_s结构，其listening成员指向服务器接受该连接的listen信息结构，见ngx_event_accept ngx_listening_t *listening; //实际上是从cycle->listening.elts中的一个ngx_listening_t off_t sent;//这个连接上已经发送出去的字节数 //ngx_linux_sendfile_chain和ngx_writev_chain没发送多少字节就加多少字节 ngx_log_t *log;//可以记录日志的ngx_log_t对象 其实就是ngx_listening_t中获取的log //赋值见ngx_event_accept /* 内存池。一般在accept -个新连接时，会创建一个内存池，而在这个连接结束时会销毁内存池。注意，这里所说的连接是指成功建立的 TCP连接，所有的ngx_connection_t结构体都是预分配的。这个内存池的大小将由listening监听对象中的pool_size成员决定 // 一般在accept一个新的连接时，会创建一个内存池 而在这个连接结束时会销毁内存池 // 内存池大小是由 listening 成员的 pool_size 决定的 */ ngx_pool_t *pool; //在accept返回成功后创建poll,见ngx_event_accept， 连接上游服务区的时候在ngx_http_upstream_connect创建 struct sockaddr *sockaddr; //连接客户端的sockaddr结构体 客户端的，本端的为下面的local_sockaddr 赋值见ngx_event_accept socklen_t socklen; //sockaddr结构体的长度 //赋值见ngx_event_accept ngx_str_t addr_text; //连接客户端字符串形式的IP地址 // 代理协议地址 ngx_str_t proxy_protocol_addr; #if (NGX_SSL) ngx_ssl_connection_t *ssl; //赋值见ngx_ssl_create_connection #endif //本机的监听端口对应的sockaddr结构体，也就是listening监听对象中的sockaddr成员 struct sockaddr *local_sockaddr; //赋值见ngx_event_accept socklen_t local_socklen; /* 用于接收、缓存客户端发来的字符流，每个事件消费模块可自由决定从连接池中分配多大的空间给buffer这个接收缓存字段。 例如，在HTTP模块中，它的大小决定于client_header_buffer_size配置项 */ ngx_buf_t *buffer; //ngx_http_request_t->header_in指向该缓冲去 /* 该字段用来将当前连接以双向链表元素的形式添加到ngx_cycle_t核心结构体的reusable_connections_queue双向链表中，表示可以重用的连接 */ ngx_queue_t queue; /* 连接使用次数。ngx_connection t结构体每次建立一条来自客户端的连接，或者用于主动向后端服务器发起连接时（ngx_peer_connection_t也使用它）， number都会加l */ ngx_atomic_uint_t number; //这个应该是记录当前连接是整个连接中的第几个连接，见ngx_event_accept ngx_event_connect_peer ngx_uint_t requests; //处理的请求次数 /* 缓存中的业务类型。任何事件消费模块都可以自定义需要的标志位。这个buffered字段有8位，最多可以同时表示8个不同的业务。第三方模 块在自定义buffered标志位时注意不要与可能使用的模块定义的标志位冲突。目前openssl模块定义了一个标志位： #define NGX_SSL_BUFFERED Ox01 HTTP官方模块定义了以下标志位： #define NGX HTTP_LOWLEVEL_BUFFERED 0xf0 #define NGX_HTTP_WRITE_BUFFERED 0x10 #define NGX_HTTP_GZIP_BUFFERED 0x20 #define NGX_HTTP_SSI_BUFFERED 0x01 #define NGX_HTTP_SUB_BUFFERED 0x02 #define NGX_HTTP_COPY_BUFFERED 0x04 #define NGX_HTTP_IMAGE_BUFFERED Ox08 同时，对于HTTP模块而言，buffered的低4位要慎用，在实际发送响应的ngx_http_write_filter_module过滤模块中，低4位标志位为1则惫味着 Nginx会一直认为有HTTP模块还需要处理这个请求，必须等待HTTP模块将低4位全置为0才会正常结束请求。检查低4位的宏如下： #define NGX_LOWLEVEL_BUFFERED OxOf */ unsigned buffered:8; //不为0，表示有数据没有发送完毕，ngx_http_request_t->out中还有未发送的报文 /* 本连接记录日志时的级别，它占用了3位，取值范围是0-7，但实际上目前只定义了5个值，由ngx_connection_log_error_e枚举表示，如下： typedef enum{ NGXi ERROR—AIERT=0， NGX' ERROR ERR, NGX ERROR_INFO， NGX, ERROR IGNORE ECONNRESET, NGX ERROR—IGNORE EIb:fVAL } ngx_connection_log_error_e ; */ unsigned log_error:3; /* ngx_connection_log_error_e */ //标志位，为1时表示不期待字符流结束，目前无意义 unsigned unexpected_eof:1; //每次处理完一个客户端请求后，都会ngx_add_timer(rev, c->listening->post_accept_timeout); /*读客户端连接的数据，在ngx_http_init_connection(ngx_connection_t *c)中的ngx_add_timer(rev, c->listening->post_accept_timeout)把读事件添加到定时器中，如果超时则置1 每次ngx_unix_recv把内核数据读取完毕后，在重新启动add epoll，等待新的数据到来，同时会启动定时器ngx_add_timer(rev, c->listening->post_accept_timeout); 如果在post_accept_timeout这么长事件内没有数据到来则超时，开始处理关闭TCP流程*/ //当ngx_event_t->timedout置1的时候，该置也同时会置1，参考ngx_http_process_request_line ngx_http_process_request_headers //在ngx_http_free_request中如果超时则会设置SO_LINGER来减少time_wait状态 unsigned timedout:1; //标志位，为1时表示连接已经超时,也就是过了post_accept_timeout多少秒还没有收到客户端的请求 unsigned error:1; //标志位，为1时表示连接处理过程中出现错误 /* 标志位，为1时表示连接已经销毁。这里的连接指是的TCP连接，而不是ngx_connection t结构体。当destroyed为1时，ngx_connection_t结 构体仍然存在，但其对应的套接字、内存池等已经不可用 */ unsigned destroyed:1; //ngx_http_close_connection中置1 unsigned idle:1; //为1时表示连接处于空闲状态，如keepalive请求中丽次请求之间的状态 unsigned reusable:1; //为1时表示连接可重用，它与上面的queue字段是对应使用的 unsigned close:1; //为1时表示连接关闭 /* 和后端的ngx_connection_t在ngx_event_connect_peer这里置为1，但在ngx_http_upstream_connect中c->sendfile &= r->connection->sendfile;， 和客户端浏览器的ngx_connextion_t的sendfile需要在ngx_http_update_location_config中判断，因此最终是由是否在configure的时候是否有加 sendfile选项来决定是置1还是置0 */ //赋值见ngx_http_update_location_config unsigned sendfile:1; //标志位，为1时表示正在将文件中的数据发往连接的另一端 /* 标志位，如果为1，则表示只有在连接套接字对应的发送缓冲区必须满足最低设置的大小阅值时，事件驱动模块才会分发该事件。这与上文 介绍过的ngx_handle_write_event方法中的lowat参数是对应的 */ unsigned sndlowat:1; //ngx_send_lowat /* 标志位，表示如何使用TCP的nodelay特性。它的取值范围是下面这个枚举类型ngx_connection_tcp_nodelay_e。 typedef enum{ NGX_TCP_NODELAY_UNSET=O, NGX_TCP_NODELAY_SET, NGX_TCP_NODELAY_DISABLED ) ngx_connection_tcp_nodelay_e; */ unsigned tcp_nodelay:2; /* ngx_connection_tcp_nodelay_e */ //域套接字默认是disable的, /* 标志位，表示如何使用TCP的nopush特性。它的取值范围是下面这个枚举类型ngx_connection_tcp_nopush_e： typedef enum{ NGX_TCP_NOPUSH_UNSET=0, NGX_TCP_NOPUSH_SET, NGX_TCP_NOPUSH_DISABLED ) ngx_connection_tcp_nopush_e */ //域套接字默认是disable的, unsigned tcp_nopush:2; /* ngx_connection_tcp_nopush_e */ unsigned need_last_buf:1; #if (NGX_HAVE_IOCP) unsigned accept_context_updated:1; #endif /* #if (NGX HAVE AIO- SENDFILE) ／／标志位，为1时表示使用异步I/O的方式将磁盘上文件发送给网络连接的另一端 unsigned aio一sendfile:l; ／／使用异步I/O方式发送的文件，busy_sendfile缓冲区保存待发送文件的信息 ngx_buf_t *busy_sendf ile; #endif */ #if (NGX_HAVE_AIO_SENDFILE) unsigned busy_count:2; #endif #if (NGX_THREADS) ngx_thread_task_t *sendfile_task; #endif };
2、ngx_connection_s对应一个ngx_event_s read和一个ngx_event_s write,其中事件的fd是从ngx_connection_s->fd获取，他们在ngx_worker_process_init->ngx_event_process_init中关联起来 ；其保存了两个事件结构：read 和 write，分别对应该连接对应的读事件和写事件，而事件模块 ngx_event_t 中提供了 ngx_event_handler_pt 回调函数，在事件触发时调用相应的方法；
/*cycle->read_events和cycle->write_events这两个数组存放的是ngx_event_s,他们是对应的， 见ngx_event_process_init ngx_event_t事件和ngx_connection_t连接是一一对应的 ngx_event_t事件和ngx_connection_t连接是处理TCP连接的基础数据结构, 在Nginx中，每一个事件都由ngx_event_t结构体来表示 /* 1.ngx_event_s可以是普通的epoll读写事件(参考ngx_event_connect_peer->ngx_add_conn或者ngx_add_event)，通过读写事件触发 2.也可以是普通定时器事件(参考ngx_cache_manager_process_handler->ngx_add_timer(ngx_event_add_timer))，通过ngx_process_events_and_timers中的 epoll_wait返回，可以是读写事件触发返回，也可能是因为没获取到共享锁，从而等待200ms返回重新获取锁来跟新事件并执行超时事件来跟新事件并且判断定 时器链表中的超时事件，超时则执行从而指向event的handler，然后进一步指向对应r或者u的->write_event_handler read_event_handler 3.也可以是利用定时器expirt实现的读写事件(参考ngx_http_set_write_handler->ngx_add_timer(ngx_event_add_timer)),触发过程见2，只是在handler中不会执行write_event_handler read_event_handler */ struct ngx_event_s { /* 事件相关的对象。通常data都是指向ngx_connection_t连接对象,见ngx_get_connection。开启文件异步I/O时，它可能会指向ngx_event_aio_t(ngx_file_aio_init)结构体 */ void *data; //赋值见ngx_get_connection //标志位，为1时表示事件是可写的。通常情况下，它表示对应的TCP连接目前状态是可写的，也就是连接处于可以发送网络包的状态 unsigned write:1; //见ngx_get_connection，可写事件ev默认为1 读ev事件应该默认还是0 //标志位，为1时表示为此事件可以建立新的连接。通常情况下，在ngx_cycle_t中的listening动态数组中，每一个监听对象ngx_listening_t对 //应的读事件中的accept标志位才会是l ngx_event_process_init中置1 unsigned accept:1; /* 这个标志位用于区分当前事件是否是过期的，它仅仅是给事件驱动模块使用的，而事件消费模块可不用关心。为什么需要这个标志位呢？ 当开始处理一批事件时，处理前面的事件可能会关闭一些连接，而这些连接有可能影响这批事件中还未处理到的后面的事件。这时， 可通过instance标志位来避免处理后面的已经过期的事件。将详细描述ngx_epoll_module是如何使用instance标志位区分 过期事件的，这是一个巧妙的设计方法 instance标志位为什么可以判断事件是否过期？instance标志位的使用其实很简单，它利用了指针的最后一位一定 是0这一特性。既然最后一位始终都是0，那么不如用来表示instance。这样，在使用ngx_epoll_add_event方法向epoll中添加事件时，就把epoll_event中 联合成员data的ptr成员指向ngx_connection_t连接的地址，同时把最后一位置为这个事件的instance标志。而在ngx_epoll_process_events方法中取出指向连接的 ptr地址时，先把最后一位instance取出来，再把ptr还原成正常的地址赋给ngx_connection_t连接。这样，instance究竟放在何处的问题也就解决了。 那么，过期事件又是怎么回事呢？举个例子，假设epoll_wait -次返回3个事件，在第 1个事件的处理过程中，由于业务的需要，所以关闭了一个连接，而这个连接恰好对应第3个事件。这样的话，在处理到第3个事件时，这个事件就 已经是过期辜件了，一旦处理必然出错。既然如此，把关闭的这个连接的fd套接字置为一1能解决问题吗？答案是不能处理所有情况。 下面先来看看这种貌似不可能发生的场景到底是怎么发生的：假设第3个事件对应的ngx_connection_t连接中的fd套接字原先是50，处理第1个事件 时把这个连接的套接字关闭了，同时置为一1，并且调用ngx_free_connection将该连接归还给连接池。在ngx_epoll_process_events方法的循环中开始处 理第2个事件，恰好第2个事件是建立新连接事件，调用ngx_get_connection从连接池中取出的连接非常可能就是刚刚释放的第3个事件对应的连接。由于套 接字50刚刚被释放，Linux内核非常有可能把刚刚释放的套接字50又分配给新建立的连接。因此，在循环中处理第3个事件时，这个事件就是过期的了！它对应 的事件是关闭的连接，而不是新建立的连接。 如何解决这个问题？依靠instance标志位。当调用ngx_get_connection从连接池中获取一个新连接时，instance标志位就会置反 */ /* used to detect the stale events in kqueue and epoll */ unsigned instance:1; //ngx_get_connection从连接池中获取一个新连接时，instance标志位就会置反 //见ngx_get_connection /* * the event was passed or would be passed to a kernel; * in aio mode - operation was posted. */ /* 标志位，为1时表示当前事件是活跃的，为0时表示事件是不活跃的。这个状态对应着事件驱动模块处理方式的不同。例如，在添加事件、 删除事件和处理事件时，active标志位的不同都会对应着不同的处理方式。在使用事件时，一般不会直接改变active标志位 */ //ngx_epoll_add_event中也会置1 在调用该函数后，该值一直为1，除非调用ngx_epoll_del_event unsigned active:1; //标记是否已经添加到事件驱动中，避免重复添加 在server端accept成功后， //或者在client端connect的时候把active置1，见ngx_epoll_add_connection。第一次添加epoll_ctl为EPOLL_CTL_ADD,如果再次添加发 //现active为1,则epoll_ctl为EPOLL_CTL_MOD /* 标志位，为1时表示禁用事件，仅在kqueue或者rtsig事件驱动模块中有效，而对于epoll事件驱动模块则无意义，这里不再详述 */ unsigned disabled:1; /* the ready event; in aio mode 0 means that no operation can be posted */ /* 标志位，为1时表示当前事件已经淮备就绪，也就是说，允许这个事件的消费模块处理这个事件。在 HTTP框架中，经常会检查事件的ready标志位以确定是否可以接收请求或者发送响应 ready标志位，如果为1，则表示在与客户端的TCP连接上可以发送数据；如果为0，则表示暂不可发送数据。 */ //如果来自对端的数据内核缓冲区没有数据(返回NGX_EAGAIN)，或者连接断开置0，见ngx_unix_recv //在发送数据的时候，ngx_unix_send中的时候，如果希望发送1000字节，但是实际上send只返回了500字节(说明内核协议栈缓冲区满，需要通过epoll再次促发write的时候才能写)，或者链接异常，则把ready置0 unsigned ready:1; //在ngx_epoll_process_events中置1,读事件触发并读取数据后ngx_unix_recv中置0 /* 该标志位仅对kqueue，eventport等模块有意义，而对于Linux上的epoll事件驱动模块则是无意叉的，限于篇幅，不再详细说明 */ unsigned oneshot:1; /* aio operation is complete */ //aio on | thread_pool方式下，如果读取数据完成，则在ngx_epoll_eventfd_handler(aio on)或者ngx_thread_pool_handler(aio thread_pool)中置1 unsigned complete:1; //表示读取数据完成，通过epoll机制返回获取 ，见ngx_epoll_eventfd_handler //标志位，为1时表示当前处理的字符流已经结束 例如内核缓冲区没有数据，你去读，则会返回0 unsigned eof:1; //见ngx_unix_recv //标志位，为1时表示事件在处理过程中出现错误 unsigned error:1; //标志位，为I时表示这个事件已经超时，用以提示事件的消费模块做超时处理 /*读客户端连接的数据，在ngx_http_init_connection(ngx_connection_t *c)中的ngx_add_timer(rev, c->listening->post_accept_timeout)把读事件添加到定时器中，如果超时则置1 每次ngx_unix_recv把内核数据读取完毕后，在重新启动add epoll，等待新的数据到来，同时会启动定时器ngx_add_timer(rev, c->listening->post_accept_timeout); 如果在post_accept_timeout这么长事件内没有数据到来则超时，开始处理关闭TCP流程*/ /* 读超时是指的读取对端数据的超时时间，写超时指的是当数据包很大的时候，write返回NGX_AGAIN，则会添加write定时器，从而判断是否超时，如果发往 对端数据长度小，则一般write直接返回成功，则不会添加write超时定时器，也就不会有write超时，写定时器参考函数ngx_http_upstream_send_request */ unsigned timedout:1; //定时器超时标记，见ngx_event_expire_timers //标志位，为1时表示这个事件存在于定时器中 unsigned timer_set:1; //ngx_event_add_timer ngx_add_timer 中置1 ngx_event_expire_timers置0 //标志位，delayed为1时表示需要延迟处理这个事件，它仅用于限速功能 unsigned delayed:1; //限速见ngx_http_write_filter /* 标志位，为1时表示延迟建立TCP连接，也就是说，经过TCP三次握手后并不建立连接，而是要等到真正收到数据包后才会建立TCP连接 */ unsigned deferred_accept:1; //通过listen的时候添加 deferred 参数来确定 /* the pending eof reported by kqueue, epoll or in aio chain operation */ //标志位，为1时表示等待字符流结束，它只与kqueue和aio事件驱动机制有关 //一般在触发EPOLLRDHUP(当对端已经关闭，本端写数据，会引起该事件)的时候，会置1，见ngx_epoll_process_events unsigned pending_eof:1; /* if (c->read->posted) { //删除post队列的时候需要检查 ngx_delete_posted_event(c->read); } */ unsigned posted:1; //表示延迟处理该事件，见ngx_epoll_process_events -> ngx_post_event 标记是否在延迟队列里面 //标志位，为1时表示当前事件已经关闭，epoll模块没有使用它 unsigned closed:1; //ngx_close_connection中置1 /* to test on worker exit */ //这两个该标志位目前无实际意义 unsigned channel:1; unsigned resolver:1; unsigned cancelable:1; #if (NGX_WIN32) /* setsockopt(SO_UPDATE_ACCEPT_CONTEXT) was successful */ unsigned accept_context_updated:1; #endif #if (NGX_HAVE_KQUEUE) unsigned kq_vnode:1; /* the pending errno reported by kqueue */ int kq_errno; #endif /* * kqueue only: * accept: number of sockets that wait to be accepted * read: bytes to read when event is ready * or lowat when event is set with NGX_LOWAT_EVENT flag * write: available space in buffer when event is ready * or lowat when event is set with NGX_LOWAT_EVENT flag * * iocp: TODO * * otherwise: * accept: 1 if accept many, 0 otherwise */ //标志住，在epoll事件驱动机制下表示一次尽可能多地建立TCP连接，它与multi_accept配置项对应，实现原理基见9.8.1节 #if (NGX_HAVE_KQUEUE) || (NGX_HAVE_IOCP) int available; #else unsigned available:1; //ngx_event_accept中 ev->available = ecf->multi_accept; #endif /* 每一个事件最核心的部分是handler回调方法，它将由每一个事件消费模块实现，以此决定这个事件究竟如何“消费” */ /* 1.event可以是普通的epoll读写事件(参考ngx_event_connect_peer->ngx_add_conn或者ngx_add_event)，通过读写事件触发 2.也可以是普通定时器事件(参考ngx_cache_manager_process_handler->ngx_add_timer(ngx_event_add_timer))，通过ngx_process_events_and_timers中的 epoll_wait返回，可以是读写事件触发返回，也可能是因为没获取到共享锁，从而等待0.5s返回重新获取锁来跟新事件并执行超时事件来跟新事件并且判断定 时器链表中的超时事件，超时则执行从而指向event的handler，然后进一步指向对应r或者u的->write_event_handler read_event_handler 3.也可以是利用定时器expirt实现的读写事件(参考ngx_http_set_write_handler->ngx_add_timer(ngx_event_add_timer)),触发过程见2，只是在handler中不会执行write_event_handler read_event_handler */ //这个事件发生时的处理方法，每个事件消费模块都会重新实现它 //ngx_epoll_process_events中执行accept /* 赋值为ngx_http_process_request_line ngx_event_process_init中初始化为ngx_event_accept 如果是文件异步i/o，赋值为ngx_epoll_eventfd_handler //当accept客户端连接后ngx_http_init_connection中赋值为ngx_http_wait_request_handler来读取客户端数据 在解析完客户端发送来的请求的请求行和头部行后，设置handler为ngx_http_request_handler */ //一般与客户端的数据读写是 ngx_http_request_handler; 与后端服务器读写为ngx_http_upstream_handler(如fastcgi proxy memcache gwgi等) /* ngx_event_accept，ngx_http_ssl_handshake ngx_ssl_handshake_handler ngx_http_v2_write_handler ngx_http_v2_read_handler ngx_http_wait_request_handler ngx_http_request_handler,ngx_http_upstream_handler ngx_file_aio_event_handler */ ngx_event_handler_pt handler; //由epoll读写事件在ngx_epoll_process_events触发 #if (NGX_HAVE_IOCP) ngx_event_ovlp_t ovlp; #endif //由于epoll事件驱动方式不使用index，所以这里不再说明 ngx_uint_t index; //可用于记录error_log日志的ngx_log_t对象 ngx_log_t *log; //可以记录日志的ngx_log_t对象 其实就是ngx_listening_t中获取的log //赋值见ngx_event_accept //定时器节点，用于定时器红黑树中 ngx_rbtree_node_t timer; //见ngx_event_timer_rbtree /* the posted queue */ /* post事件将会构成一个队列再统一处理，这个队列以next和prev作为链表指针，以此构成一个简易的双向链表，其中next指向后一个事件的地址， prev指向前一个事件的地址 */ ngx_queue_t queue; #if 0 /* the threads support */ /* * the event thread context, we store it here * if $(CC) does not understand __thread declaration * and pthread_getspecific() is too costly */ void *thr_ctx; #if (NGX_EVENT_T_PADDING) /* event should not cross cache line in SMP */ uint32_t padding[NGX_EVENT_T_PADDING]; #endif #endif };
在 http 初始化中，ngx_http_block 调用 ngx_http_optimize_servers 调用 ngx_http_init_listening 调用 ngx_http_add_listening 创建并初始化了 nginx 监听结构，进行了一步赋值：
ls->handler = ngx_http_init_connection
指定了 http 建立连接的回调函数为 ngx_http_init_connection
在 事件驱动函数 ngx_process_events_and_timers 中调用了 ngx_event_process_posted 函数调用了事件的 handler 回调函数 ngx_event_accept，在 ngx_event_accept 函数中，首先调用了 accept 函数，随后在连接建立以后，调用了 nginx 连接结构的 ngx_listening_t 的回调函数
/*/设置ngx_listening_t的handler，这个handler会在监听到客户端连接时被调用，具体就是在ngx_event_accept函数中， ngx_http_init_connection函数顾名思义，就是初始化这个新建的连接*/ void ngx_http_init_connection(ngx_connection_t *c) { ngx_uint_t i; ngx_event_t *rev; struct sockaddr_in *sin; ngx_http_port_t *port; ngx_http_in_addr_t *addr; ngx_http_log_ctx_t *ctx; ngx_http_connection_t *hc; #if (NGX_HAVE_INET6) struct sockaddr_in6 *sin6; ngx_http_in6_addr_t *addr6; #endif //注意ngx_connection_t和ngx_http_connection_t的区别，前者是建立连接accept前使用的结构，后者是连接成功后使用的结构 // 创建 http 连接描述结构 hc = ngx_pcalloc(c->pool, sizeof(ngx_http_connection_t)); if (hc == NULL) { ngx_http_close_connection(c); return; } //在服务器端accept客户端连接成功(ngx_event_accept)后，会通过ngx_get_connection从连接池获取一个ngx_connection_t结构，也就是每个客户端连接对于一个ngx_connection_t结构， //并且为其分配一个ngx_http_connection_t结构，ngx_connection_t->data = ngx_http_connection_t，见ngx_http_init_connection c->data = hc; /* find the server configuration for the address:port */ port = c->listening->servers; if (port->naddrs > 1) { /* * there are several addresses on this port and one of them * is an "*:port" wildcard so getsockname() in ngx_http_server_addr() * is required to determine a server address */ //说明listen ip:port存在几条没有bind选项，并且存在通配符配置，如listen *:port,那么就需要通过ngx_connection_local_sockaddr来确定 //究竟客户端是和那个本地ip地址建立的连接 if (ngx_connection_local_sockaddr(c, NULL, 0) != NGX_OK) { // ngx_http_close_connection(c); return; } switch (c->local_sockaddr->sa_family) { #if (NGX_HAVE_INET6) case AF_INET6: sin6 = (struct sockaddr_in6 *) c->local_sockaddr; addr6 = port->addrs; /* the last address is "*" */ for (i = 0; i < port->naddrs - 1; i++) { if (ngx_memcmp(&addr6[i].addr6, &sin6->sin6_addr, 16) == 0) { break; } } hc->addr_conf = &addr6[i].conf; break; #endif default: /* AF_INET */ sin = (struct sockaddr_in *) c->local_sockaddr; addr = port->addrs; /* the last address is "*" */ //根据上面的ngx_connection_local_sockaddr函数获取到客户端连接到本地，本地IP地址获取到后，遍历ngx_http_port_t找到对应 //的IP地址和端口，然后赋值给ngx_http_connection_t->addr_conf，这里面存储有server_name配置信息以及该ip:port对应的上下文信息 for (i = 0; i < port->naddrs - 1; i++) { if (addr[i].addr == sin->sin_addr.s_addr) { break; } } /* 这里也体现了在ngx_http_init_connection中获取http{}上下文ctx，如果客户端请求中带有host参数，则会继续在ngx_http_set_virtual_server 中重新获取对应的server{}和location{}，如果客户端请求不带host头部行，则使用默认的server{},见 ngx_http_init_connection */ hc->addr_conf = &addr[i].conf; break; } } else { switch (c->local_sockaddr->sa_family) { #if (NGX_HAVE_INET6) case AF_INET6: addr6 = port->addrs; hc->addr_conf = &addr6[0].conf; break; #endif default: /* AF_INET */ addr = port->addrs; hc->addr_conf = &addr[0].conf; break; } } /* the default server configuration for the address:port */ //listen add:port对于的 server{}配置块的上下文ctx hc->conf_ctx = hc->addr_conf->default_server->ctx; ctx = ngx_palloc(c->pool, sizeof(ngx_http_log_ctx_t)); if (ctx == NULL) { ngx_http_close_connection(c); return; } ctx->connection = c; ctx->request = NULL; ctx->current_request = NULL; c->log->connection = c->number; c->log->handler = ngx_http_log_error; c->log->data = ctx; c->log->action = "waiting for request"; c->log_error = NGX_ERROR_INFO; rev = c->read; rev->handler = ngx_http_wait_request_handler; c->write->handler = ngx_http_empty_handler; #if (NGX_HTTP_V2) /* 这里放在SSL的前面是，如果没有配置SSL，则直接不用进行SSL协商而进行HTTP2处理ngx_http_v2_init */ if (hc->addr_conf->http2) { rev->handler = ngx_http_v2_init; } #endif #if (NGX_HTTP_SSL) { ngx_http_ssl_srv_conf_t *sscf; sscf = ngx_http_get_module_srv_conf(hc->conf_ctx, ngx_http_ssl_module); if (sscf->enable || hc->addr_conf->ssl) { c->log->action = "SSL handshaking"; if (hc->addr_conf->ssl && sscf->ssl.ctx == NULL) { ngx_log_error(NGX_LOG_ERR, c->log, 0, "no "ssl_certificate" is defined " "in server listening on SSL port"); ngx_http_close_connection(c); return; } hc->ssl = 1; rev->handler = ngx_http_ssl_handshake; } } #endif if (hc->addr_conf->proxy_protocol) { hc->proxy_protocol = 1; c->log->action = "reading PROXY protocol"; } /* 如果新连接的读事件ngx_event_t结构体中的标志位ready为1，实际上表示这个连接对应的套接字缓存上已经有用户发来的数据， 这时就可调用上面说过的ngx_http_init_request方法处理请求。 */ //这里只可能是当listen的时候添加了defered参数并且内核支持，在ngx_event_accept的时候才会置1，才可能执行下面的if里面的内容，否则不会只需if里面的内容 if (rev->ready) { /* the deferred accept(), iocp */ if (ngx_use_accept_mutex) { //如果是配置了accept_mutex，则把该rev->handler延后处理， //实际上执行的地方为ngx_process_events_and_timers中的ngx_event_process_posted ngx_post_event(rev, &ngx_posted_events); return; } rev->handler(rev); //ngx_http_wait_request_handler return; } /* 在有些情况下，当TCP连接建立成功时同时也出现了可读事件（例如，在套接字listen配置时设置了deferred选项时，内核仅在套接字上确实收到请求时才会通知epoll 调度事件的回调方法。当然，在大部分情况下，ngx_http_init_request方法和 ngx_http_init_connection方法都是由两个事件（TCP连接建立成功事件和连接上的可读事件）触发调用的 */ /* 调用ngx_add_timer方法把读事件添加到定时器中，设置的超时时间则是nginx.conf中client_header_timeout配置项指定的参数。 也就是说，如果经过client_header_timeout时间后这个连接上还没有用户数据到达，则会由定时器触发调用读事件的ngx_http_init_request处理方法。 */ ngx_add_timer(rev, c->listening->post_accept_timeout, NGX_FUNC_LINE); //把接收事件添加到定时器中，当post_accept_timeout秒还没有客户端数据到来，就关闭连接 ngx_reusable_connection(c, 1); if (ngx_handle_read_event(rev, 0, NGX_FUNC_LINE) != NGX_OK) { //当下次有数据从客户端发送过来的时候，会在ngx_epoll_process_events把对应的ready置1。 ngx_http_close_connection(c); return; } }
accept 中调用关系
/* 如何建立新连接 下面对流程中的7个步骤进行说明。 1)首先调用accept方法试图建立新连接，如果没有准备好的新连接事件，ngx_event_accept方法会直接返回。 2)设置负载均衡阈值ngx_accept_disabled，这个阈值是进程允许的总连接数的1/8减去空闲连接数， 3)调用ngx_get_connection方法由连接池中获取一个ngx_connection_t连接对象。 4)为ngx_connection_t中的pool指针建立内存池。在这个连接释放到空闲连接池时，释放pool内存池。 5)设置套接字的属性，如设为非阻塞套接字。 6)将这个新连接对应的读事件添加到epoll等事件驱动模块中，这样，在这个连接上如果接收到用户请求epoll_wait，就会收集到这个事件。 7)调用监听对象ngx_listening_t中的handler回调方法。ngx_listening_t结构俸的handler回调方法就是当新的TCP连接刚刚建立完成时在这里调用的。 最后，如果监听事件的available标志位为1，再次循环到第1步，否则ngx_event_accept方法结束。事件的available标志位对应着multi_accept配置 项。当available为l时，告诉Nginx -次性尽量多地建立新连接，它的实现原理也就在这里 */ //这里的event是在ngx_event_process_init中从连接池中获取的 ngx_connection_t中的->read读事件 //accept是在ngx_event_process_init(但进程或者不配置负载均衡的时候)或者(多进程，配置负载均衡)的时候把accept事件添加到epoll中 void //该形参中的ngx_connection_t(ngx_event_t)是为accept事件连接准备的空间，当accept返回成功后，会重新获取一个ngx_connection_t(ngx_event_t)用来读写该连接 ngx_event_accept(ngx_event_t *ev) //在ngx_process_events_and_timers中执行 { //一个accept事件对应一个ev，如当前一次有4个客户端accept，应该对应4个ev事件，一次来多个accept的处理在下面的do {}while中实现 socklen_t socklen; ngx_err_t err; ngx_log_t *log;bangmang ngx_uint_t level; ngx_socket_t s; //如果是文件异步i/o中的ngx_event_aio_t，则它来自ngx_event_aio_t->ngx_event_t(只有读),如果是网络事件中的event,则为ngx_connection_s中的event(包括读和写) ngx_event_t *rev, *wev; ngx_listening_t *ls; ngx_connection_t *c, *lc; ngx_event_conf_t *ecf; u_char sa[NGX_SOCKADDRLEN]; #if (NGX_HAVE_ACCEPT4) static ngx_uint_t use_accept4 = 1; #endif if (ev->timedout) { if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) { return; } ev->timedout = 0; } ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module); if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) { ev->available = ecf->multi_accept; } lc = ev->data; ls = lc->listening; ev->ready = 0; ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0, "accept on %V, ready: %d", &ls->addr_text, ev->available); do { /* 如果是一次读取一个accept事件的话，循环体只执行一次， 如果是一次性可以读取所有的accept事件，则这个循环体执行次数为accept事件数*/ socklen = NGX_SOCKADDRLEN; #if (NGX_HAVE_ACCEPT4) //ngx_close_socket可以关闭套接字 if (use_accept4) { s = accept4(lc->fd, (struct sockaddr *) sa, &socklen, SOCK_NONBLOCK); } else { s = accept(lc->fd, (struct sockaddr *) sa, &socklen); } #else /* 针对非阻塞I/O执行的系统调用则总是立即返回，而不管事件足否已经发生。如果事件没有眭即发生，这些系统调用就 返回—1．和出错的情况一样。此时我们必须根据errno来区分这两种情况。对accept、send和recv而言，事件未发牛时errno 通常被设置成EAGAIN（意为“再来一次”）或者EWOULDBLOCK（意为“期待阻塞”）：对conncct而言，errno则被 设置成EINPROGRESS（意为“在处理中"）。 */ s = accept(lc->fd, (struct sockaddr *) sa, &socklen); #endif if (s == (ngx_socket_t) -1) { err = ngx_socket_errno; /* 如果要去一次性读取所有的accept信息，当读取完毕后，通过这里返回。所有的accept事件都读取完毕 */ if (err == NGX_EAGAIN) { //如果event{}开启multi_accept，则在accept完该listen ip:port对应的ip和端口连接后，会通过这里返回 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, err, "accept() not ready"); return; } level = NGX_LOG_ALERT; if (err == NGX_ECONNABORTED) { level = NGX_LOG_ERR; } else if (err == NGX_EMFILE || err == NGX_ENFILE) { level = NGX_LOG_CRIT; } #if (NGX_HAVE_ACCEPT4) ngx_log_error(level, ev->log, err, use_accept4 ? "accept4() failed" : "accept() failed"); if (use_accept4 && err == NGX_ENOSYS) { use_accept4 = 0; ngx_inherited_nonblocking = 0; continue; } #else ngx_log_error(level, ev->log, err, "accept() failed"); #endif if (err == NGX_ECONNABORTED) { if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) { ev->available--; } if (ev->available) { continue; } } if (err == NGX_EMFILE || err == NGX_ENFILE) {// Too many descriptors are in use by this process. if (ngx_disable_accept_events((ngx_cycle_t *) ngx_cycle, 1) != NGX_OK) { return; } if (ngx_use_accept_mutex) { if (ngx_accept_mutex_held) { ngx_shmtx_unlock(&ngx_accept_mutex); ngx_accept_mutex_held = 0; } //当前进程连接accpet失败，则可以暂时设置为1，下次来的时候由其他进程竞争accpet锁，下下次该进程继续竞争该accept，因为在下次的时候ngx_process_events_and_timers //ngx_accept_disabled = 1; 减去1后为0，可以继续竞争 ngx_accept_disabled = 1; } else { ////如果是不需要实现负载均衡，则扫尾延时下继续在ngx_process_events_and_timers中accept ngx_add_timer(ev, ecf->accept_mutex_delay, NGX_FUNC_LINE); } } return; } #if (NGX_STAT_STUB) (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1); #endif //设置负载均衡阀值 最开始free_connection_n=connection_n，见ngx_event_process_init ngx_accept_disabled = ngx_cycle->connection_n / 8 - ngx_cycle->free_connection_n; //判断可用连接的数目和总数目的八分之一大小，如果可用的小于八分之一，为正 //在服务器端accept客户端连接成功(ngx_event_accept)后，会通过ngx_get_connection从连接池获取一个ngx_connection_t结构，也就是每个客户端连接对于一个ngx_connection_t结构， //并且为其分配一个ngx_http_connection_t结构，ngx_connection_t->data = ngx_http_connection_t，见ngx_http_init_connection //从连接池中获取一个空闲ngx_connection_t，用于客户端连接建立成功后向该连接读写数据，函数形参中的ngx_event_t对应的是为accept事件对应的 //ngx_connection_t中对应的event c = ngx_get_connection(s, ev->log); //ngx_get_connection中c->fd = s; //注意，这里的ngx_connection_t是从连接池中从新获取的，和ngx_epoll_process_events中的ngx_connection_t是两个不同的。 if (c == NULL) { if (ngx_close_socket(s) == -1) { ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_socket_errno, ngx_close_socket_n " failed"); } return; } #if (NGX_STAT_STUB) (void) ngx_atomic_fetch_add(ngx_stat_active, 1); #endif c->pool = ngx_create_pool(ls->pool_size, ev->log); if (c->pool == NULL) { ngx_close_accepted_connection(c); return; } c->sockaddr = ngx_palloc(c->pool, socklen); if (c->sockaddr == NULL) { ngx_close_accepted_connection(c); return; } ngx_memcpy(c->sockaddr, sa, socklen); log = ngx_palloc(c->pool, sizeof(ngx_log_t)); if (log == NULL) { ngx_close_accepted_connection(c); return; } /* set a blocking mode for iocp and non-blocking mode for others */ if (ngx_inherited_nonblocking) { if (ngx_event_flags & NGX_USE_IOCP_EVENT) { if (ngx_blocking(s) == -1) { ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_socket_errno, ngx_blocking_n " failed"); ngx_close_accepted_connection(c); return; } } } else { if (!(ngx_event_flags & NGX_USE_IOCP_EVENT)) { if (ngx_nonblocking(s) == -1) { ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_socket_errno, ngx_nonblocking_n " failed"); ngx_close_accepted_connection(c); return; } } } *log = ls->log; c->recv = ngx_recv; c->send = ngx_send; c->recv_chain = ngx_recv_chain; c->send_chain = ngx_send_chain; c->log = log; c->pool->log = log; c->socklen = socklen; c->listening = ls; c->local_sockaddr = ls->sockaddr; c->local_socklen = ls->socklen; c->unexpected_eof = 1; #if (NGX_HAVE_UNIX_DOMAIN) if (c->sockaddr->sa_family == AF_UNIX) { c->tcp_nopush = NGX_TCP_NOPUSH_DISABLED; c->tcp_nodelay = NGX_TCP_NODELAY_DISABLED; #if (NGX_SOLARIS) /* Solaris's sendfilev() supports AF_NCA, AF_INET, and AF_INET6 */ c->sendfile = 0; #endif } #endif //注意，这里的ngx_connection_t是从连接池中从新获取的，和ngx_epoll_process_events中的ngx_connection_t是两个不同的。 rev = c->read; wev = c->write; wev->ready = 1; if (ngx_event_flags & NGX_USE_IOCP_EVENT) { rev->ready = 1; } if (ev->deferred_accept) { rev->ready = 1; #if (NGX_HAVE_KQUEUE) rev->available = 1; #endif } rev->log = log; wev->log = log; /* * TODO: MT: - ngx_atomic_fetch_add() * or protection by critical section or light mutex * * TODO: MP: - allocated in a shared memory * - ngx_atomic_fetch_add() * or protection by critical section or light mutex */ c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1); #if (NGX_STAT_STUB) (void) ngx_atomic_fetch_add(ngx_stat_handled, 1); #endif if (ls->addr_ntop) { c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len); if (c->addr_text.data == NULL) { ngx_close_accepted_connection(c); return; } c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen, c->addr_text.data, ls->addr_text_max_len, 0); if (c->addr_text.len == 0) { ngx_close_accepted_connection(c); return; } } #if (NGX_DEBUG) { ngx_str_t addr; struct sockaddr_in *sin; ngx_cidr_t *cidr; ngx_uint_t i; u_char text[NGX_SOCKADDR_STRLEN]; #if (NGX_HAVE_INET6) struct sockaddr_in6 *sin6; ngx_uint_t n; #endif cidr = ecf->debug_connection.elts; for (i = 0; i < ecf->debug_connection.nelts; i++) { if (cidr[i].family != (ngx_uint_t) c->sockaddr->sa_family) { goto next; } switch (cidr[i].family) { #if (NGX_HAVE_INET6) case AF_INET6: sin6 = (struct sockaddr_in6 *) c->sockaddr; for (n = 0; n < 16; n++) { if ((sin6->sin6_addr.s6_addr[n] & cidr[i].u.in6.mask.s6_addr[n]) != cidr[i].u.in6.addr.s6_addr[n]) { goto next; } } break; #endif #if (NGX_HAVE_UNIX_DOMAIN) case AF_UNIX: break; #endif default: /* AF_INET */ sin = (struct sockaddr_in *) c->sockaddr; if ((sin->sin_addr.s_addr & cidr[i].u.in.mask) != cidr[i].u.in.addr) { goto next; } break; } log->log_level = NGX_LOG_DEBUG_CONNECTION|NGX_LOG_DEBUG_ALL; break; next: continue; } if (log->log_level & NGX_LOG_DEBUG_EVENT) { addr.data = text; addr.len = ngx_sock_ntop(c->sockaddr, c->socklen, text, NGX_SOCKADDR_STRLEN, 1); ngx_log_debug3(NGX_LOG_DEBUG_EVENT, log, 0, "*%uA accept: %V fd:%d", c->number, &addr, s); } } #endif 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(c);//ngx_http_init_connection if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) { ev->available--; } } while (ev->available); //一次性读取所有当前的accept，直到accept返回NGX_EAGAIN，然后退出 }