14.8.2　原子变量实现的ngx_shmtx_t锁

当Nginx判断当前操作系统支持原子变量时，将会优先使用原子变量实现表14-1中的5种方法（即原子变量锁的优先级高于文件锁）。不过，同时还需要判断其是否支持信号量，因为支持信号量后进程有可能进入睡眠状态。下面介绍一下如何使用原子变量和信号量来实现ngx_shmtx_t互斥锁，注意，它比文件锁的实现要复杂许多。

ngx_shmtx_t结构中的lock原子变量表示当前锁的状态。为了便于理解，我们还是用接近自然语言的方式来说明这个锁，当lock值为0或者正数时表示没有进程持有锁；当lock值为负数时表示有进程正持有锁（这里的正、负数仅相对于32位系统下有符号的整型变量）。Nginx是怎样快速判断lock值为“正数”或者“负数”的呢？很简单，因为有符号整型的最高位是用于表示符号的，其中0表示正数，1表示负数，所以，在确定整型val是负数或者正数时，可通过判断（val＆0x80000000）==0语句的真假进行。

下面看一下初始化ngx_shmtx_t互斥锁的ngx_shmtx_create方法究竟做了些什么事情。

ngx_int_t ngx_shmtx_create（ngx_shmtx_tmtx,voidaddr,u_char*name）

{

mtx-＞lock=addr；

//注意，当spin值为-1时，表示不能使用信号量，这时直接返回成功

if（mtx-＞spin==（ngx_uint_t）-1）{

return NGX_OK；

}

//spin值默认为2048

mtx-＞spin=2048；

//同时使用信号量

if（NGX_HAVE_POSIX_SEM）

//以多进程使用的方式初始化sem信号量，sem初始值为0

if（sem_init（＆mtx-＞sem，1，0）==-1）{

ngx_log_error（NGX_LOG_ALERT,ngx_cycle-＞log,ngx_errno，

"sem_init（）failed"）；

}else{

//在信号量初始化成功后，设置semaphore标志位为1

mtx-＞semaphore=1；

}

endif

return NGX_OK；

}

spin和semaphore成员都将决定ngx_shmtx_lock阻塞锁的行为。

ngx_shmtx_destory方法的唯一目的就是释放信号量，如下所示。

void ngx_shmtx_destory（ngx_shmtx_t*mtx）

{

//支持信号量时才有代码需要执行

if（NGX_HAVE_POSIX_SEM）

/当这把锁的spin值不为（ngx_uint_t）-1时，且初始化信号量成功，semaphore标志位才为1/

if（mtx-＞semaphore）{

//销毁信号量

if（sem_destroy（＆mtx-＞sem）==-1）{

ngx_log_error（NGX_LOG_ALERT,ngx_cycle-＞log,ngx_errno，"sem_destroy（）failed"）；

}

}

endif

}

以非阻塞方式获取锁的ngx_shmtx_trylock方法较为简单，可直接判断lock原子变量的值，当它为非负数时，直接将其置为负数即表示持有锁成功。怎样把0或者正数置为负数呢？很简单，使用语句val|0x80000000即可把非负数的val变为负数，这种方法效率最高，即直接修改val的最高符号标志位为1。

ngx_uint_t ngx_shmtx_trylock（ngx_shmtx_t*mtx）

{

ngx_atomic_uint_t val；

//取出lock锁的值，通过判断它是否为非负数来确定锁状态

val=*mtx-＞lock；

/如果val为0或者正数，则说明没有进程持有锁，这时调用ngx_atomic_cmp_set方法将lock锁改为负数，表示当前进程持有了互斥锁/

return（（val＆0x80000000）==0＆＆

ngx_atomic_cmp_set（mtx-＞lock,val,val|0x80000000））；

}

注意（val＆0x80000000）==0是一行语句，而ngx_atomic_cmp_set（mtx-＞lock,val,val|0x80000000）又是一行语句，多进程的Nginx服务将有可能出现虽然第1行语句执行成功（表示锁未被任何进程持有），但在执行第2行语句前，又有一个进程拿到了锁，这时第2行语句将会执行失败。这正是ngx_atomic_cmp_set方法自身先判断lock值是否为非负数val的原因，只有lock值为非负数val，它才会确定将lock值赋为负数val|0x80000000并返回1，否则返回0（详见14.3.2节）。

阻塞式获取互斥锁的ngx_shmtx_lock方法较为复杂，在不支持信号量时它与14.3.3节介绍的自旋锁几乎完全相同，但在支持了信号量后，它将有可能使进程进入睡眠状态。下面我们分析一下它的操作步骤。

void ngx_shmtx_lock（ngx_shmtx_t*mtx）

{

ngx_uint_t i,n；

ngx_atomic_uint_t val；

//没有拿到锁之前是不会跳出循环的

for（；）{

/lock值是当前的锁状态。注意，lock一般是在共享内存中的，它可能会时刻变化，而val是当前进程的栈中变量，下面代码的执行中它可能与lock值不一致/

val=*mtx-＞lock；

/如果val为非负数，则说明锁未被持有。下面试图通过修改lock值为负数来持有锁/

if（（val＆0x80000000）==0

＆＆ngx_atomic_cmp_set（mtx-＞lock,val,val|0x80000000））

{

/*在成功地将lock值由原先的val改为非负数后，表示成功地持有了锁，ngx_shmtx_lock

方法结束*/

return；

}

//仅在多处理器状态下spin值才有意义，否则PAUSE指令是不会执行的

if（ngx_ncpu＞1）{

//循环执行PAUSE，检查锁是否已经释放

for（n=1；n＜mtx-＞spin；n＜＜=1）{

//随着长时间没有获得到锁，将会执行更多次PAUSE才会检查锁

for（i=0；i＜n；i++）{

//对于多处理器系统，执行ngx_cpu_pause可以降低功耗

ngx_cpu_pause（）；

}

//再次由共享内存中获得lock原子变量的值

val=*mtx-＞lock；

/检查lock是否已经为非负数，即锁是否已经被释放，如果锁已经释放，那么会通过将lock原子变量值设置为负数来表示当前进程持有了锁/

if（（val＆0x80000000）==0

＆＆ngx_atomic_cmp_set（mtx-＞lock,val,val|0x80000000））

{

//持有锁成功后立刻返回

return；

}

}

}

//支持信号量时才继续执行

if（NGX_HAVE_POSIX_SEM）

//semaphore标志位为1才使用信号量

if（mtx-＞semaphore）{

//重新获取一次可能在共享内存中的lock原子变量

val=*mtx-＞lock；

//如果lock值为负数，则lock值加上1

if（（val＆0x80000000）

＆＆ngx_atomic_cmp_set（mtx-＞lock,val,val+1））

{

/检查信号量sem的值，如果sem值为正数，则sem值减1，表示拿到了信号量互斥锁，同时sem_wait方法返回0。如果sem值为0或者负数，则当前进程进入睡眠状态，等待其他进程使用ngx_shmtx_unlock方法释放锁（等待sem信号量变为正数），到时Linux内核会重新调度当前进程，继续检查sem值是否为正，重复以上流程/

while（sem_wait（＆mtx-＞sem）==-1）{

ngx_err_t err；

err=ngx_errno；

//当EINTR信号出现时，表示sem_wait只是被打断，并不是出错

if（err！=NGX_EINTR）{

break；

}

}

}

//循环检查lock锁的值，注意，当使用信号量后不会调用sched_yield

continue；

}

endif

//在不使用信号量时，调用sched_yield将会使当前进程暂时“让出”处理器

ngx_sched_yield（）；

}

}

可以看到，在不使用信号量时（例如，NGX_HAVE_POSIX_SEM宏没打开，或者spin的值为（ngx_uint_t）-1），ngx_shmtx_lock方法与ngx_spinlock方法非常相似，而在使用信号量后将会使用可能让进程进入睡眠的sem_wait方法代替“让出”处理器的ngx_sched_yield方法。这里不建议在Nginx worker进程中使用带信号量的ngx_shmtx_lock取锁方法。

ngx_shmtx_unlock方法会释放锁，虽然这个释放过程不会阻塞进程，但设置原子变量lock值时是可能失败的，因为多进程在同时修改lock值，而ngx_atomic_cmp_set方法要求参数old的值与lock值相同时才能修改成功，因此，ngx_atomic_cmp_set方法会在循环中反复执行，直到返回成功为止。该方法的实现如下所示：

void ngx_shmtx_unlock（ngx_shmtx_t*mtx）

{

ngx_atomic_uint_t val,old,wait；

//试图循环重置lock值为正数，此时务必将互斥锁释放

for（；）{

//由共享内存中的lock原子变量取出锁状态

old=*mtx-＞lock；

//通过把最高位置为0，将lock变为正数

wait=old＆0x7fffffff；

//如果变为正数的lock不是0，则减去1

val=wait?wait-1：0；

//将lock锁的值设为非负数val

if（ngx_atomic_cmp_set（mtx-＞lock,old,val））{

//设置锁成功后才能跳出循环，否则将持续地试图修改lock值为非负数

break；

}

}

if（NGX_HAVE_POSIX_SEM）

/如果lock锁原先的值为0，也就是说，并没有让某个进程持有锁，这时直接返回；或者，semaphore标志位为0，表示不需要使用信号量，也立即返回/

if（wait==0||！mtx-＞semaphore）{

return；

}

/通过sem_post将信号量sem加1，表示当前进程释放了信号量互斥锁，通知其他进程的sem_wait继续执行/

if（sem_post（＆mtx-＞sem）==-1）{

ngx_log_error（NGX_LOG_ALERT,ngx_cycle-＞log,ngx_errno，

"sem_post（）failed while wake shmtx"）；

}

endif

}

由于原子变量实现的这5种互斥锁方法是Nginx中使用最广泛的同步方式，当需要Nginx支持数以万计的并发TCP请求时，通常都会把spin值设为（ngx_uint_t）-1。这时的互斥锁在取锁时都会采用自旋锁，对于Nginx这种单进程处理大量请求的场景来说是非常适合的，能够大量降低不必要的进程间切换带来的消耗。