题外话

有幸认识了一位做NGINX开源的大咖，加对方微信，竟然很爽快地同意了。闲聊几句关于写文章，他说，现在的NGINX的文章网上已经非常多了，完全不像十年前他开始学习NGINX时，几乎是一片空白，所有东西都需要自己一行一行地去看代码自己去理解和领悟。而且，写这些文章也不用指望有太多的点击量，更多是作为对自己学习的一个记录就好。这一点和我的初衷是一致的。写文章更多是记录和帮助自己更好地去理解问题。如果恰好也能对别人有所帮助也算是意外的惊喜了。很多时候自己觉得都明白了，真正地开始写文章才发现自己的欠缺。而且，即使是真的理解了怎么样更好地用文字清晰简洁地表述出来也不是一件容易的事情。

另外，谈话的过程中，深感他的纯粹，专注，平易和大气。这也是接触的几位从事开源开发工程师身上一个普遍的特质。向他们致敬。甚至不敢说向他们学习，因为学习意味着改变，改变是特别痛苦和困难的事情。认识自己都很难，更何况改变。。。

有些跑题，言归正传。

一 简介

NGINX采用多进程加IO多路复用的方式来处理并发请求。进程之间采用信号量，共享内存等方式进行信息共享和同步。采用共享内存方式在多进程之间进行数据共享，必然需要进程之间采用同步机制进行同步。

NGINX中最有名的同步锁就是ngx_accept_mutex。 NGINX使用它用来解决多进程并发处理的“网络惊群”问题。只有获得ngx_accept_mutex的进程才有资格处理当前的新的连接。这样避免了新的网络连接会唤醒所有的工作进程，竞争处理当前新的连接(因为kernel层面的支持，现在ngx_accept_mutex已经不再被需要了）。下面我们通过分析NGINX中ngx_accept_mutex的实现和使用，来理解NGINX中的同步机制。

二 原理

NGINX 通过变量类型ngx_shmtx_t自身实现了进程数据同步机制。根据如下ngx_shmtx_t的定义我们可以看到，根本不同的平台具体的实现方式可以是原子操作，文件锁两大类。

typedef struct {

#if (NGX_HAVE_ATOMIC_OPS)

    ngx_atomic_t  *lock;

#if (NGX_HAVE_POSIX_SEM)

    ngx_atomic_t  *wait;

    ngx_uint_t     semaphore;

    sem_t          sem;

#endif

#else

    ngx_fd_t       fd;

    u_char        *name;

#endif

    ngx_uint_t     spin;

} ngx_shmtx_t;

在有原子操作支持的情况下，如果同时有信号量的支持，可以加入信号量来减少加锁的开销。其原理就是在获取锁时，如果等待时间很长，在没有信号量支持的情况下，进程只能循环检查是否满足条件，这时候对cpu的消耗会比较大。引入了POSIX信号量的支持后，如果等待时候过长，进程可以通过信号量进行睡眠，让出cpu。等锁被别的进程释放时再进行唤起。这样可以达到节省cpu的目的。

文件锁的实现方式相对简单，通过对同一文件调用fctnl的不同操作，依靠fcntl函数本身提供的帮助来实现同步操作。下面我们着重分析支持原子操作的情况下，NGINX的同步机制的实现原理。

三代码解读

初始化

原子操作实现同步操作的情况需要共享内存的支持。每个进程拥有的ngx_shmtx_t结构成员lock需要指向共同的一段内存区域。

变量ngx_accept_mutex对应的成员lock初始化代码位于函数ngx_event_module_init中。在该函数中，首先通过ngx_shm_alloc函数分配一片共享内存区域。函数ngx_shm_alloc本身根据不同的平台通过系统调用mmap或者shmget来获取共享内存区域。在使用mmap或者shmget获取共享内存区域时，没有指定文件句柄，而且指定了MAP_ANON和MAP_SHARED标志。这样父进程创建完毕以后，通过fork就可以和子进程共享mmap获得的这片内存区域。

这函数里还有一个细节，对每一个需要被各个进程共享的变量，分配的size是128。而且，有一行注释明确说明了，为每一个变量分配的size不能小于cpu的cache line大小。这个要求是因为在x86和arm的cpu中，锁总线指令lock每次可以锁定的内存区域大小最小是一个cache line。如果几个共享变量共同位于同一个cache line中，那么锁定一个的同时也会把别的变量进行锁定。这样会影响系统执行的效率。

创建

调用共享内存相关的函数完成初始化以后，然后再调用函数ngx_shmtx_create函数创建ngx_accept_mutex。

在此函数中，把共享ngx_accept_mutex共享的锁的地址赋值给ngx_accept_mutex的lock成员。如果有POSIX的支持，需要同时调用sem_init初始化对应的信号量。

在调用sem_init时，第二个参数pshared的值为1，表明semaphore是进程间共享。创建成功后把信号量数量初始化为1。其中spin被赋值为2048，这个数值在有信号量支持的情况下用来规定忙等的时长。

ngx_int_t

ngx_shmtx_create(ngx_shmtx_t *mtx, ngx_shmtx_sh_t *addr, u_char *name)

{

    mtx->lock = &addr->lock;

    if (mtx->spin == (ngx_uint_t) -1) {

        return NGX_OK;

    }

    mtx->spin = 2048;

#if (NGX_HAVE_POSIX_SEM)

    mtx->wait = &addr->wait;

    if (sem_init(&mtx->sem, 1, 0) == -1) {

        ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, ngx_errno,

                      "sem_init() failed");

    } else {

        mtx->semaphore = 1;

    }

#endif

    return NGX_OK;

}

加锁

创建完成以后，就可以通过lock和unlock函数进行使用了。对应的lock函数是ngx_shmtx_trylock 和ngx_shmtx_lock。这两者的区别是，trylock尝试去获得lock资源，如果不成功就会立刻返回，由上层调用者根据返回结果再进行处理。而lock则一定要等到获取成功才能返回。在有信号量的情况下，如果等待时间太长，则通过信号量进行睡眠，否则则一直循环忙碌等待下去。

其中最关键的语句如下：

if (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid)) {

 }

首先判断lock成员是不是0，如果是，则调用ngx_atomic_cmp_set进行比较和赋值。在ngx_atomic_cmp_set函数中，使用了锁总线的操作来保证原子性。函数ngx_atomic_cmp_set的作用是，在锁总线期间如果lock的数值依然是0，则把它的值重新赋值为本worker的进程号。

函数ngx_atomic_cmp_set本身可以保证原子性，但是前面的*mtx->lock == 0判断语句却不可以保证比较的原子性。如果有多个worker进程同时执行上面的语句，在判断*mtx->lock == 0时，都满足条件，那么在执行ngx_atomic_cmp_set的时候，因为锁总线操作是互斥的，所以，总是有一个进程先锁总线成功并且把lock的数值修改成自己的进程好。这样，别的进程在虽然判断mtx->lock == 0满足条件，但是在ngx_atomic_cmp_set的时候，lock的值已经变成了别的进程的进程号。

但是，这样也不影响功能，ngx_atomic_cmp_set会首先再次比较lock的数值是否等于0,如果不等于0就说明被别的进程使用，然后ngx_shtm_lock函数进入自身循环进行尝试锁定。如果锁定失败，在多cpu的情况下，会循环检查是否可以锁定一段时间，并调用ngx_cpu_pause用来优化cpu使用时间。在忙等一段时间仍然不能获得信号量以后，如果有信号量的支持，则把进程睡眠让出cpu。

这其中的函数ngx_atomic_fetch_add也借助NGX_SMP_LOCK或者别原子性指令的支持实现原子操作。

void

ngx_shmtx_lock(ngx_shmtx_t *mtx)

{

    ngx_uint_t         i, n;

    ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0, "shmtx lock");

    for ( ;; ) {

        if (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid)) {

            return;

        }

        if (ngx_ncpu > 1) {

            for (n = 1; n < mtx->spin; n <<= 1) {

                for (i = 0; i < n; i++) {

                    ngx_cpu_pause();

                }

                if (*mtx->lock == 0

                    && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid))

                {

                    return;

                }

            }

        }

#if (NGX_HAVE_POSIX_SEM)

        if (mtx->semaphore) {

            (void) ngx_atomic_fetch_add(mtx->wait, 1);

            if (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid)) {

                (void) ngx_atomic_fetch_add(mtx->wait, -1);

                return;

            }

            ngx_log_debug1(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,

                           "shmtx wait %uA", *mtx->wait);

            while (sem_wait(&mtx->sem) == -1) {

                ngx_err_t  err;

                err = ngx_errno;

                if (err != NGX_EINTR) {

                    ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, err,

                                  "sem_wait() failed while waiting on shmtx");

                    break;

                }

            }

            ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,

                           "shmtx awoke");

            continue;

        }

#endif

        ngx_sched_yield();

    }

}

解锁

使用lock函数锁定以后，可以使用unlock函数进行释放。对应的函数是ngx_shmtx_unlock。它的逻辑相对简单，首先通过ngx_atomic_cmp_set函数比较当前的lock数值是否是自己的进程号，如果不是则就返回。如果是，则把lock数值设置为0并且，在有信号量支持的情况下通过ngx_shmtx_wakeup唤醒睡眠在信号量上的进程。

void

ngx_shmtx_unlock(ngx_shmtx_t *mtx)

{

    if (mtx->spin != (ngx_uint_t) -1) {

        ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0, "shmtx unlock");

    }

    if (ngx_atomic_cmp_set(mtx->lock, ngx_pid, 0)) {

        ngx_shmtx_wakeup(mtx);

    }

}

销毁

使用完毕以后可以通过函数ngx_shmtx_destroy 进行销毁。

三 总结

以上就是NGINX中通过原子性支持实现的同步操作。除此以外,NGINX本身还是先了spinlock，rwlock等同步机制，但是大体原理都类似。