Nginx之進程間的通信機制（共享內存、原子操作）

1. 概述

Linux 提供了多種進程間傳遞消息的方式，如共享內存、套接字、管道、消息隊列、信號等，而 Nginx 框架使用了 3 種傳遞消息的傳遞方式：共享內存、套接字、信號。

在進程間訪問共享資源時，還需要提供一種機制使各個進程有序、安全地訪問資源，避免並發訪問帶來的未知結果。Nginx 主要使用了 3 種同步方式：原子操作、信號量、文件鎖。

由於 Nginx 的每個 worker 進程都會同時處理千萬個請求，所以處理任何一個請求時都不應該阻塞當前進程處理后續的其他請求。如，不要隨意地使用信號量互斥鎖，這會使得 worker 進程在得不到鎖時進入睡眠狀態，從而導致這個 worker 進程上的其他請求被 "餓死"。

2. 共享內存

共享內存是 Linux 下提供的最基本的進程間通信方法，它通過 mmap 或者 shmgat 系統調用在內存中創建了一塊連續的線性地址空間，而通過 munmap 或者 shmdt 系統調用可以釋放這塊內存。使用共享內存的好處是當多個進程使用同一塊共享內存時，在任何一個進程修改了共享內存中的內容后，其他進程通過訪問這段共享內存都能夠得到修改后的內容。

雖然 mmap 可以以磁盤文件的方式映射共享內存，但在 Nginx 封裝的共享內存操作方法中是沒有使用到映射文件功能的。

Nginx 定義了 ngx_shm_t 結構體，用於描述一塊共享內存：

typedef struct {
    /* 執行共享內存的起始地址 */
    u_char      *addr;
    /* 共享內存的長度 */
    size_t       size;
    /* 這塊共享內存的名稱 */
    ngx_str_t    name;
    /* 記錄日志的 ngx_log_t 對象 */
    ngx_log_t   *log;
    /* 表示共享內存是否已經分配過的標志位，為 1 時表示已經存在 */
    ngx_uint_t   exists; /* unsigned exists:1 */
}ngx_shm_t;

操作 ngx_shm_t 結構體的方法有以下兩個：

• ngx_shm_alloc：用於分配新的共享內存；
• ngx_shm_free：用於釋放已經存在的共享內存。

mmap 系統調用簡述

void *mmap(void *start, size_t length, int prot, int flags, 
                int fd, off_t offset);

mmap 可以將磁盤文件映射到內存中，直接操作內存時 Linux 內核將負責同步內存和磁盤文件中的數據：

• fd 參數就指向需要同步的磁盤文件
• offset 則代表從文件的這個偏移量開始共享。
• 當 flags 參數中加入 MAP_ANON 或者 MAP_ANONYMOUS 參數時表示不使用文件映射方式，這時 fd 和 offset 參數就沒有意義了，也不需要傳遞，此時的 mmap 方法和 ngx_shm_alloc 的功能幾乎完全相同。
• length 參數就是將要在內存中開辟的線性地址空間大小。
• prot 參數則是操作這段共享內存的方式（如只讀或可讀可寫）。
• start 參數說明希望的共享內存起始映射地址，通常設為 NULL，即由內存選擇映射的起始地址。

MAP_ANON 是 MAP_ANONYMOUS 的同義詞，已過時。表示不使用文件映射方式，並且共享內存被初始化為0，因此忽略 mmap 中的 fd 和 offset 參數，但是為了可移植性，當 MAP_ANONYMOUS（或 MAP_ANON）被指定時，fd 應該設置為 -1。

如下為使用 mmap 實現的 ngx_shm_alloc 方法：

ngx_int_t 
ngx_shm_alloc(ngx_shm_t *shm)
{
    /* 開辟一塊 shm->size 大小且可讀/寫的共享內存，內存首地址存放在 shm->addr 中 */
    shm->addr = (u_char *)mmap(NULL, shm->size, 
                               PROT_READ|PROT_WRITE,
                               MAP_ANON|MAP_SHARED, -1, 0);
                
    if (shm->addr == MAP_FAILED) {
        ngx_log_error(NGX_LOG_ALERT, shm->log, ngx_errno,
                      "mmap(MAP_ANON|MAP_SHARED, %uz) failed", shm->size);
        return NGX_ERROR;
    }
    
    return NGX_OK;
}

當不在使用共享內存時，需要調用 munmap 或者 shmdt 來釋放共享內存：

int munmap(void *start, size_t length);

• start：指向共享內存的首地址
• length：表示這段共享內存的長度

Nginx 的 ngx_shm_free 方法封裝了該 munmap 方法：

void 
ngx_shm_free(ngx_shm_t *shm)
{
    if (munmap((void*) shm->addr, shm->size) == -1) {
        ngx_log_error(NGX_LOG_ALERT, shm->log, ngx_errno,
                      "munmap(%p, %uz) failed", shm->addr, shm->size);
    }
}

Nginx 各進程間共享數據的主要方式就是使用共享內存（在使用共享內存時，Nginx 一般是由 master 進程創建，在 master 進程 fork 出 worker 子進程后，所有的進程開始使用這塊內存中的數據）。

Nginx 的共享內存有三種實現：

• 不映射文件使用 mmap 分配共享內存（即上面的代碼）
• 以 /dev/zero 文件使用 mmap 映射共享內存
• 用 shmget 調用來分配共享內存

3. 原子操作

能夠執行原子操作的原子變量只有整型，包括無符號整型 ngx_atomic_uint_t 和有符號整型 ngx_aotmic_t，這兩種類型都是用了 volatile 關鍵字告訴 C 編譯器不要進行優化。

typedef volatile ngx_atomic_uint_t  ngx_atomic_t;

Nginx 提供了兩個方法來修改原子變量的值.

ngx_atomic_cmp_set

static ngx_inline ngx_atomic_uint_t
ngx_atomic_cmp_set(ngx_atomic_t *lock, ngx_atomic_uint_t old,
    ngx_atomic_uint_t set)

該方法會將 old 參數與原子變量 lock 的值進行比較，若相等，則將 lock 設為參數 set，同時返回 1；若不等，則直接返回 0。

ngx_atomic_fetch_add

static ngx_inline ngx_atomic_int_t
ngx_atomic_fetch_add(ngx_atomic_t *value, ngx_atomic_int_t add)

該方法會把原子變量 value 的值加上參數 add，同時返回之前 value 的值。

由於各種硬件體系架構，原子操作的實現不盡相同，如下為 Nginx 基於幾個硬件體系關於原子操作的實現。

3.1 不支持原子庫下的原子操作

當無法實現原子操作時，就只能用 volatile 關鍵字在 C 語言級別上模擬原子操作了。事實上，絕大多數體系架構都支持原子操作。

ngx_atomic_cmp_set 的實現如下：

static ngx_inline ngx_atomic_uint_t
ngx_atomic_cmp_set(ngx_atomic_t *lock, ngx_atomic_uint_t old,
    ngx_atomic_uint_t set)
{
    /* 當原子變量 lock 與 old 相等時，才能把 set 設置到 lock 中 */
    if (*lock == old) {
        *lock = set;
        return 1;
    }

    /* 若 lock 與 set 不等，返回 0 */
    return 0;
}

ngx_atomic_fetch_add 的實現如下：

static ngx_inline ngx_atomic_int_t
ngx_atomic_fetch_add(ngx_atomic_t *value, ngx_atomic_int_t add)
{
    ngx_atomic_int_t  old;

    /* 將原子變量 value 加上 add 后，返回原先 value 的值 */
    old = *value;
    *value += add;

    return old;
}

3.2 x86 架構下的原子操作

使用 GCC 編譯器在 C 語言中嵌入匯編語言的方式是使用 asm 關鍵字，如下：

__asm__ volatile ( 匯編語句部分
        : 輸出部分              /* 可選 */
        : 輸入部分              /* 可選 */
        : 破壞描述部分          /* 可選 */
        );

加入 volatile 關鍵字用於限制 GCC 編譯器對這段代碼做優化。

這段內聯的匯編語言包括 4 個部分。

1. 匯編語句部分

引號中所包含的匯編語句可以直接用占位符 % 來引用 C 語言中的變量（最多 10 個，%0 ~ %9）.

介紹兩個用到的匯編語句。第一個匯編語句是：

cmpxchgl r, [m]

Nginx 中對這一匯編語句有一段偽代碼注釋：

/*
 * "cmpxchgl r, [m]":
 *    
 *     /* 如果 eax 寄存器中的值等於 m */
 *     if (eax == [m]) {
 *         // 將 zf 標志設為 1
 *         zf = 1;
 *         // 將 m 值設為 r
 *         [m] = r;
 *     } else {
 *         zf = 0;
 *         eax = [m];
 *     }
 *
 * The "r" means the general register.
 * The "=a" and "a" are the %eax register.
 * Although we can return result in any register, we use "a" because it is
 * used in cmpxchgl anyway. The result is actually in %al but not in %eax
 * however, as the code is inlined gcc test %al as well as %eax,
 * and icc adds "movzbl %al, %eax" by itself.
 * 
 * The "cc" means that flags were changed.
 */

如上可以看出，cmpxchgl r, [m] 語句首先會將 m 與 eax 寄存器中的值進行比較，若相等，則把 m 中的值設為 r，並將 zf 標志位設為 1；否則將 zf 標志位設為 0。

第二個匯編語句是：sete [m]，它正好配合 cmpxchgl 語句使用。簡單的認為它的作用就是將 zf 標志位中的 0 或 1 設置到 m 中。

2. 輸出部分

這部分可以將寄存器中的值設置到 C 語言的變量中。

3. 輸入部分

可以將 C 語言中的變量設置到寄存器中。

4. 破壞描述部分

通知編譯器使用了哪些寄存器、內存。

如下為 ngx_atomic_cmp_set 方法在 x86 架構下的實現：

#if (NGX_SMP)
#define NGX_SMP_LOCK  "lock;"
#else
#define NGX_SMP_LOCK
#endif


static ngx_inline ngx_atomic_uint_t 
ngx_aotmic_cmp_set(ngx_aotmic_t *lock， ngx_atomic_uint_t old, 
    ngx_atomic_uint_t set)
{
    u_char res;
    
    // 在 C 語言中嵌入匯編語句
    __asm__ volatile (
    
    // 多核架構下首先鎖住總線
        NGX_SMP_LOCK
    // 將 *lock 的值與寄存器%eax中的 old 相比較，如果相等，則置 *lock 的置為 set
    "   cmpxchgl   %3, %1;   "
    // cmpxchgl 的比較若相等，則把 zf 標志位 1 寫入 res 變量，否則 res 為 0
    "   sete       %0;       "
    
    : "=a" (res) : "m" (*lock), "a" (old), "r" (set) : "cc", "memory");
    
    return res;
}

在嵌入匯編語句的輸入部分：

• "m" (*lock)：表示 *lock 變量是在內存中，操作 *lock 時直接通過內存（不使用寄存器）處理
• "a" (old)：表示把 old 變量寫入 eax 寄存器中
• "r" (set)：表示把 set 變量寫入通用寄存器中

這些都是為 cmpxchgl 語句做准備。

• "comxchgl %3, %1": 相當於 "cmpxchgl set, *lock"

因此，上面三行匯編語句的意思是：首先鎖住總線防止多核的並發執行，接着判斷原子變量 *lock 與 old 值是否相等，若相等，則把 *lock 值設為 set，同時設 res 為 1，返回該 res；若不相等，則設 res 為 0，返回該 res。

再介紹一個匯編語句：xaddl。Nginx 源碼對該語句 "xaddl r, [m]" 做的偽代碼注釋:

/*
 * "xaddl  r, [m]":
 *
 *     temp = [m];
 *     [m] += r;
 *     r = temp;
 *
 *
 * The "+r" means the general register.
 * The "cc" means that flags were changed.
 */

從該偽代碼知，xaddl 執行后 [m] 值將為 r 和 [m] 之和，而 r 中的值為原 [m] 值。

如下為 ngx_atomic_fetch_add 方法在 x86 架構下的實現：

/*
 * icc 8.1 and 9.0 compile broken code with -march=pentium4 option:
 * ngx_atomic_fetch_add() always return the input "add" value,
 * so we use the gcc 2.7 version.
 *
 * icc 8.1 and 9.0 with -march=pentiumpro option or icc 7.1 compile
 * correct code.
 */
static ngx_inline ngx_atomic_int_t
ngx_atomic_fetch_add(ngx_atomic_t *value, ngx_atomic_int_t add)
{
    __asm__ volatile (

    // 首先鎖住總線
         NGX_SMP_LOCK
    // *value 的值將會等於原先 *value 值與 add 值之和，而 add 為原 *value 值
    "    xaddl  %0, %1;   "

    : "+r" (add) : "m" (*value) : "cc", "memory");

    return add;
}

因此，ngx_atomic_fetch_add 將使得 *value 原子變量的值加上 add，同時返回原先 *value 的值。

3.3 自旋鎖

基於原子的操作，Nginx 實現了一個自旋鎖。自旋鎖是一種非睡眠鎖，也就是說，某進程如果試圖獲得自旋鎖，當發現鎖已經被其他進程獲得時，那么不會使得當前進程進入睡眠狀態，而是始終保持進程的可執行狀態，每當內核調度到這個進程執行時就持續檢查是否可以獲取到鎖。在拿不到鎖時，這個進程的代碼將會一直在自旋鎖代碼處執行，直到其他進程釋放了鎖且當前進程獲取到了鎖后，代碼才會繼續向下執行。

自旋鎖主要是為多處理器操作系統而設置的，它要解決的共享資源保護場景就是進程使用鎖的時間非常短（如果鎖的使用時間很久，自旋鎖就不合適，會占用大量的 CPU 資源）。如果使用鎖的進程不太希望自己進入睡眠狀態，特別它處理的是非常核心的事件時，這時就應該使用自旋鎖，其實大部分情況下 Nginx 的 worker 進程最好不要進入睡眠狀態，因為它非常繁忙，在這個進程的 epoll 上可能會有十萬甚至百萬的 TCP 連接等待着處理，進程一旦睡眠后必須等待其他事件的喚醒，這中間及其頻繁的進程間切換帶來的負載消耗可能無法讓用戶接受。

自旋鎖對於單處理器操作系統來說一樣是有效的，不進入睡眠狀態並不意味着其他可執行狀態的進程得不到執行。Linux 內核中對於每個處理器都有一個運行隊列，自旋鎖可以僅僅調整當前進程在運行隊列中的順序，或者調整進程的時間片，這都會為當前處理器上的其他進程提供被調度的機會，以使得鎖被其他進程釋放。

如下為 Nginx 實現的基於原子操作的自旋鎖方法 ngx_spinlock：

void
ngx_spinlock(ngx_atomic_t *lock, ngx_atomic_int_t value, ngx_uint_t spin)
{
    ngx_uint_t i, n;
    
    // 無法獲取鎖時進程的代碼將一直在這個循環中執行
    for ( ;; ) {
        // lock 為 0 表示鎖是沒有被其他進程持有的，這時將 lock 值設為 value 
        // 參數表示當前進程持有了鎖
        if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {
            // 獲取到鎖后 ngx_spinlock 方法才會返回
            return;
        }
        
        // 該變量是處理器的個數，當它大於 1 時表示處理多處理器系統中
        if (ngx_ncpu > 1) {
            // 在多處理器下，更好的做法是當前進程不要立刻"讓出"正在使用的 CPU 
            // 處理器，而是等待一段時間，看看其他處理器上的進程是否會釋放鎖，
            // 這會減少進程間切換的次數
            for (n = 1; n < spin; n <<= 1) {
            
                // 注意，隨着等待的次數越來越多，實際去檢查 lock 是否被釋放
                // 的頻繁會越來越小。為什么？因為檢查 lock 值更消耗 CPU，
                // 而執行 ngx_cpu_pause 對於 CPU 的能耗來說更為省電
                for (i = 0; i < n; i++) {
                
                    // ngx_cpu_pause 是在許多架構體系中專門為了自旋鎖而提供的
                    // 指令，它會告訴CPU現在處於自旋鎖等待狀態，通常一些CPU
                    // 會將自己置於節能狀態，降低功耗。注意，在執行
                    // ngx_cpu_pause 后，當前進程沒有 "讓出" 正使用的處理器
                    ngx_cpu_pasue();
                }
                
                // 檢查鎖是否被釋放了，如果 lock 值為0且釋放了鎖后，就把它的值設為
                // value，當前進程持有鎖成功並返回
                if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {
                    return;
                }
            }
        }
        
        // 當前進程仍然處理可執行狀態，但暫時"讓出"處理器，使得處理器優先調度其他
        // 可執行狀態的進程，這樣，在進程被內核再次調度時，在 for 循環代碼中可以期望
        // 其他進程釋放鎖。注意，不同的內核版本對於 sched_yield 系統調用的實現可能
        // 不同，但它們的目的都是暫時 "讓出" 處理器
        ngx_sched_yield();
    }
}

釋放鎖時需要 Nginx 模塊通過 ngx_atomic_cmp_set 方法將原子變量 lock 值設為 0。