Implementation of nginx worker process loop

typedef struct {
  // The rest of the properties...
  
  ngx_socket_t channel[2];
} ngx_process_t;

/**
 * Start n worker child processes and set up socketpair between each child process and the master parent process
 * Socket handle communication mechanism established by system call*/
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 means laying eggs, which means generating a child process. The event loop of the child process is // ngx_worker_process_cycle() method. Here ngx_worker_process_cycle is the cycle of the worker process processing events.
  // The worker process is in an infinite for loop, constantly checking whether there is a corresponding event in the corresponding event model.
  // Then separate the accept event and the read and write events into two queues, and finally continuously process the events in the event loop ngx_spawn_process(cycle, ngx_worker_process_cycle, 
           (void *) (intptr_t) i, "worker process", type);

  // The main function of the following code is to notify other processes of the event of the new process. In the above // ​​ch.command = NGX_CMD_OPEN_CHANNEL;, NGX_CMD_OPEN_CHANNEL means that a new process is currently created.
  // ngx_process_slot stores the array location where the new process is stored. The reason why broadcasting is needed here is that
  // After each child process is created, its memory data is copied from the parent process, but each process has a copy of the ngx_processes array.
  // Therefore, the subprocess created first in the array does not have the data of the subprocess created later, but the master process has the data of all subprocesses.
  // Therefore, after the master process creates a child process, it will write the current broadcast event to channel[0] of each process in the ngx_processes array, that is, ch here. In this way, after each child process receives this event,
  // Will try to update the saved ngx_processes data information ch.pid = ngx_processes[ngx_process_slot].pid;
  ch.slot = ngx_process_slot;
  ch.fd = ngx_processes[ngx_process_slot].channel[0];

  // Broadcast event 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 {
  // All currently created processes are stored in the ngx_processes array, and ngx_last_process is the index of the next position of the last // process currently recorded in ngx_processes, but some of the processes recorded in ngx_processes may have // ​​expired. The current loop starts from the beginning to find out whether a process has failed. If it has failed, the process position is reused.
  // Otherwise, directly use the position pointed to by ngx_last_process for (s = 0; s < ngx_last_process; s++) {
   if (ngx_processes[s].pid == -1) {
    break;
   }
  }

  // This indicates that the number of created processes has reached the maximum 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;
  }
 }

 // The NGX_PROCESS_DETACHED flag indicates that the current forked process has no relationship with the original parent process. For example, when upgrading nginx,
 // The newly generated master process has nothing to do with the original master process if (respawn != NGX_PROCESS_DETACHED) {

  /* Solaris 9 still has no AF_LOCAL */

  // The main function of the socketpair() method here is to generate a pair of socket streams for communication between the main process and the child process. This pair of sockets will be // stored in ngx_processes[s].channel. Essentially, this field is an integer array of length 2. Before the main process and the child process // communicate, the main process will close one of them, and the child process will close the other, and then they can communicate by writing or reading data from the other unclosed file descriptor.
  // AF_UNIX indicates that the socket address family in the form of a UNIX file is currently used. // SOCK_STREAM specifies that the communication method established by the current socket is a pipe stream, and this pipe stream is bidirectional.
  // Both sides of the pipe can perform read and write operations // The third parameter protocol must be 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]);

  // Set ngx_processes[s].channel[0] to non-blocking mode 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;
  }

  // Set ngx_processes[s].channel[1] to non-blocking mode 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;
  // Set the ngx_processes[s].channel[0] socket pipe to asynchronous mode 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;
  }

  // Currently in the main process, ngx_pid here points to the process id of the main process. The main function of this method is to set the operation permission of // ngx_processes[s].channel[0] to the main process, that is, the main process communicates with the child process by writing and reading data to // 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 indicates that the currently specified socket pipe can be used in the child process, but cannot be used in the program executed by 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 indicates that the currently specified socket pipe can be used in the child process, but cannot be used in the program executed by 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] is used to listen to related events for the child process. When the parent process publishes an event to // ngx_processes[s].channel[0], ngx_processes[s].channel[1] will receive the // corresponding event and perform corresponding processing ngx_channel = ngx_processes[s].channel[1];

 } else {
  // If it is NGX_PROCESS_DETACHED mode, it means that the current process is a new master process, so its pipeline value is set to -1
  ngx_processes[s].channel[0] = -1;
  ngx_processes[s].channel[1] = -1;
 }

 ngx_process_slot = s;


 // The fork() method will generate a new process. The relationship between this process and the parent process is that the memory data of the child process will completely copy the parent process.
 // It should also be noted that the code executed by the child process from fork() starts from after fork(), while for the parent process,
 // The return value of this method is the parent process id, while for the child process, the return value of this method is 0, so through the if-else statement, the parent process // and the child process can call different subsequent code snippets respectively pid = fork();

 switch (pid) {

  Case -1:
   // fork error 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:
   // The branch of child process execution. The proc() method here is passed in from the outside. That is to say, the current method just creates a new process.
   // The specific process processing logic will be defined by the external code block. The ngx_getpid() method obtains the process id of the newly created child process.
   ngx_pid = ngx_getpid();
   proc(cycle, data);
   break;

  default:
   //The parent process will go here break;
 }

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

 // The parent process will come here, the current pid is the pid of the newly created child process obtained by the parent process after fork()
 ngx_processes[s].pid = pid;
 ngx_processes[s].exited = 0;

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

 // Set the various attributes of the current process and store them in the corresponding position in the ngx_processes array 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;
}

/**
 * Here we mainly initialize the current process and set parameters such as priority and open file limit for it.
 * Finally, a connection to listen to channel[1] will be added to the current process to continuously read messages from the master process and perform corresponding processing*/
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;

 // Set time zone related information if (ngx_set_environment(cycle, NULL) == NULL) {
  /* fatal */
  exit(2);
 }

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

 // Set the priority of the current process 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);
  }
 }

 // Set the number of file handles that the current process can open 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.
 // In short, it sets the maximum size that the core file can use 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() returns the user ID that executes the current program. Here, 0 indicates whether it is the root user if (geteuid() == 0) {
  // The purpose of the setgid() method is to change the group 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() is to change the id of the additional group
  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);
  }

  // Change the user's id
  if (setuid(ccf->user) == -1) {
   ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
          "setuid(%d) failed", ccf->user);
   /* fatal */
   exit(2);
  }
 }

 // Note that for the cache manager and cache loader processes, the worker here passes in -1.
 // Indicates that these two processes do not need to set nucleophilicity if (worker >= 0) {
  // Get the CPU affinity of the current worker cpu_affinity = ngx_get_cpu_affinity(worker);

  if (cpu_affinity) {
   // Set the worker's affinity core 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) {
  // The function of chdir() is to change the current working directory to the path passed in as its parameter 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);
  }
 }

 // Initialize the empty set instruction set sigemptyset(&set);

 // ◆ SIG_BLOCK: Add the signal in the signal set pointed to by the set parameter to the signal mask.
 // ◆ SIG_UNBLOCK: Delete the signal in the signal set pointed to by the set parameter from the signal mask.
 // ◆ SIG_SETMASK: Set the signal set pointed to by the set parameter to the signal mask.
 // Here is to directly initialize the signal set to be blocked, the default is an empty 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;
 }

 // Here, the init_process() method of each module is called to initialize the process module 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);
   }
  }
 }

 // This is mainly to close the channel[1] pipe handles of other processes in the current process 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");
  }
 }

 // Close the channel[0] pipe handle of the current process 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 points to the channel[1] handle of the current process, which is also the handle that listens to messages sent by the master process.
 // In the current method, a connection object is first created for the current handle, and it is encapsulated as an event. Then the event is added to the // corresponding event model queue to listen to the events of the current handle. The event processing logic is mainly here ngx_channel_handler()
 // Method proceeds. The main processing logic of ngx_channel_handler here is to set some flags of the current process according to the currently received message.
 // Or update some cached data, so that in the current event loop, by constantly checking these flags, // the real logic is processed in the event process. Therefore, the processing efficiency of ngx_channel_handler here is very high 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 (;;) {

  //Continuously read the messages sent by the master process in an infinite for loop n = ngx_read_channel(c->fd, &ch, sizeof(ngx_channel_t), ev->log);

  // If there is an error in reading the message, the current handle may be invalid, and you need to close the current connection 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;
  }

  // Process the sent message switch (ch.command) {
   // If it is a quit message, set the quit flag case NGX_CMD_QUIT:
    ngx_quit = 1;
    break;

    // If the message is terminated, set the terminate flag case NGX_CMD_TERMINATE:
    ngx_terminate = 1;
    break;

    // If it is a reopen message, set the reopen flag case NGX_CMD_REOPEN:
    ngx_reopen = 1;
    break;

    // If it is a new process message, update the data at the corresponding position of the current ngx_processes array case NGX_CMD_OPEN_CHANNEL:
    ngx_processes[ch.slot].pid = ch.pid;
    ngx_processes[ch.slot].channel[0] = ch.fd;
    break;

    // If it is a message to close the channel, close the handle at the corresponding position in the ngx_processes array 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 {
  // The current event type ngx_uint_t command;
  // The pid of the event
  ngx_pid_t pid;
  // The subscript ngx_int_t slot of the process where the event occurred in the ngx_processes array;
  // The value of the channel[0] descriptor of the process where the event occurred ngx_fd_t fd;
} ngx_channel_t;

/**
 * Enter the worker process working loop*/
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;

 // Initialize the worker process. The source code of this method is explained above. ngx_worker_process_init(cycle, worker);

 ngx_setproctitle("worker process");

 for (;;) {

  if (ngx_exiting) {
   // Here we mainly check whether there are any events in a non-cancelable state, that is, whether all events have been canceled. If so,
   // It will return NGX_OK. The logic here can be understood as follows: if it is marked as ngx_exiting, then at this time, if there are still uncancelled // events, it will go to the following ngx_process_events_and_timers() method, so that the unfinished events will be processed.
   // Then go to this position again in the loop, and finally the if condition is true, so as to execute the work of exiting the worker process 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");

  // Here we check whether there is a corresponding event in the corresponding event model, and then put it into the queue for processing.
  // Here is the core method of the worker process to handle events ngx_process_events_and_timers(cycle);

  // Here ngx_terminate is an option to force nginx to shut down. If a command to force nginx to shut down is sent to nginx, the current process will exit directly if (ngx_terminate) {
   ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");
   ngx_worker_process_exit(cycle);
  }

  // Here ngx_quit is an option for graceful exit. Here, ngx_exiting is set to 1 to indicate that the current process needs to exit.
  // Then the following three tasks will be performed:
  // 1. Add an event to the event queue to handle the currently active connection, set its close flag to 1, and execute the connection // current processing method to complete the connection event as soon as possible;
  // 2. Close the socket handle monitored in the current cycle;
  // 3. Mark the close status of all currently idle connections as 1, and then call their connection processing methods.
  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 mainly reopens all nginx files, such as switching nginx log files, etc. if (ngx_reopen) {
   ngx_reopen = 0;
   ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reopening logs");
   ngx_reopen_files(cycle, -1);
  }
 }
}