Linux 惊群效应之 Nginx 解决方案

// 当accept的时候，如果没有连接则会一直阻塞（没有设置非阻塞）
// 其阻塞函数就是：inet_csk_accept（accept的原型函数）
struct sock *inet_csk_accept(struct sock *sk, int flags, int *err)
{
    ...
      // 等待连接
    error = inet_csk_wait_for_connect(sk, timeo);
    ...
  }


static int inet_csk_wait_for_connect(struct sock *sk, long timeo)
{
    ...
    for (;;) {
        // 只有一个进程会被唤醒。
        // 非exclusive的元素会加在等待队列前头，exclusive的元素会加在所有非exclusive元素的后头。
        prepare_to_wait_exclusive(sk_sleep(sk), &wait,TASK_INTERRUPTIBLE);
      }
      ...
 }


void prepare_to_wait_exclusive(wait_queue_head_t *q, wait_queue_t *wait, int state)
  {
      unsigned long flags;
      // 设置等待队列的flag为EXCLUSIVE，设置这个就是表示一次只会有一个进程被唤醒，我们等会就会看到这个标记的作用。
      // 注意这个标志，唤醒的阶段会使用这个标志。
      wait->flags |= WQ_FLAG_EXCLUSIVE;
      spin_lock_irqsave(&q->lock, flags);
      if (list_empty(&wait->task_list))
          // 加入等待队列
          __add_wait_queue_tail(q, wait);
      set_current_state(state);
      spin_unlock_irqrestore(&q->lock, flags);
  }

// 当有tcp连接完成，就会从半连接队列拷贝socket到连接队列，这个时候我们就可以唤醒阻塞的accept了。
int tcp_v4_do_rcv(struct sock *sk, struct sk_buff *skb)
{
    ...
    // 关注此函数
    if (tcp_child_process(sk, nsk, skb)) {
        rsk = nsk;
        goto reset;
      }
    ...
}


int tcp_child_process(struct sock *parent, struct sock *child, struct sk_buff *skb)
{
    ...
    // Wakeup parent, send SIGIO 唤醒父进程
    if (state == TCP_SYN_RECV && child->sk_state != state)
        // 调用sk_data_ready通知父进程
        // 查阅资料我们知道tcp中这个函数对应是sock_def_readable
        // 而sock_def_readable会调用wake_up_interruptible_sync_poll来唤醒队列
        parent->sk_data_ready(parent, 0);
      }
    ...
}


void __wake_up_sync_key(wait_queue_head_t *q, unsigned int mode, int nr_exclusive, void *key)
{
      ...
      // 关注此函数
    __wake_up_common(q, mode, nr_exclusive, wake_flags, key);
      spin_unlock_irqrestore(&q->lock, flags);
      ...
}


static void __wake_up_common(wait_queue_head_t *q, unsigned int mode, int nr_exclusive, int wake_flags, void *key)
{   ...
   // 传进来的nr_exclusive是1   // 所以flags & WQ_FLAG_EXCLUSIVE为真的时候，执行一次，就会跳出循环   // 我们记得accept的时候，加到等待队列的元素就是WQ_FLAG_EXCLUSIVE的   list_for_each_entry_safe(curr, next, &q->task_list, task_list) {         unsigned flags = curr->flags;         if (curr->func(curr, mode, wake_flags, key)       && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
        break;   }   ...}

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;
  }

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

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;
  }

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

// 如果使用了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_accept_disabled = ngx_cycle->connection_n / 8
                  - ngx_cycle->free_connection_n;

// 如果有使用mutex，则才会进行处理。
  if (ngx_use_accept_mutex)
 {
      // 如果大于0,则跳过下面的锁的处理，并减一。
      if (ngx_accept_disabled > 0) {
          ngx_accept_disabled--;
    } else {
          // 试着获得锁，如果出错则返回。
          if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
              return;
          }
          // 如果ngx_accept_mutex_held为1,则说明已经获得锁，此时设置flag，这个flag后面会解释。
        if (ngx_accept_mutex_held) {
              flags |= NGX_POST_EVENTS;
          } else {
              // 否则，设置timer，也就是定时器。接下来会解释这段。
              if (timer == NGX_TIMER_INFINITE
                   || timer > ngx_accept_mutex_delay) {
                  timer = ngx_accept_mutex_delay;
              }
          }
      }
}

// 如果ngx_posted_accept_events不为NULL，则说明有accept event需要nginx处理。
  if (ngx_posted_accept_events) {
          ngx_event_process_posted(cycle, &ngx_posted_accept_events);


  }

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;
}