Nginx accept鎖的機制和實現

本文基於Nginx 0.8.55源代碼，並基於epoll機制分析

1. accept鎖的實現

1.1 accpet鎖是個什麼東西

提到accept鎖，就不得不提起驚羣問題。

所謂驚羣問題，就是指的像Nginx這種多進程的服務器，在fork後同時監聽同一個端口時，若是有一個外部鏈接進來，會致使全部休眠的子進程被喚醒，而最終只有一個子進程可以成功處理accept事件，其餘進程都會從新進入休眠中。這就致使出現了不少沒必要要的schedule和上下文切換，而這些開銷是徹底沒必要要的。

而在Linux內核的較新版本中，accept調用自己所引發的驚羣問題已經獲得瞭解決，可是在Nginx中，accept是交給epoll機制來處理的，epoll的accept帶來的驚羣問題並無獲得解決（應該是epoll_wait自己並無區別讀事件是否來自於一個Listen套接字的能力，因此全部監聽這個事件的進程會被這個epoll_wait喚醒。），因此Nginx的accept驚羣問題仍然須要定製一個本身的解決方案。

accept鎖就是nginx的解決方案，本質上這是一個跨進程的互斥鎖，以這個互斥鎖來保證只有一個進程具有監聽accept事件的能力。

實現上accept鎖是一個跨進程鎖，其在Nginx中是一個全局變量，聲明以下：

ngx_shmtx_t           ngx_accept_mutex;
複製代碼

這是一個在event模塊初始化時就分配好的鎖，放在一塊進程間共享的內存中，以保證全部進程都能訪問這一個實例，其加鎖解鎖是藉由linux的原子變量來作CAS，若是加鎖失敗則當即返回，是一種非阻塞的鎖。加解鎖代碼以下：

static ngx_inline ngx_uint_t ngx_shmtx_trylock(ngx_shmtx_t *mtx) {                                                                              
    return (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid));     
}                                                                              
                                                                               
#define ngx_shmtx_lock(mtx) ngx_spinlock((mtx)->lock, ngx_pid, 1024) 
                                                                               
#define ngx_shmtx_unlock(mtx) (void) ngx_atomic_cmp_set((mtx)->lock, ngx_pid, 0)
複製代碼

能夠看出，調用ngx_shmtx_trylock失敗後會馬上返回而不會阻塞。

1.2 accept鎖如何保證只有一個進程可以處理新鏈接

要解決epoll帶來的accept鎖的問題也很簡單，只須要保證同一時間只有一個進程註冊了accept的epoll事件便可。

Nginx採用的處理模式也沒什麼特別的，大概就是以下的邏輯：

嘗試獲取accept鎖
if 獲取成功：
	在epoll中註冊accept事件
else:
	在epoll中註銷accept事件
處理全部事件
釋放accept鎖
複製代碼

固然這裏忽略了延後事件的處理，這部分咱們放到後面討論。

對於accept鎖的處理和epoll中註冊註銷accept事件的的處理都是在ngx_trylock_accept_mutex中進行的。而這一系列過程則是在nginx主體循環中反覆調用的void ngx_process_events_and_timers(ngx_cycle_t *cycle)中進行。

也就是說，每輪事件的處理都會首先競爭accept鎖，競爭成功則在epoll中註冊accept事件，失敗則註銷accept事件，而後處理完事件以後，釋放accept鎖。由此只有一個進程監聽一個listen套接字，從而避免了驚羣問題。

1.3 事件處理機制爲不長時間佔用accept鎖做了哪些努力

accept鎖處理驚羣問題的方案看起來彷佛很美，但若是徹底使用上述邏輯，就會有一個問題：若是服務器很是忙，有很是多事件要處理，那麼「處理全部事件這一步」就會消耗很是長的時間，也就是說，某一個進程長時間佔用accept鎖，而又無暇處理新鏈接；其餘進程又沒有佔用accept鎖，一樣沒法處理新鏈接——至此，新鏈接就處於無人處理的狀態，這對服務的實時性無疑是很要命的。

爲了解決這個問題，Nginx採用了將事件處理延後的方式。即在ngx_process_events的處理中，僅僅將事件放入兩個隊列中：

ngx_thread_volatile ngx_event_t  *ngx_posted_accept_events;                             
ngx_thread_volatile ngx_event_t  *ngx_posted_events; 
複製代碼

返回後先處理ngx_posted_accept_events後馬上釋放accept鎖，而後再慢慢處理其餘事件。

即ngx_process_events僅對epoll_wait進行處理，事件的消費則放到accept鎖釋放以後，來最大限度地縮短佔有accept的時間，來讓其餘進程也有足夠的時機處理accept事件。

那麼具體是怎麼實現的呢？其實就是在static ngx_int_t ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)的flags參數中傳入一個NGX_POST_EVENTS的標誌位，處理事件時檢查這個標誌位便可。

這裏只是避免了事件的消費對於accept鎖的長期佔用，那麼萬一epoll_wait自己佔用的時間很長呢？這種事情也不是不可能發生。這方面的處理也很簡單，epoll_wait自己是有超時時間的，限制住它的值就能夠了，這個參數保存在ngx_accept_mutex_delay這個全局變量中。

下面放上ngx_process_events_and_timers 的實現代碼，能夠大概一觀相關的處理：

void ngx_process_events_and_timers(ngx_cycle_t *cycle) {                                                                              
    ngx_uint_t  flags;                                                         
    ngx_msec_t  timer, delta;                                                  
              
	
	/* 省略一些處理時間事件的代碼 */                                                                               
    // 這裏是處理負載均衡鎖和accept鎖的時機 
    if (ngx_use_accept_mutex) {                                                
        // 若是負載均衡token的值大於0, 則說明負載已滿，此時再也不處理accept, 同時把這個值減一
        if (ngx_accept_disabled > 0) {                                         
            ngx_accept_disabled--;                                             
                                                                               
        } else {                                                               
            // 嘗試拿到accept鎖 
            if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {                
                return;                                                        
            }                                                                  
                                                                               
            // 拿到鎖以後把flag加上post標誌，讓全部事件的處理都延後 
            // 以避免太長時間佔用accept鎖 
            if (ngx_accept_mutex_held) {                                       
                flags |= NGX_POST_EVENTS;                                      
                                                                               
            } else {                                                           
                if (timer == NGX_TIMER_INFINITE                                
                    || timer > ngx_accept_mutex_delay)                         
                {                                                              
                    timer = ngx_accept_mutex_delay; // 最多等ngx_accept_mutex_delay個毫秒，防止佔用過久accept鎖
                }                                                              
            }                                                                  
        }                                                                      
    }                                                                                
    delta = ngx_current_msec;                                                  
                                                                               
    // 調用事件處理模塊的process_events，處理一個epoll_wait的方法 
    (void) ngx_process_events(cycle, timer, flags);                            
                                                                               
    delta = ngx_current_msec - delta; //計算處理events事件所消耗的時間 
                                                                               
    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,                         
                   "timer delta: %M", delta);                                  
                                                                               
    // 若是有延後處理的accept事件，那麼延後處理這個事件 
    if (ngx_posted_accept_events) {                                            
        ngx_event_process_posted(cycle, &ngx_posted_accept_events);            
    }                                                                          
                                                                               
    // 釋放accept鎖 
    if (ngx_accept_mutex_held) {                                               
        ngx_shmtx_unlock(&ngx_accept_mutex);                                   
    }                                                                          
                                                                               
    // 處理全部的超時事件 
    if (delta) {                                                               
        ngx_event_expire_timers();                                             
    }                                                                          
                                                                               
    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,                         
                   "posted events %p", ngx_posted_events);                     
                                                                               
    if (ngx_posted_events) {                                                   
        if (ngx_threaded) {                                                    
            ngx_wakeup_worker_thread(cycle);                                   
                                                                               
        } else {                                                               
            // 處理全部的延後事件 
            ngx_event_process_posted(cycle, &ngx_posted_events);               
        }                                                                      
    }                                                                          
}                                                                              

複製代碼

再來看看ngx_epoll_process_events的相關處理：

// 讀事件 
        if ((revents & EPOLLIN) && rev->active) {
            if ((flags & NGX_POST_THREAD_EVENTS) && !rev->accept) {
                rev->posted_ready = 1;

            } else {
                rev->ready = 1;
            }                                                                                                                                                               
            if (flags & NGX_POST_EVENTS) {
                queue = (ngx_event_t **) (rev->accept ?
                               &ngx_posted_accept_events : &ngx_posted_events);
                ngx_locked_post_event(rev, queue);
            } else {
                rev->handler(rev);
            }
        }                                                                                                                                                                   
        wev = c->write;

        // 寫事件
        if ((revents & EPOLLOUT) && wev->active) {
            if (flags & NGX_POST_THREAD_EVENTS) {
                wev->posted_ready = 1;
            } else {
                wev->ready = 1;
            }

            if (flags & NGX_POST_EVENTS) {
                ngx_locked_post_event(wev, &ngx_posted_events);
            } else {
                wev->handler(wev);
            }
        }
複製代碼

處理也相對簡單，若是拿到了accept鎖，就會有NGX_POST_EVENTS標誌那麼就會放到相應的隊列中。沒有的話就會直接處理事件。