nginx worker进程循环的实现

typedef struct {
  // 其余属性...
  
  ngx_socket_t channel[2];
} ngx_process_t;

/**
 * 启动n个worker子进程，并设置好每个子进程与master父进程之间使用socketpair
 * 系统调用建立起来的socket句柄通信机制
 */
static void ngx_start_worker_processes(ngx_cycle_t *cycle, ngx_int_t n, ngx_int_t type) {
 ngx_int_t i;
 ngx_channel_t ch;
 
 ngx_memzero(&ch, sizeof(ngx_channel_t));
 ch.command = NGX_CMD_OPEN_CHANNEL;

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

  // spawn是产卵的意思，这里就是生成一个子进程的意思，而该子进程所进行的事件循环就是
  // ngx_worker_process_cycle()方法，这里的ngx_worker_process_cycle是worker进程处理事件的循环，
  // worker进程在一个无限for循环中，不断的检查相应的事件模型中是否存在对应的事件，
  // 然后将accept事件和read、write事件分开放入两个队列中，最后在事件循环中不断的处理事件
  ngx_spawn_process(cycle, ngx_worker_process_cycle, 
           (void *) (intptr_t) i, "worker process", type);

  // 下面的这段代码的主要作用是将新建进程这个事件通知到其他的进程，上面的
  // ch.command = NGX_CMD_OPEN_CHANNEL;中NGX_CMD_OPEN_CHANNEL表示的就是当前是新建了一个进程，
  // 而ngx_process_slot存储的就是该新建进程所存放的数组位置，这里需要进行广播的原因在于，
  // 每个子进程被创建后，其内存数据都是复制的父进程的，但是ngx_processes数组是每个进程都有一份的，
  // 因而数组中先创建的子进程是没有后创建的子进程的数据的，但是master进程是有所有子进程的数据的，
  // 因而这里master进程创建子进程之后，其就会向ngx_processes数组的每个进程的channel[0]上
  // 写入当前广播的事件，也即这里的ch，通过这种方式，每个子进程接收到这个事件之后，
  // 都会尝试更新其所保存的ngx_processes数据信息
  ch.pid = ngx_processes[ngx_process_slot].pid;
  ch.slot = ngx_process_slot;
  ch.fd = ngx_processes[ngx_process_slot].channel[0];

  // 广播事件
  ngx_pass_open_channel(cycle, &ch);
 }
}

ngx_pid_t ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data, char *name, ngx_int_t respawn) {
 u_long on;
 ngx_pid_t pid;
 ngx_int_t s;

 if (respawn >= 0) {
  s = respawn;

 } else {
  // 在ngx_processes数组中存储了当前创建的所有进程，而ngx_last_process则是当前当前记录的最后一个
  // process在ngx_processes中的下一个位置的索引，只不过ngx_processes中记录的进程有可能有部分
  // 已经失效了。当前循环就是从头开始查找是否有某个进程已经失效了，如果已经失效了，则复用该进程位置，
  // 否则直接使用ngx_last_process所指向的位置
  for (s = 0; s < ngx_last_process; s++) {
   if (ngx_processes[s].pid == -1) {
    break;
   }
  }

  // 这里说明所创建的进程数达到了最大限度
  if (s == NGX_MAX_PROCESSES) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
          "no more than %d processes can be spawned",
          NGX_MAX_PROCESSES);
   return NGX_INVALID_PID;
  }
 }

 // NGX_PROCESS_DETACHED标志表示当前fork出来的进程与原来的父进程没有任何关系，比如进行nginx升级时，
 // 新生成的master进程就与原先的master进程没有关系
 if (respawn != NGX_PROCESS_DETACHED) {

  /* Solaris 9 still has no AF_LOCAL */

  // 这里的socketpair()方法的主要作用是生成一对套接字流，用于主进程和子进程的通信，这一对套接字会
  // 存储在ngx_processes[s].channel中，本质上这个字段是一个长度为2的整型数组。在主进程和子进程
  // 进行通信的之前，主进程会关闭其中一个，而子进程会关闭另一个，然后相互之间往未关闭的另一个文件描述符中
  // 写入或读取数据即可实现通信。
  // AF_UNIX表示当前使用的是UNIX文件形式的socket地址族
  // SOCK_STREAM指定了当前套接字建立的通信方式是管道流，并且这个管道流是双向的，
  // 即管道双方都可以进行读写操作
  // 第三个参数protocol必须为0
  if (socketpair(AF_UNIX, SOCK_STREAM, 0, ngx_processes[s].channel) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "socketpair() failed while spawning \"%s\"", name);
   return NGX_INVALID_PID;
  }

  ngx_log_debug2(NGX_LOG_DEBUG_CORE, cycle->log, 0,
          "channel %d:%d",
          ngx_processes[s].channel[0],
          ngx_processes[s].channel[1]);

  // 将ngx_processes[s].channel[0]设置为非阻塞模式
  if (ngx_nonblocking(ngx_processes[s].channel[0]) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          ngx_nonblocking_n
            " failed while spawning \"%s\"",
          name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;
  }

  // 将ngx_processes[s].channel[1]设置为非阻塞模式
  if (ngx_nonblocking(ngx_processes[s].channel[1]) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          ngx_nonblocking_n
            " failed while spawning \"%s\"",
          name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;
  }

  on = 1;
  // 将ngx_processes[s].channel[0]套接字管道设置为异步模式
  if (ioctl(ngx_processes[s].channel[0], FIOASYNC, &on) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "ioctl(FIOASYNC) failed while spawning \"%s\"", name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;
  }

  // 当前还处于主进程中，这里的ngx_pid指向了主进程的进程id，当前方法的作用主要是将
  // ngx_processes[s].channel[0]的操作权限设置给主进程，也就是说主进程通过向
  // ngx_processes[s].channel[0]写入和读取数据来与子进程进行通信
  if (fcntl(ngx_processes[s].channel[0], F_SETOWN, ngx_pid) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "fcntl(F_SETOWN) failed while spawning \"%s\"", name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;
  }

  // FD_CLOEXEC表示当前指定的套接字管道在子进程中可以使用，但是在execl()执行的程序中不可使用
  if (fcntl(ngx_processes[s].channel[0], F_SETFD, FD_CLOEXEC) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "fcntl(FD_CLOEXEC) failed while spawning \"%s\"",
          name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;
  }

  // FD_CLOEXEC表示当前指定的套接字管道在子进程中可以使用，但是在execl()执行的程序中不可使用
  if (fcntl(ngx_processes[s].channel[1], F_SETFD, FD_CLOEXEC) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "fcntl(FD_CLOEXEC) failed while spawning \"%s\"",
          name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;
  }

  // ngx_processes[s].channel[1]是用于给子进程监听相关事件使用的，当父进程向
  // ngx_processes[s].channel[0]发布事件之后，ngx_processes[s].channel[1]中就会接收到
  // 对应的事件，从而进行相应的处理
  ngx_channel = ngx_processes[s].channel[1];

 } else {
  // 如果是NGX_PROCESS_DETACHED模式，则表示当前是另外新起的一个master进程，因而将其管道值都置为-1
  ngx_processes[s].channel[0] = -1;
  ngx_processes[s].channel[1] = -1;
 }

 ngx_process_slot = s;


 // fork()方法将产生一个新的进程，这个进程与父进程的关系是子进程的内存数据将完全复制父进程的。
 // 还需要注意的是，fork()出来的子进程执行的代码是从fork()之后开始执行的，而对于父进程而言，
 // 该方法的返回值为父进程id，而对于子进程而言，该方法返回值为0，因而通过if-else语句就可以让父进程
 // 和子进程分别调用后续不同的代码片段
 pid = fork();

 switch (pid) {

  case -1:
   // fork出错
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "fork() failed while spawning \"%s\"", name);
   ngx_close_channel(ngx_processes[s].channel, cycle->log);
   return NGX_INVALID_PID;

  case 0:
   // 子进程执行的分支，这里的proc()方法是外部传进来的，也就是说，当前方法只是创建一个新的进程，
   // 具体的进程处理逻辑，将交由外部代码块进行定义ngx_getpid()方法获取的就是当前新创建的子进程的进程id
   ngx_pid = ngx_getpid();
   proc(cycle, data);
   break;

  default:
   // 父进程会走到这里
   break;
 }

 ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start %s %P", name, pid);

 // 父进程会走到这里，当前的pid是fork()之后父进程得到的新创建的子进程的pid
 ngx_processes[s].pid = pid;
 ngx_processes[s].exited = 0;

 if (respawn >= 0) {
  return pid;
 }

 // 设置当前进程的各个属性，并且存储到ngx_processes数组中的对应位置
 ngx_processes[s].proc = proc;
 ngx_processes[s].data = data;
 ngx_processes[s].name = name;
 ngx_processes[s].exiting = 0;

 switch (respawn) {

  case NGX_PROCESS_NORESPAWN:
   ngx_processes[s].respawn = 0;
   ngx_processes[s].just_spawn = 0;
   ngx_processes[s].detached = 0;
   break;

  case NGX_PROCESS_JUST_SPAWN:
   ngx_processes[s].respawn = 0;
   ngx_processes[s].just_spawn = 1;
   ngx_processes[s].detached = 0;
   break;

  case NGX_PROCESS_RESPAWN:
   ngx_processes[s].respawn = 1;
   ngx_processes[s].just_spawn = 0;
   ngx_processes[s].detached = 0;
   break;

  case NGX_PROCESS_JUST_RESPAWN:
   ngx_processes[s].respawn = 1;
   ngx_processes[s].just_spawn = 1;
   ngx_processes[s].detached = 0;
   break;

  case NGX_PROCESS_DETACHED:
   ngx_processes[s].respawn = 0;
   ngx_processes[s].just_spawn = 0;
   ngx_processes[s].detached = 1;
   break;
 }

 if (s == ngx_last_process) {
  ngx_last_process++;
 }

 return pid;
}

/**
 * 这里主要是对当前进程进行初始化，为其设置优先级和打开的文件限制等参数。
 * 最后会为当前进程添加一个监听channel[1]的连接，以不断读取master进程的消息，从而进行相应的处理
 */
static void ngx_worker_process_init(ngx_cycle_t *cycle, ngx_int_t worker) {
 sigset_t set;
 ngx_int_t n;
 ngx_time_t *tp;
 ngx_uint_t i;
 ngx_cpuset_t *cpu_affinity;
 struct rlimit rlmt;
 ngx_core_conf_t *ccf;
 ngx_listening_t *ls;

 // 设置时区相关的信息
 if (ngx_set_environment(cycle, NULL) == NULL) {
  /* fatal */
  exit(2);
 }

 ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);

 // 设置当前进程的优先级
 if (worker >= 0 && ccf->priority != 0) {
  if (setpriority(PRIO_PROCESS, 0, ccf->priority) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "setpriority(%d) failed", ccf->priority);
  }
 }

 // 设置当前进程能够打开的文件句柄数
 if (ccf->rlimit_nofile != NGX_CONF_UNSET) {
  rlmt.rlim_cur = (rlim_t) ccf->rlimit_nofile;
  rlmt.rlim_max = (rlim_t) ccf->rlimit_nofile;

  if (setrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "setrlimit(RLIMIT_NOFILE, %i) failed",
          ccf->rlimit_nofile);
  }
 }

 // Changes the limit on the largest size of a core file(RLIMIT_CORE) for worker processes.
 // 简而言之就是设置核心文件能够使用的最大大小
 if (ccf->rlimit_core != NGX_CONF_UNSET) {
  rlmt.rlim_cur = (rlim_t) ccf->rlimit_core;
  rlmt.rlim_max = (rlim_t) ccf->rlimit_core;

  if (setrlimit(RLIMIT_CORE, &rlmt) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "setrlimit(RLIMIT_CORE, %O) failed",
          ccf->rlimit_core);
  }
 }

 // geteuid()返回执行当前程序的用户id，这里的0表示是否为root用户
 if (geteuid() == 0) {
  // setgid()方法的作用是更改组的id
  if (setgid(ccf->group) == -1) {
   ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
          "setgid(%d) failed", ccf->group);
   /* fatal */
   exit(2);
  }

  // initgroups()是更改附加组的id
  if (initgroups(ccf->username, ccf->group) == -1) {
   ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
          "initgroups(%s, %d) failed",
          ccf->username, ccf->group);
  }

  // 更改用户的id
  if (setuid(ccf->user) == -1) {
   ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
          "setuid(%d) failed", ccf->user);
   /* fatal */
   exit(2);
  }
 }

 // 需要注意的是，对于cache manager和cache loader进程，这里的worker传入的是-1，
 // 表示这两个进程不需要设置亲核性
 if (worker >= 0) {
  // 获取当前worker的CPU亲核性
  cpu_affinity = ngx_get_cpu_affinity(worker);

  if (cpu_affinity) {
   // 设置worker的亲核心
   ngx_setaffinity(cpu_affinity, cycle->log);
  }
 }

#if (NGX_HAVE_PR_SET_DUMPABLE)
 if (prctl(PR_SET_DUMPABLE, 1, 0, 0, 0) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "prctl(PR_SET_DUMPABLE) failed");
 }

#endif

 if (ccf->working_directory.len) {
  // chdir()的作用是将当前的工作目录更改为其参数所传入的路径
  if (chdir((char *) ccf->working_directory.data) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "chdir(\"%s\") failed", ccf->working_directory.data);
   /* fatal */
   exit(2);
  }
 }

 // 初始化空的set指令集合
 sigemptyset(&set);

 // ◆ SIG_BLOCK：将 set 参数指向信号集中的信号加入到信号掩码中。
 // ◆ SIG_UNBLOCK：将 set 参数指向的信号集中的信号从信号掩码中删除。
 // ◆ SIG_SETMASK：将 set 参数指向信号集设置为信号掩码。
 // 这里就是直接初始化要阻塞的信号集，默认为空集
 if (sigprocmask(SIG_SETMASK, &set, NULL) == -1) {
  ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
         "sigprocmask() failed");
 }

 tp = ngx_timeofday();
 srandom(((unsigned) ngx_pid << 16) ^ tp->sec ^ tp->msec);

 ls = cycle->listening.elts;
 for (i = 0; i < cycle->listening.nelts; i++) {
  ls[i].previous = NULL;
 }

 // 这里调用各个模块的init_process()方法进行进程模块的初始化
 for (i = 0; cycle->modules[i]; i++) {
  if (cycle->modules[i]->init_process) {
   if (cycle->modules[i]->init_process(cycle) == NGX_ERROR) {
    /* fatal */
    exit(2);
   }
  }
 }

 // 这里主要是关闭当前进程中其他各个进程的channel[1]管道句柄
 for (n = 0; n < ngx_last_process; n++) {

  if (ngx_processes[n].pid == -1) {
   continue;
  }

  if (n == ngx_process_slot) {
   continue;
  }

  if (ngx_processes[n].channel[1] == -1) {
   continue;
  }

  if (close(ngx_processes[n].channel[1]) == -1) {
   ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
          "close() channel failed");
  }
 }

 // 关闭当前进程的channel[0]管道句柄
 if (close(ngx_processes[ngx_process_slot].channel[0]) == -1) {
  ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
         "close() channel failed");
 }

#if 0
 ngx_last_process = 0;
#endif

 // ngx_channel指向的是当前进程的channel[1]句柄，也即监听master进程发送消息的句柄。
 // 当前方法中，首先会为当前的句柄创建一个connection对象，并且将其封装为一个事件，然后将该事件添加到
 // 对应的事件模型队列中以监听当前句柄的事件，事件的处理逻辑则主要有这里的ngx_channel_handler()
 // 方法进行。这里的ngx_channel_handler的主要处理逻辑是，根据当前收到的消息设置当前进程的一些标志位，
 // 或者更新某些缓存数据，如此，在当前进行的事件循环中，通过不断检查这些标志位，从而实现在事件进程中
 // 处理真正的逻辑。因而这里的ngx_channel_handler的处理效率是非常高的
 if (ngx_add_channel_event(cycle, ngx_channel, NGX_READ_EVENT,
              ngx_channel_handler)
   == NGX_ERROR) {
  /* fatal */
  exit(2);
 }
}

static void ngx_channel_handler(ngx_event_t *ev) {
 ngx_int_t n;
 ngx_channel_t ch;
 ngx_connection_t *c;

 if (ev->timedout) {
  ev->timedout = 0;
  return;
 }

 c = ev->data;

 for (;;) {

  // 在无限for循环中不断读取master进程发过来的消息
  n = ngx_read_channel(c->fd, &ch, sizeof(ngx_channel_t), ev->log);

  // 如果读取消息出错，说明当前的句柄可能失效了，就需要关闭当前连接
  if (n == NGX_ERROR) {
   if (ngx_event_flags & NGX_USE_EPOLL_EVENT) {
    ngx_del_conn(c, 0);
   }

   ngx_close_connection(c);
   return;
  }

  if (ngx_event_flags & NGX_USE_EVENTPORT_EVENT) {
   if (ngx_add_event(ev, NGX_READ_EVENT, 0) == NGX_ERROR) {
    return;
   }
  }

  if (n == NGX_AGAIN) {
   return;
  }

  // 对发送过来的消息进行处理
  switch (ch.command) {
   // 如果是quit消息，则设置quit标志位
   case NGX_CMD_QUIT:
    ngx_quit = 1;
    break;

    // 如果terminate消息，则设置terminate标志位
   case NGX_CMD_TERMINATE:
    ngx_terminate = 1;
    break;

    // 如果是reopen消息，则设置reopen标志位
   case NGX_CMD_REOPEN:
    ngx_reopen = 1;
    break;

    // 如果是新建进程消息，则更新当前ngx_processes数组对应位置的数据
   case NGX_CMD_OPEN_CHANNEL:
    ngx_processes[ch.slot].pid = ch.pid;
    ngx_processes[ch.slot].channel[0] = ch.fd;
    break;

    // 如果是关闭channel的消息，则关闭ngx_processes数组对应位置的句柄
   case NGX_CMD_CLOSE_CHANNEL:
    if (close(ngx_processes[ch.slot].channel[0]) == -1) {
     ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
            "close() channel failed");
    }

    ngx_processes[ch.slot].channel[0] = -1;
    break;
  }
 }
}

typedef struct {
  // 当前发生的事件类型
  ngx_uint_t command;
  // 发生事件的pid
  ngx_pid_t pid;
  // 发生事件的进程在ngx_processes数组中的下标
  ngx_int_t slot;
  // 发生事件的进程的channel[0]描述符的值
  ngx_fd_t fd;
} ngx_channel_t;

/**
 * 进入worker进程工作的循环
 */
static void ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data) {
 ngx_int_t worker = (intptr_t) data;

 ngx_process = NGX_PROCESS_WORKER;
 ngx_worker = worker;

 // 初始化worker进程，前面对该方法的源码进行了讲解
 ngx_worker_process_init(cycle, worker);

 ngx_setproctitle("worker process");

 for (;;) {

  if (ngx_exiting) {
   // 这里主要是检查有没有事件是非cancelable状态的，也就是说是否所有的事件都已经取消了，如果取消了，
   // 就会返回NGX_OK。这里的逻辑可以理解为，如果被标记为了ngx_exiting，那么此时，如果还有未取消的
   // 事件存在，则会走到下面的ngx_process_events_and_timers()方法，如此就会处理未完成的事件，
   // 然后在循环中再次走到这个位置，最终if条件为true，从而执行退出worker进程的工作
   if (ngx_event_no_timers_left() == NGX_OK) {
    ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");
    ngx_worker_process_exit(cycle);
   }
  }

  ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "worker cycle");

  // 这里通过检查相应的事件模型中是否存在对应的事件，然后将其放入队列中进行处理，
  // 这里是worker进程处理事件的核心方法
  ngx_process_events_and_timers(cycle);

  // 这里ngx_terminate是强制关闭nginx的选项，如果向nginx发送了强制关闭nginx命令，则当前进程会直接退出
  if (ngx_terminate) {
   ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");
   ngx_worker_process_exit(cycle);
  }

  // 这里ngx_quit是优雅退出的选项。这里主要是将ngx_exiting置为1，用于表征当前进程需要退出，
  // 然后会执行如下三个工作：
  // 1. 往事件队列中添加一个事件，用于处理当前处于活跃状态的连接，将其close标志位置为1，并且执行该连接
  //  当前的处理方法，以尽快完成连接事件；
  // 2. 关闭当前cycle中监听的socket句柄；
  // 3. 将当前所有处于空闲状态的连接的close状态标记为1，然后调用其连接处理方法.
  if (ngx_quit) {
   ngx_quit = 0;
   ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "gracefully shutting down");
   ngx_setproctitle("worker process is shutting down");

   if (!ngx_exiting) {
    ngx_exiting = 1;
    ngx_set_shutdown_timer(cycle);
    ngx_close_listening_sockets(cycle);
    ngx_close_idle_connections(cycle);
   }
  }

  // ngx_reopen主要是重新打开nginx的所有文件，比如切换nginx的日志文件等等
  if (ngx_reopen) {
   ngx_reopen = 0;
   ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reopening logs");
   ngx_reopen_files(cycle, -1);
  }
 }
}