Nginx 事件驅動模塊連接處理 · 理解 Nginx 源碼

### 概述 ? ? ? 由于 Nginx 工作在 master-worker 多進程模式，若所有 worker 進程在同一時間監聽同一個端口，當該端口有新的連接事件出現時，每個worker 進程都會調用函數ngx_event_accept 試圖與新的連接建立通信，即所有worker 進程都會被喚醒，這就是所謂的“驚群”問題，這樣會導致系統性能下降。幸好在Nginx 采用了ngx_accept_mutex 同步鎖機制，即只有獲得該鎖的worker 進程才能去處理新的連接事件，也就在同一時間只能有一個worker 進程監聽某個端口。雖然這樣做解決了“驚群”問題，但是隨之會出現另一個問題，若每次出現的新連接事件都被同一個worker 進程獲得鎖的權利并處理該連接事件，這樣會導致進程之間不均衡的狀態，即在所有worker 進程中，某些進程處理的連接事件數量很龐大，而某些進程基本上不用處理連接事件，一直處于空閑狀態。因此，這樣會導致worker 進程之間的負載不均衡，會影響Nginx 的整體性能。為了解決負載失衡的問題，Nginx 在已經實現同步鎖的基礎上定義了負載閾值ngx_accept_disabled，當某個worker 進程的負載閾值大于 0 時，表示該進程處于負載超重的狀態，則Nginx 會控制該進程，使其沒機會試圖與新的連接事件進行通信，這樣就會為其他沒有負載超重的進程創造了處理新連接事件的機會，以此達到進程間的負載均衡。 ### 連接事件處理新連接事件由函數 ngx_event_accept 處理。 ~~~ /* 處理新連接事件 */ void ngx_event_accept(ngx_event_t *ev) { socklen_t socklen; ngx_err_t err; ngx_log_t *log; ngx_uint_t level; ngx_socket_t s; 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; } /* 獲取ngx_event_core_module模塊的配置項參數結構 */ ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module); if (ngx_event_flags & NGX_USE_RTSIG_EVENT) { ev->available = 1; } else 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 { socklen = NGX_SOCKADDRLEN; /* accept 建立一個新的連接 */ #if (NGX_HAVE_ACCEPT4) if (use_accept4) { s = accept4(lc->fd, (struct sockaddr *) sa, &socklen, SOCK_NONBLOCK); } else { s = accept(lc->fd, (struct sockaddr *) sa, &socklen); } #else s = accept(lc->fd, (struct sockaddr *) sa, &socklen); #endif /* 連接建立錯誤時的相應處理 */ if (s == (ngx_socket_t) -1) { err = ngx_socket_errno; if (err == NGX_EAGAIN) { 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) { if (ngx_disable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) { return; } if (ngx_use_accept_mutex) { if (ngx_accept_mutex_held) { ngx_shmtx_unlock(&ngx_accept_mutex); ngx_accept_mutex_held = 0; } ngx_accept_disabled = 1; } else { ngx_add_timer(ev, ecf->accept_mutex_delay); } } return; } #if (NGX_STAT_STUB) (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1); #endif /* * ngx_accept_disabled 變量是負載均衡閾值，表示進程是否超載； * 設置負載均衡閾值為每個進程最大連接數的八分之一減去空閑連接數； * 即當每個進程accept到的活動連接數超過最大連接數的7/8時， * ngx_accept_disabled 大于0，表示該進程處于負載過重； */ ngx_accept_disabled = ngx_cycle->connection_n / 8 - ngx_cycle->free_connection_n; /* 從connections數組中獲取一個connection連接來維護新的連接 */ c = ngx_get_connection(s, ev->log); 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 /* 為新的連接創建一個連接池pool，直到關閉該連接時才釋放該連接池pool */ 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 aio and non-blocking mode for others */ /* 設置套接字的屬性 */ if (ngx_inherited_nonblocking) { if (ngx_event_flags & NGX_USE_AIO_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 { /* 使用epoll模型時，套接字的屬性為非阻塞模式 */ if (!(ngx_event_flags & (NGX_USE_AIO_EVENT|NGX_USE_RTSIG_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 /* 獲取新連接的讀事件、寫事件 */ rev = c->read; wev = c->write; /* 寫事件準備就緒 */ wev->ready = 1; if (ngx_event_flags & (NGX_USE_AIO_EVENT|NGX_USE_RTSIG_EVENT)) { /* rtsig, aio, iocp */ 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 (NGX_THREADS) rev->lock = &c->lock; wev->lock = &c->lock; rev->own_lock = &c->lock; wev->own_lock = &c->lock; #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) { struct sockaddr_in *sin; ngx_cidr_t *cidr; ngx_uint_t i; #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; } } #endif ngx_log_debug3(NGX_LOG_DEBUG_EVENT, log, 0, "*%uA accept: %V fd:%d", c->number, &c->addr_text, s); /* 將新連接對應的讀事件注冊到事件監控機制中； * 注意：若是epoll事件機制，這里是不會執行， * 因為epoll事件機制會在調用新連接處理函數ls->handler(c) *（實際調用ngx_http_init_connection）時，才會把新連接對應的讀事件注冊到epoll事件機制中； */ 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; /* * 設置回調函數，完成新連接的最后初始化工作， * 由函數ngx_http_init_connection完成 */ ls->handler(c); /* 調整事件available標志位，該標志位為1表示Nginx一次盡可能多建立新連接 */ if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) { ev->available--; } } while (ev->available); } /* 將監聽socket連接的讀事件加入到監聽事件中 */ static ngx_int_t ngx_enable_accept_events(ngx_cycle_t *cycle) { ngx_uint_t i; ngx_listening_t *ls; ngx_connection_t *c; /* 獲取監聽數組的首地址 */ ls = cycle->listening.elts; /* 遍歷整個監聽數組 */ for (i = 0; i < cycle->listening.nelts; i++) { /* 獲取當前監聽socket所對應的連接 */ c = ls[i].connection; /* 當前連接的讀事件是否處于active活躍狀態 */ if (c->read->active) { /* 若是處于active狀態，表示該連接的讀事件已經在事件監控對象中 */ continue; } /* 若當前連接沒有加入到事件監控對象中，則將該鏈接注冊到事件監控中 */ if (ngx_event_flags & NGX_USE_RTSIG_EVENT) { if (ngx_add_conn(c) == NGX_ERROR) { return NGX_ERROR; } } else { /* 若當前連接的讀事件不在事件監控對象中，則將其加入 */ if (ngx_add_event(c->read, NGX_READ_EVENT, 0) == NGX_ERROR) { return NGX_ERROR; } } } return NGX_OK; } /* 將監聽連接的讀事件從事件驅動模塊中刪除 */ static ngx_int_t ngx_disable_accept_events(ngx_cycle_t *cycle) { ngx_uint_t i; ngx_listening_t *ls; ngx_connection_t *c; /* 獲取監聽接口 */ ls = cycle->listening.elts; for (i = 0; i < cycle->listening.nelts; i++) { /* 獲取監聽接口對應的連接 */ c = ls[i].connection; if (!c->read->active) { continue; } /* 從事件驅動模塊中移除連接 */ if (ngx_event_flags & NGX_USE_RTSIG_EVENT) { if (ngx_del_conn(c, NGX_DISABLE_EVENT) == NGX_ERROR) { return NGX_ERROR; } } else { /* 從事件驅動模塊移除連接的讀事件 */ if (ngx_del_event(c->read, NGX_READ_EVENT, NGX_DISABLE_EVENT) == NGX_ERROR) { return NGX_ERROR; } } } return NGX_OK; } ~~~ 當出現新連接事件時，只有獲得同步鎖的進程才可以處理該連接事件，避免了“驚群”問題，進程試圖處理新連接事件由函數 ngx_trylock_accept_mutex 實現。 ~~~ /* 試圖處理監聽端口的新連接事件 */ ngx_int_t ngx_trylock_accept_mutex(ngx_cycle_t *cycle) { /* 獲取ngx_accept_mutex鎖，成功返回1，失敗返回0 */ if (ngx_shmtx_trylock(&ngx_accept_mutex)) { ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "accept mutex locked"); /* * 標志位ngx_accept_mutex_held為1表示當前進程已經獲取了ngx_accept_mutex鎖； * 滿足下面條件時，表示當前進程在之前已經獲得ngx_accept_mutex鎖； * 則直接返回； */ if (ngx_accept_mutex_held && ngx_accept_events == 0 && !(ngx_event_flags & NGX_USE_RTSIG_EVENT)) { return NGX_OK; } /* 將所有監聽連接的讀事件添加到當前的epoll事件驅動模塊中 */ if (ngx_enable_accept_events(cycle) == NGX_ERROR) { /* 若添加失敗，則釋放該鎖 */ ngx_shmtx_unlock(&ngx_accept_mutex); return NGX_ERROR; } /* 設置當前進程獲取鎖的情況 */ ngx_accept_events = 0; ngx_accept_mutex_held = 1;/* 表示當前進程已經得到ngx_accept_mutex鎖 */ return NGX_OK; } ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "accept mutex lock failed: %ui", ngx_accept_mutex_held); /* * 若當前進程獲取ngx_accept_mutex鎖失敗，并且ngx_accept_mutex_held為1， * 此時是錯誤情況 */ if (ngx_accept_mutex_held) { /* 將所有監聽連接的讀事件從事件驅動模塊中移除 */ if (ngx_disable_accept_events(cycle) == NGX_ERROR) { return NGX_ERROR; } ngx_accept_mutex_held = 0; } return NGX_OK; } ~~~ Nginx 通過負載閾值 ngx_accept_disabled 控制進程是否處理新連接事件，避免進程間負載均衡問題。 ~~~ if(ngx_accept_disabled > 0){ ngx_accept_disabled --; }else{ if(ngx_trylock_accept_mutex(cycle) == NGX_ERROR){ return; } ... } ~~~ 參考資料：《深入理解Nginx》《深入剖析Nginx》《[Nginx源碼分析-事件循環](http://www.alidata.org/archives/1267)》《關[于ngx_trylock_accept_mutex的一些解釋](http://blog.csdn.net/brainkick/article/details/9081017)》