惊群效应之Nginx处理

惊群概述

在说nginx前，先来看看什么是“惊群”？简单说来，多线程/多进程（linux下线程进程也没多大区别）等待同一个socket事件，当这个事件发生时，这些线程/进程被同时唤醒，就是惊群。可以想见，效率很低下，许多进程被内核重新调度唤醒，同时去响应这一个事件，当然只有一个进程能处理事件成功，其他的进程在处理该事件失败后重新休眠（也有其他选择）。这种性能浪费现象就是惊群。

惊群通常发生在server 上，当父进程绑定一个端口监听socket，然后fork出多个子进程，子进程们开始循环处理（比如accept）这个socket。每当用户发起一个TCP连接时，多个子进程同时被唤醒，然后其中一个子进程accept新连接成功，余者皆失败，重新休眠。

那么，我们不能只用一个进程去accept新连接么？然后通过消息队列等同步方式使其他子进程处理这些新建的连接，这样惊群不就避免了？没错，惊群是避免了，但是效率低下，因为这个进程只能用来accept连接。对多核机器来说，仅有一个进程去accept，这也是程序员在自己创造accept瓶颈。所以，我仍然坚持需要多进程处理accept事件。

其实，在linux2.6内核上，accept系统调用已经不存在惊群了（至少我在2.6.18内核版本上已经不存在）。大家可以写个简单的程序试下，在父进程中bind,listen，然后fork出子进程，所有的子进程都accept这个监听句柄。这样，当新连接过来时，大家会发现，仅有一个子进程返回新建的连接，其他子进程继续休眠在accept调用上，没有被唤醒。

但是很不幸，通常我们的程序没那么简单，不会愿意阻塞在accept调用上，我们还有许多其他网络读写事件要处理，linux下我们爱用epoll解决非阻塞socket。所以，即使accept调用没有惊群了，我们也还得处理惊群这事，因为epoll有这问题。上面说的测试程序，如果我们在子进程内不是阻塞调用accept，而是用epoll_wait，就会发现，新连接过来时，多个子进程都会在epoll_wait后被唤醒！

nginx就是这样，master进程监听端口号（例如80），所有的nginx worker进程开始用epoll_wait来处理新事件（linux下），如果不加任何保护，一个新连接来临时，会有多个worker进程在epoll_wait后被唤醒，然后发现自己accept失败。

Nginx 解决方案之锁的设计

首先我们要知道在用户空间进程间锁实现的原理，起始原理很简单，就是能弄一个让所有进程共享的东西，比如 mmap 的内存，比如文件，然后通过这个东西来控制进程的互斥。

Nginx 中使用的锁是自己来实现的，这里锁的实现分为两种情况，一种是支持原子操作的情况，也就是由 NGX_HAVE_ATOMIC_OPS 这个宏来进行控制的，一种是不支持原子操作，这是是使用文件锁来实现。

锁结构体

如果支持原子操作，则我们可以直接使用 mmap，然后 lock 就保存 mmap 的内存区域的地址
如果不支持原子操作，则我们使用文件锁来实现，这里 fd 表示进程间共享的文件句柄，name 表示文件名

typedef struct {  
#if (NGX_HAVE_ATOMIC_OPS)  ngx_atomic_t  *lock;  
#else  ngx_fd_t       fd;  u_char        *name;  
#endif  
} ngx_shmtx_t;

原子锁创建

// 如果支持原子操作的话，非常简单，就是将共享内存的地址付给loc这个域
ngx_int_t ngx_shmtx_create(ngx_shmtx_t *mtx, void *addr, u_char *name)  
{  mtx->lock = addr;  return NGX_OK;  
} 

原子锁获取

TryLock，它是非阻塞的，也就是说它会尝试的获得锁，如果没有获得的话，它会直接返回错误。
Lock，它也会尝试获得锁，而当没有获得他不会立即返回，而是开始进入循环然后不停的去获得锁，知道获得。不过 Nginx 这里还有用到一个技巧，就是每次都会让当前的进程放到 CPU 的运行队列的最后一位，也就是自动放弃 CPU。

原子锁实现

如果系统库支持的情况，此时直接调用OSAtomicCompareAndSwap32Barrier，即 CAS。

#define ngx_atomic_cmp_set(lock, old, new)                                   OSAtomicCompareAndSwap32Barrier(old, new, (int32_t *) lock) 

如果系统库不支持这个指令的话，Nginx 自己还用汇编实现了一个。

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)  
{  u_char  res;  __asm__ volatile (  NGX_SMP_LOCK  "    cmpxchgl  %3, %1;   "  "    sete      %0;       "  : "=a" (res) : "m" (*lock), "a" (old), "r" (set) : "cc", "memory");  return res;  
}

原子锁释放

Unlock 比较简单，和当前进程 id 比较，如果相等，就把 lock 改为 0，说明放弃这个锁。

#define ngx_shmtx_unlock(mtx) (void) ngx_atomic_cmp_set((mtx)->lock, ngx_pid, 0)  

Nginx 解决方案之惊群效应

nginx的每个worker进程在函数ngx_process_events_and_timers中处理事件，(void) ngx_process_events(cycle, timer, flags);封装了不同的事件处理机制，在linux上默认就封装了epoll_wait调用。我们来看看ngx_process_events_and_timers为解决惊群做了什么：

void
ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
。。。 。。。//ngx_use_accept_mutex表示是否需要通过对accept加锁来解决惊群问题。//当nginx worker进程数>1时且配置文件中打开accept_mutex时，这个标志置为1if (ngx_use_accept_mutex) {//ngx_accept_disabled表示此时满负荷，没必要再处理新连接了，我们在nginx.conf曾经配置了每一个nginx worker//进程能够处理的最大连接数，当达到最大数的7/8时，ngx_accept_disabled为正，说明本nginx worker进程非常繁忙，//将不再去处理新连接，这也是个简单的负载均衡if (ngx_accept_disabled > 0) {ngx_accept_disabled--;} else {//获得accept锁，多个worker仅有一个可以得到这把锁。获得锁不是阻塞过程，都是立刻返回，获取成功的话//ngx_accept_mutex_held被置为1。拿到锁，意味着监听句柄被放到本进程的epoll中了，如果没有拿到锁，//则监听句柄会被从epoll中取出。if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {return;}//拿到锁的话，置flag为NGX_POST_EVENTS，这意味着ngx_process_events函数中，任何事件都将延后处理，//会把accept事件都放到ngx_posted_accept_events链表中，epollin|epollout事件都放到//ngx_posted_events链表中if (ngx_accept_mutex_held) {flags |= NGX_POST_EVENTS;} else {//拿不到锁，也就不会处理监听的句柄，这个timer实际是传给epoll_wait的超时时间，//修改为最大ngx_accept_mutex_delay意味着epoll_wait更短的超时返回，以免新连接长时间没有得到处理if (timer == NGX_TIMER_INFINITE|| timer > ngx_accept_mutex_delay){timer = ngx_accept_mutex_delay;}}}}
。。。 。。。//linux下，调用ngx_epoll_process_events函数开始处理(void) ngx_process_events(cycle, timer, flags);
。。。 。。。//如果ngx_posted_accept_events链表有数据，就开始accept建立新连接if (ngx_posted_accept_events) {ngx_event_process_posted(cycle, &ngx_posted_accept_events);}//释放锁后再处理下面的EPOLLIN EPOLLOUT请求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);//然后再处理正常的数据读写请求。因为这些请求耗时久，所以在ngx_process_events里NGX_POST_EVENTS//标志将事件都放入ngx_posted_events链表中，延迟到锁释放了再处理。if (ngx_posted_events) {if (ngx_threaded) {ngx_wakeup_worker_thread(cycle);} else {ngx_event_process_posted(cycle, &ngx_posted_events);}}
｝

从上面的注释可以看到，无论有多少个nginx worker进程，同一时刻只能有一个worker进程在自己的epoll中加入监听的句柄。这个处理accept的nginx worker进程置flag为NGX_POST_EVENTS，这样它在接下来的ngx_process_events函数（在linux中就是ngx_epoll_process_events函数）中不会立刻处理事件，延后，先处理完所有的accept事件后，释放锁，然后再处理正常的读写socket事件。我们来看下ngx_epoll_process_events是怎么做的：

static ngx_int_t
ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
。。。 。。。events = epoll_wait(ep, event_list, (int) nevents, timer);
。。。 。。。ngx_mutex_lock(ngx_posted_events_mutex);for (i = 0; i < events; i++) {c = event_list[i].data.ptr;。。。 。。。rev = c->read;if ((revents & EPOLLIN) && rev->active) {
。。。 。。。
//有NGX_POST_EVENTS标志的话，就把accept事件放到ngx_posted_accept_events队列中，
//把正常的事件放到ngx_posted_events队列中延迟处理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) {
。。。 。。。
//同理，有NGX_POST_EVENTS标志的话，写事件延迟处理，放到ngx_posted_events队列中if (flags & NGX_POST_EVENTS) {ngx_locked_post_event(wev, &ngx_posted_events);} else {wev->handler(wev);}}}ngx_mutex_unlock(ngx_posted_events_mutex);return NGX_OK;
}

看看ngx_use_accept_mutex在何种情况下会被打开：

// 如果使用了 master worker，并且 worker 个数大于 1，并且配置文件里面有设置使用
// accept_mutex. 的话，设置ngx_use_accept_mutex  if (ccf->master && ccf->worker_processes > 1 && ecf->accept_mutex) {  ngx_use_accept_mutex = 1;  // 下面这两个变量后面会解释。  ngx_accept_mutex_held = 0;  ngx_accept_mutex_delay = ecf->accept_mutex_delay;  } else {  ngx_use_accept_mutex = 0;  }

ngx_use_accept_mutex 这个变量，如果有这个变量，说明 Nginx 有必要使用 accept 互斥体，这个变量的初始化在 ngx_event_process_init 中。
ngx_accept_mutex_held 表示当前是否已经持有锁。
ngx_accept_mutex_delay 表示当获得锁失败后，再次去请求锁的间隔时间，这个时间可以在配置文件中设置的。

再看看有些负载均衡作用的ngx_accept_disabled是怎么维护的，在ngx_event_accept函数中：

ngx_accept_disabled = ngx_cycle->connection_n / 8 - ngx_cycle->free_connection_n;

表明，当已使用的连接数占到在nginx.conf里配置的worker_connections总数的7/8以上时，ngx_accept_disabled为正，这时本worker将ngx_accept_disabled减1，而且本次不再处理新连接。

最后，我们看下ngx_trylock_accept_mutex函数是怎么玩的：

ngx_int_t ngx_trylock_accept_mutex(ngx_cycle_t *cycle)  
{  // 尝试获得锁  if (ngx_shmtx_trylock(&ngx_accept_mutex)) {  // 如果本来已经获得锁，则直接返回Ok  if (ngx_accept_mutex_held  && ngx_accept_events == 0  && !(ngx_event_flags & NGX_USE_RTSIG_EVENT))  {  return NGX_OK;  }  // 到达这里，说明重新获得锁成功，因此需要打开被关闭的listening句柄。  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;  return NGX_OK;  }  // 如果我们前面已经获得了锁，然后这次获得锁失败// 则说明当前的listen句柄已经被其他的进程锁监听// 因此此时需要从epoll中移出调已经注册的listen句柄// 这样就很好的控制了子进程的负载均衡  if (ngx_accept_mutex_held) {  if (ngx_disable_accept_events(cycle) == NGX_ERROR) {  return NGX_ERROR;  }  // 设置锁的持有为0.  ngx_accept_mutex_held = 0;  }  return NGX_OK;  
}                            

如上代码，当一个连接来的时候，此时每个进程的 epoll 事件列表里面都是有该 fd 的。抢到该连接的进程先释放锁，在 accept。没有抢到的进程把该 fd 从事件列表里面移除，不必再调用 accept，造成资源浪费。

同时由于锁的控制(以及获得锁的定时器)，每个进程都能相对公平的 accept 句柄，也就是比较好的解决了子进程负载均衡。

总结：

简单了说，就是同一时刻只允许一个nginx worker在自己的epoll中处理监听句柄。它的负载均衡也很简单，当达到最大connection的7/8时，本worker不会去试图拿accept锁，也不会去处理新连接，这样其他nginx worker进程就更有机会去处理监听句柄，建立新连接了。而且，由于timeout的设定，使得没有拿到锁的worker进程，去拿锁的频繁更高。