nginx进程模型 master/worker

时间:2016-08-08 22:19:57   收藏:0   阅读:2016

nginx有两类进程,一类称为master进程(相当于管理进程),另一类称为worker进程(实际工作进程)。启动方式有两种:

(1)单进程启动:此时系统中仅有一个进程,该进程既充当master进程的角色,也充当worker进程的角色。

(2)多进程启动:此时系统有且仅有一个master进程,至少有一个worker进程工作。

master进程主要进行一些全局性的初始化工作和管理worker的工作;事件处理是在worker中进行的。

     

首先简要的浏览一下nginx的启动过程,如下图:

 

2.实现原理

         这里只分析多进程下的工作原理。

nginx的进程启动过程是在ngx_master_process_cycle(src/os/unix /ngx_process_cycle.c)中完成的,在ngx_master_process_cycle中,会根据配置文件的 worker_processes值创建多个子进程,即一个master进程和多个worker进程。进程之间、进程与外部之间保持通信。如下图所示:图 中w1表示worker进程1,以此类推。虚线表示信号通信,实现表示socketpair通信。

 

nginx 的进程模型采用的是prefork方式,预先分配的worker子进程数量由配置文件指定,默认为1。master主进程创建监听套接口,fork子进程 以后,由worker进程监听客户连接,每个worker子进程独自尝试accept已连接套接口,accept是否上锁可以配置,默认会上锁,如果操作 系统支持原子整型,才会使用共享内存实现原子上锁,否则使用文件上锁。如果不使用锁,当多个进程同时accept,当一个连接来的时候多个进程同时被唤 起,会导致惊群问题。使用锁的时候,只会有一个worker阻塞在accept上,其他的进程则会不能获取锁而阻塞,这样就解决了惊群的问题。 master进程通过socketpair向worker子进程发送命令,终端也可以向master发送各种命令,子进程通过发送信号给master进程 的方式与其通信,worker之间通过unix套接口通信。

当master接收到worker发回的SIGCHLD信号时,(worker进程的退出信号),它会逐个检查每一个worker进程,如果发现有 worker进程是异常退出,就会重新启动这个worker进程。另外nginx还有两个用于管理cache的进程,一个是cache manager process,另外一个是cache loader process,它们是专门服务于文件cache的进程,也服从master进程的管理,类似于worker进程,后面的分析将略去它们。下面从代码的角 度,详细分析实现细节。

 

master启动的时候,有一些重要的全局数据会被设置,最重要的是进程表ngx_processes,master每创建一个worker都会把 一个设置好的ngx_process_t结构变量放入ngx_processes中,新创建的进程存放在ngx_process_slot位 置,ngx_last_process是进程表中最后一个存量进程的下一个位置,ngx_process_t是进程在nginx中的抽象:

 
 1 typedef struct {
 2     ngx_pid_t            pid;                //进程的ID
 3     int                     status;             //进程的退出状态  
 4     ngx_socket_t       channel[2];        //用于socketpair通信的一对socket句柄
 5     ngx_spawn_proc_pt   proc;           //进程的执行函数
 6     void                  *data;              //proc的参数
 7     char                  *name;             //进程的title标识
 8     unsigned             respawn:1;        //进程的状态:重新创建的
 9     unsigned             just_spawn:1;     //进程的状态: 第一次创建的
10     unsigned             detached:1;       //进程的状态: 分离的,独立的
11     unsigned             exiting:1;         //进程的状态: 正在退出的
12     unsigned             exited:1;          //进程的状态: 已经退出的
13 } ngx_process_t;(src/os/unix/ngx_process.h)
 

master进程向worker子进程发送命令是通过socketpair创建的一对socket实现的,之间传输的是ngx_channel_t结构变量:

 
1 typedef struct {
2      ngx_uint_t  command;  //发送的命令
3      ngx_pid_t   pid;         //发送方进程的进程id
4      ngx_int_t   slot;         //发送方进程在进程表中偏移位置
5      ngx_fd_t    fd;           //发送给对方的文件句柄
6 } ngx_channel_t;(src/os/unix/ngx_channel.h)
 

command是要发送的命令,有5种:

 

1 #define NGX_CMD_OPEN_CHANNEL   1
2 #define NGX_CMD_CLOSE_CHANNEL  2
3 #define NGX_CMD_QUIT           3
4 #define NGX_CMD_TERMINATE      4
5 #define NGX_CMD_REOPEN         5

 

1).首先分析master进程的代码的功能,(Ngx_process_cycle.c中):

 

main()函数首先做一系列的初始化工作调用各模块的初始化代码(例如创建监听套接口等)然后就会调用 ngx_master_process_cycle代码(多进程启动情况下),cycle是一个全局结构体变量,存储有系统运行的所需要的一些信息。在分 析进程关系的的时候可以先忽略它。

 

 
  1 void ngx_master_process_cycle(ngx_cycle_t *cycle)
  2 {
  3 ?    1.master设置一些需要处理的信号,信号包括:
  4 SIGCHLD,                               //子进程退出时发送给父进程的
  5 SIGALRM,                               //计时器信号
  6 SIGIO,                                  //描述符上可以进行I/O时发出的信号
  7 SIGINT,                                 //中断信号
  8 NGX_RECONFIGURE_SIGNAL(SIGHUP), //终端线路挂断
  9 NGX_REOPEN_SIGNAL(SIGUSR1),      //用户自定义usr1信号
 10 NGX_NOACCEPT_SIGNAL(SIGWINCH), //控制中断大小改变
 11 NGX_TERMINATE_SIGNAL(SIGTERM),  //请求终端
 12 NGX_SHUTDOWN_SIGNAL(SIGQUIT),  //终端发送的quit信号
 13 NGX_CHANGEBIN_SIGNAL(SIGUSR2);//用户自定义usr1信号
 14 
 15 ?    2.调用ngx_setproctilte设置进程标题;
 16 
 17 ?    3. 调用ngx_start_worker_processes()启动worker进程;
 18 
 19 //有些模块需要文件cache,比如fastcgi模块,这些模块会把文件cache路径添加到//cycle->paths中,文件cache管理进程会定期调用这些模块的文件cache处理钩子处//理一下文件cache,其实一共会启动两个进程,这些进程的detached会被设置为1
 20 ?    4.调用ngx_start_cache_manager_processes()启动文件cache管理进程;
 21 
 22 ?    5.master循环处理信号量。
 23     ngx_new_binary = 0;
 24     delay = 0;
 25     live = 1;
 26 
 27     for ( ;; ) {
 28         // delay用来设置等待worker退出的时间,master接收了退出信号后首先发送                        //退出信号给worker,而worker退出需要一些时间
 29         if (delay) {
 30             delay *= 2;
 31             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
 32                            "temination cycle: %d", delay);
 33             itv.it_interval.tv_sec = 0;
 34             itv.it_interval.tv_usec = 0;
 35             itv.it_value.tv_sec = delay / 1000;
 36             itv.it_value.tv_usec = (delay % 1000 ) * 1000;
 37 
 38             // 设置定时器
 39             if (setitimer(ITIMER_REAL, &itv, NULL) == -1) {
 40                 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 41                               "setitimer() failed");
 42             }
 43         }
 44 
 45         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "sigsuspend");
 46 
 47         // 挂起信号量,等待定时器
 48         sigsuspend(&set);
 49         ngx_time_update(0, 0);
 50         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "wake up");
 51 
 52         // 收到了SIGCHLD信号,有worker退出(ngx_reap==1)
 53         if (ngx_reap) {
 54             ngx_reap = 0;
 55             ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "reap children");
 56 
 57        // 处理所有worker,如果有worker异常退出则重启这个worker,如果所有
 58 //  worker都退出,返回0赋值给live
 59             live = ngx_reap_children(cycle);
 60         }
 61 
 62  // 如果worker都已经退出,并且收到了NGX_CMD_TERMINATE命令或者  //SIGTERM信号或者SIGINT信号(ngx_terminate=1)
 63 // 或者NGX_CMD_QUIT命令或者SIGQUIT信号(ngx_quit=1),则master退出
 64         if (!live && (ngx_terminate || ngx_quit)) {
 65             ngx_master_process_exit(cycle);
 66         }
 67 
 68         // 收到了NGX_CMD_TERMINATE命令或者SIGTERM信号或者SIGINT信号,
 69         // 通知所有worker退出,并且等待worker退出
 70         if (ngx_terminate) {
 71             if (delay == 0) {
 72                 delay = 50;
 73             }
 74 
 75          // 给所有worker发送SIGTERM,通知worker退出
 76             if (delay > 1000) {
 77                 ngx_signal_worker_processes(cycle, SIGKILL);
 78             } else {
 79                
 80                 ngx_signal_worker_processes(cycle,
 81                                        ngx_signal_value(NGX_TERMINATE_SIGNAL));
 82             }
 83 
 84             continue;
 85         }
 86 
 87         // 收到了NGX_CMD_QUIT命令或者SIGQUIT信号
 88         if (ngx_quit) {
 89             // 给所有worker发送SIGQUIT信号
 90             ngx_signal_worker_processes(cycle,
 91                                         ngx_signal_value(NGX_SHUTDOWN_SIGNAL));
 92 
 93             // 关闭所有监听的socket
 94             ls = cycle->listening.elts;
 95             for (n = 0; n < cycle->listening.nelts; n++) {
 96                 if (ngx_close_socket(ls[n].fd) == -1) {
 97                     ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_socket_errno,
 98                                   ngx_close_socket_n " %V failed",
 99                                   &ls[n].addr_text);
100                 }
101             }
102             cycle->listening.nelts = 0;
103 
104             continue;
105         }
106 
107         // 收到了SIGHUP信号
108         if (ngx_reconfigure) {
109             ngx_reconfigure = 0;
110 
111             // 代码已经被替换,重启worker,不需要重新初始化配置
112             if (ngx_new_binary) {
113                 ngx_start_worker_processes(cycle, ccf->worker_processes,
114                                            NGX_PROCESS_RESPAWN);
115                 ngx_start_cache_manager_processes(cycle, 0);
116                 ngx_noaccepting = 0;
117 
118                 continue;
119             }
120 
121             ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reconfiguring");
122 
123             // 重新初始化配置
124             cycle = ngx_init_cycle(cycle);
125             if (cycle == NULL) {
126                 cycle = (ngx_cycle_t *) ngx_cycle;
127                 continue;
128             }
129 
130             // 重启worker
131             ngx_cycle = cycle;
132             ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx,
133                                                    ngx_core_module);
134             ngx_start_worker_processes(cycle, ccf->worker_processes,
135                                        NGX_PROCESS_JUST_RESPAWN);
136             ngx_start_cache_manager_processes(cycle, 1);
137             live = 1;
138             ngx_signal_worker_processes(cycle,
139                                         ngx_signal_value(NGX_SHUTDOWN_SIGNAL));
140         }
141 
142         // 当ngx_noaccepting=1的时候会把ngx_restart设为1,重启worker
143         if (ngx_restart) {
144             ngx_restart = 0;
145             ngx_start_worker_processes(cycle, ccf->worker_processes,
146                                        NGX_PROCESS_RESPAWN);
147             ngx_start_cache_manager_processes(cycle, 0);
148             live = 1;
149         }
150 
151         // 收到SIGUSR1信号,重新打开log文件
152         if (ngx_reopen) {
153             ngx_reopen = 0;
154             ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reopening logs");
155             ngx_reopen_files(cycle, ccf->user);
156             ngx_signal_worker_processes(cycle,
157                                         ngx_signal_value(NGX_REOPEN_SIGNAL));
158         }
159 
160         // 收到SIGUSR2信号,热代码替换
161         if (ngx_change_binary) {
162             ngx_change_binary = 0;
163             ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "changing binary");
164             // 调用execve执行新的代码
165             ngx_new_binary = ngx_exec_new_binary(cycle, ngx_argv);
166         }
167 
168         // 收到SIGWINCH信号,不再接收请求,worker退出,master不退出
169         if (ngx_noaccept) {
170             ngx_noaccept = 0;
171             ngx_noaccepting = 1;
172             ngx_signal_worker_processes(cycle,
173                                         ngx_signal_value(NGX_SHUTDOWN_SIGNAL));
174         }
175     }
176 }
 

 

从代码中可以看书master主进程的逻辑是非常清晰的,如下图:

 

2)接下来分析worker进程启动的代码ngx_start_worker_processes(),由于使用了socketpair通信,这里也包过了对socket设置的一些代码:

 

 

 
 1 static void ngx_start_worker_processes(ngx_cycle_t *cycle, ngx_int_t n, ngx_int_t type)
 2 {
 3     ngx_int_t      i;
 4     ngx_channel_t  ch;//用于socketpair通信的数据
 5     ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start worker processes");
 6         // 传递给其他worker子进程的命令:打开通信管道
 7     ch.command = NGX_CMD_OPEN_CHANNEL;
 8 for (i = 0; i < n; i++) {
 9         cpu_affinity = ngx_get_cpu_affinity(i);
10         ngx_spawn_process(cycle, ngx_worker_process_cycle, NULL,
11                           "worker process", type);
12            // 向之前已经创建的所有worker广播当前创建的worker进程的信息
13         ch.pid = ngx_processes[ngx_process_slot].pid;//当前进程的pid
14         ch.slot = ngx_process_slot;               //当前进程在进程表中的位置
15 //fd是发送给对方的句柄
16 ch.fd = ngx_processes[ngx_process_slot].channel[0];                ngx_pass_open_channel(cycle, &ch);
17     }
18 }
 

 

循环体中使用ngx_spawn_process来生成worker进程,这个后面说明。每次创建一个新的worker进程之后,都需要向之前创建的所有worker进程广播新创建的worker进程的信息。

ngx_pass_open_channel()会利用一个循环,将ch信息发送给其他的worker进程的channel[0]的socket 上,worker收到以后就会将ch的信息添加到自己的进程表中,这样每个worker进程自己的进程表和master进程的进程表就会保持一致。在子进 程创建的过程中,后面会有代码来设置各自的进程表项的ngx_socket_t字段。

 

3).第2个函数中新建了一个进程以后,然后调用ngx_pass_open_channel(cycle,&ch)将ch数据对其他进程进行广播处理,下面分析它的实现。

 

 
 1 static void ngx_pass_open_channel(ngx_cycle_t *cycle, ngx_channel_t *ch)
 2 {   //ch是要向其他的worker进程广播的消息
 3     ngx_int_t  i;
 4         //逐个遍历所有的worker进程关联的ngx_process
 5     for (i = 0; i < ngx_last_process; i++) {
 6             // 跳过自己和异常的进程
 7         if (i == ngx_process_slot|| ngx_processes[i].pid == -1
 8             || ngx_processes[i].channel[0] == -1)
 9         {
10             continue;
11         }
12         ngx_log_debug6(NGX_LOG_DEBUG_CORE, cycle->log, 0,
13                       "pass channel s:%d pid:%P fd:%d to s:%i pid:%P fd:%d",
14                       ch->slot, ch->pid, ch->fd,
15                       i, ngx_processes[i].pid,
16                       ngx_processes[i].channel[0]);
17         /* TODO: NGX_AGAIN */
18            // 发送消息给其他的worker,发送到每个进程的
19 //ngx_processes[i].channel[0]的socket上
20         ngx_write_channel(ngx_processes[i].channel[0],
21                           ch, sizeof(ngx_channel_t), cycle->log);
22     }
23 }
 

 

从代码中可以看出函数发送给除自己外而且正常工作的worker进程发送自己的进程信息,worker进程收到以后会将它添加到自己的进程表中。

 

4).接下来分析ngx_pid_t ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data, char *name, ngx_int_t respawn)函

 
  1 ngx_pid_t ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data,char *name, ngx_int_t respawn)
  2 {   //proc是子进程的执行函数,data是其参数,name是子进程的名字
  3     u_long     on;
  4     ngx_pid_t  pid;
  5     ngx_int_t  s;   //将要创建的子进程在进程表中的位置
  6 
  7     if (respawn >= 0) {
  8             // 替换进程ngx_processes[respawn],可安全重用该进程表项
  9         s = respawn;
 10     } else {
 11           // 先找到一个被回收的进程表项
 12         for (s = 0; s < ngx_last_process; s++) {
 13             if (ngx_processes[s].pid == -1) {
 14                 break;
 15             }
 16         }
 17             // 进程表已满
 18         if (s == NGX_MAX_PROCESSES) {
 19             ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
 20                           "no more than %d processes can be spawned",
 21                           NGX_MAX_PROCESSES);
 22             return NGX_INVALID_PID;
 23         }
 24     }
 25 
 26 
 27         
 28         // 不是分离的子进程,指的就是worker进程,cache进程是分离出去的进程
 29         //cache进程不等同于worker进程,系统把它看成额外的独立进程
 30     if (respawn != NGX_PROCESS_DETACHED) {
 31         /* Solaris 9 still has no AF_LOCAL */
 32             // 创建一对已经连接的无名socket,用于socketpair通信的
 33         if (socketpair(AF_UNIX, SOCK_STREAM, 0, ngx_processes[s].channel) == -1)
 34         {
 35             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 36                           "socketpair() failed while spawning \"%s\"", name);
 37             return NGX_INVALID_PID;
 38         }
 39         ngx_log_debug2(NGX_LOG_DEBUG_CORE, cycle->log, 0,
 40                        "channel %d:%d",
 41                        ngx_processes[s].channel[0],
 42                        ngx_processes[s].channel[1]);
 43     
 44             // 设置socket为非阻塞模式
 45         if (ngx_nonblocking(ngx_processes[s].channel[0]) == -1) {
 46             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 47                           ngx_nonblocking_n " failed while spawning \"%s\"",
 48                           name);
 49             ngx_close_channel(ngx_processes[s].channel, cycle->log);
 50             return NGX_INVALID_PID;
 51         }
 52 // 设置socket为非阻塞模式
 53         if (ngx_nonblocking(ngx_processes[s].channel[1]) == -1) {
 54             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 55                           ngx_nonblocking_n " failed while spawning \"%s\"",
 56                           name);
 57             ngx_close_channel(ngx_processes[s].channel, cycle->log);
 58             return NGX_INVALID_PID;
 59         }
 60 
 61             // 设置channel[0]的信号驱动异步I/O 标志
 62         on = 1;
 63         if (ioctl(ngx_processes[s].channel[0], FIOASYNC, &on) == -1) {
 64             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 65                           "ioctl(FIOASYNC) failed while spawning \"%s\"", name);
 66             ngx_close_channel(ngx_processes[s].channel, cycle->log);
 67             return NGX_INVALID_PID;
 68         }
 69 
 70         
 71         // 设置channel[0]的属主,控制channel[0]的SIGIO信号只发给master进程,//ngx_pid为全局变量,指的是master主进程,因为master与worker进程通//信时通过将数据发送到进程的channel[0]上通知的,因此channel[0]的属主是//manster进程
 72                 if (fcntl(ngx_processes[s].channel[0], F_SETOWN, ngx_pid) == -1) {
 73             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 74                          "fcntl(F_SETOWN) failed while spawning \"%s\"", name);
 75             ngx_close_channel(ngx_processes[s].channel, cycle->log);
 76             return NGX_INVALID_PID;
 77         }
 78 
 79             // 设置channel[0]的close-on-exec标识,失败则关闭channel
 80         if (fcntl(ngx_processes[s].channel[0], F_SETFD, FD_CLOEXEC) == -1) {
 81             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 82                     "fcntl(FD_CLOEXEC) failed while spawning \"%s\"",name);
 83             ngx_close_channel(ngx_processes[s].channel, cycle->log);
 84             return NGX_INVALID_PID;
 85         }
 86 
 87 // 设置channel[1]的close-on-exec标识,失败则关闭channel
 88         if (fcntl(ngx_processes[s].channel[1], F_SETFD, FD_CLOEXEC) == -1) {
 89             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
 90                   "fcntl(FD_CLOEXEC) failed while spawning \"%s\"",name);
 91             ngx_close_channel(ngx_processes[s].channel, cycle->log);
 92             return NGX_INVALID_PID;
 93         }
 94 
 95             // 用于监听可读事件的socket,ngx_channel是全局变量
 96         ngx_channel = ngx_processes[s].channel[1];
 97     } else {
 98        //说明是分离的独立进程,则不需要socket进行通信,都设置成无效的
 99         ngx_processes[s].channel[0] = -1;
100         ngx_processes[s].channel[1] = -1;
101     }
102 
103         //新建进程在进程表中的实际位置
104     ngx_process_slot = s;
105     pid = fork();
106     switch (pid) {
107     case -1:
108         ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
109                       "fork() failed while spawning \"%s\"", name);
110         ngx_close_channel(ngx_processes[s].channel, cycle->log);
111         return NGX_INVALID_PID;
112 
113     case 0:
114             //子进程在这里运行
115         ngx_pid = ngx_getpid();
116         //调用子进程需要执行的函数,即ngx_worker_process_cycle
117             proc(cycle, data);
118         break;
119     default:
120         break;
121     }
122     ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start %s %P", name, pid);
123     ngx_processes[s].pid = pid;//设置子进程的pid
124     ngx_processes[s].exited = 0;//子进程没有退出
125 
126         // 如果替换进程ngx_processes[respawn],不用设置其他进程表项字段了
127     if (respawn >= 0) {
128         return pid;
129     }
130 
131         // 设置其他的进程表项字段
132     ngx_processes[s].proc = proc;
133     ngx_processes[s].data = data;
134     ngx_processes[s].name = name;
135     ngx_processes[s].exiting = 0;
136 
137         // 设置进程表项的一些状态字段
138     switch (respawn) {
139     case NGX_PROCESS_NORESPAWN:
140         ngx_processes[s].respawn = 0;
141         ngx_processes[s].just_spawn = 0;
142         ngx_processes[s].detached = 0;
143         break;
144 
145     case NGX_PROCESS_JUST_SPAWN: 
146         ngx_processes[s].respawn = 0;
147         ngx_processes[s].just_spawn = 1;
148         ngx_processes[s].detached = 0;
149         break;
150 
151     case NGX_PROCESS_RESPAWN: 
152         ngx_processes[s].respawn = 1;
153         ngx_processes[s].just_spawn = 0;
154         ngx_processes[s].detached = 0;
155         break;
156 
157     case NGX_PROCESS_JUST_RESPAWN:
158         ngx_processes[s].respawn = 1;
159         ngx_processes[s].just_spawn = 1;
160         ngx_processes[s].detached = 0;
161         break;
162 
163     case NGX_PROCESS_DETACHED://
164         ngx_processes[s].respawn = 0;
165         ngx_processes[s].just_spawn = 0;
166         ngx_processes[s].detached = 1;
167         break;
168     }
169     //检查是否需要更新ngx_last_process
170     if (s == ngx_last_process) {
171         ngx_last_process++;
172     }
173     return pid;
174 }
 

 

5)下面分析worker工作进程执行的函数:static voidngx_worker_process_cycle(ngx_cycle_t *cycle, void *data)。

 

 
  1 static void ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data)
  2 {
  3     ngx_uint_t         i;
  4     ngx_connection_t  *c;//用于连接的
  5     ngx_process = NGX_PROCESS_WORKER;
  6 
  7         //初始化工作,这个工作很重要,后面会详细说明它
  8     ngx_worker_process_init(cycle, 1);
  9     //设置进程标题
 10     ngx_setproctitle("worker process");
 11 
 12     //使用线程时,则会执行的代码,线程相关的代码不影响分析系统的进程结构,主//要是做一些创建线程之前的初始化和准备,然后创建线程,执行线程函数,处理//用户请求
 13 #if (NGX_THREADS)
 14     {
 15     ngx_int_t         n;
 16     ngx_err_t         err;
 17     ngx_core_conf_t  *ccf;
 18     ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, ngx_core_module);
 19     if (ngx_threads_n) 
 20         if (ngx_init_threads(ngx_threads_n, ccf->thread_stack_size, cycle)
 21             == NGX_ERROR)
 22         {
 23             /* fatal */
 24             exit(2);
 25         }
 26 
 27         err = ngx_thread_key_create(&ngx_core_tls_key);
 28         if (err != 0) {
 29             ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
 30                           ngx_thread_key_create_n " failed");
 31             /* fatal */
 32             exit(2);
 33         }
 34 
 35         for (n = 0; n < ngx_threads_n; n++) {
 36             ngx_threads[n].cv = ngx_cond_init(cycle->log);
 37             if (ngx_threads[n].cv == NULL) {
 38                 /* fatal */
 39                 exit(2);
 40             }
 41             if (ngx_create_thread((ngx_tid_t *) &ngx_threads[n].tid,
 42                                   ngx_worker_thread_cycle,
 43                                   (void *) &ngx_threads[n], cycle->log)
 44                 != 0)
 45             {
 46                 /* fatal */
 47                 exit(2);
 48             }
 49         }
 50     }
 51     }
 52 #endif
 53 
 54 //worker进程工作的主循环
 55     for ( ;; ) {     
 56             // 如果退出状态已设置,关闭所有连接
 57         if (ngx_exiting) {
 58             c = cycle->connections;
 59             for (i = 0; i < cycle->connection_n; i++) {
 60                 /* THREAD: lock */
 61 //链接存在,而且连接是空闲的,就将它关闭
 62                 if (c[i].fd != -1 && c[i].idle) {                    
 63 c[i].close = 1;
 64                     c[i].read->handler(c[i].read);
 65                 }
 66             }
 67 
 68             if (ngx_event_timer_rbtree.root == ngx_event_timer_rbtree.sentinel)
 69             {
 70                 ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");
 71                 ngx_worker_process_exit(cycle);
 72             }
 73         }
 74 
 75         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "worker cycle");
 76         
 77             // 处理事件和计时
 78         ngx_process_events_and_timers(cycle);
 79 
 80             // 收到NGX_CMD_TERMINATE命令
 81         if (ngx_terminate) {
 82             ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "exiting");
 83                 // 清理后进程退出,会调用所有模块的钩子exit_process
 84             ngx_worker_process_exit(cycle);
 85         }
 86 
 87             // 收到NGX_CMD_QUIT命令
 88         if (ngx_quit) {
 89             ngx_quit = 0;
 90             ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0,
 91                           "gracefully shutting down");
 92             ngx_setproctitle("worker process is shutting down");
 93             if (!ngx_exiting) {
 94                 // 关闭监听socket,设置退出状态
 95                 ngx_close_listening_sockets(cycle);
 96                 ngx_exiting = 1;
 97             }
 98         }
 99 
100             // 收到NGX_CMD_REOPEN命令,重新打开log
101         if (ngx_reopen) {
102             ngx_reopen = 0;
103             ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "reopening logs");
104             ngx_reopen_files(cycle, -1);
105         }
106     }
107 }
 

 

6).接下来分析static void ngx_worker_process_init(ngx_cycle_t *cycle, ngx_uint_t priority),主要做的是work进程创建之前的初始化操作。

 
 1 static void ngx_worker_process_init(ngx_cycle_t *cycle, ngx_uint_t priority)
 2 {
 3 ?    1、设置ngx_process = NGX_PROCESS_WORKER,在master进程中这个变量被设置为NGX_PROCESS_MASTER;
 4 
 5 ?    2、全局性的设置,根据全局的配置信息设置执行环境、优先级、限制、setgid、setuid、信号初始化等;
 6 
 7 ?    3、调用所有模块的钩子init_process;
 8 
 9 ?    4、关闭不使用的socket,关闭当前worker的channel[0]句柄和其他worker的channel[1]句柄,当前worker会使用其他worker的channel[0]句柄发送消息,使用当前worker的channel[1]句柄监听可读事件:
10 
11     for (n = 0; n < ngx_last_process; n++) {
12       //跳过无效进程
13         if (ngx_processes[n].pid == -1) {
14             continue;
15         }    
16       //跳过自己
17         if (n == ngx_process_slot) {
18             continue;
19         }    
20       //跳过独立的进程,因为独立的进程的channel句柄别设置为-1,
21 //或者是关闭channel的进程
22 
23         if (ngx_processes[n].channel[1] == -1) {
24             continue;
25         }    
26 //关闭其他进程的channel[1]句柄
27         if (close(ngx_processes[n].channel[1]) == -1) {
28             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
29                           "close() channel failed");
30         }    
31     }    
32 //关闭自己进程的channel[0]句柄
33     if (close(ngx_processes[ngx_process_slot].channel[0]) == -1) {
34         ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
35                       "close() channel failed");
36     }    
37 
38 ?    5、在当前worker的channel[1]句柄监听可读事件:
39     if (ngx_add_channel_event(cycle, ngx_channel, NGX_READ_EVENT,
40                               ngx_channel_handler)
41         == NGX_ERROR)
42     {
43         exit(2);
44     }
45 }
 

可以看出,通过第4步的操作,worker进程就可以再channel[1]上监听事件了,而master进程正好是将命令发往worker进程对 应的channel[0]上,因此便实现了socketpair通信。当前worker还可以使用其他进程的channel[0]句柄发送消息,使用很 少,但主要是监听channel[1]句柄上的事件消息。

 

7) ngx_add_channel_event()把句柄ngx_channel(当前worker的channel[1])上建立的连接的可读事件加入事 件监控队列,事件处理函数为ngx_channel_hanlder(ngx_event_t *ev)。当有可读事件的时候,ngx_channel_handler负责处理消息,下面分析其实现:

 

 
 1 static voidngx_channel_handler(ngx_event_t *ev)
 2 {
 3     ngx_int_t          n;
 4     ngx_channel_t      ch;
 5     ngx_connection_t  *c;
 6 
 7     if (ev->timedout) {
 8         ev->timedout = 0;
 9         return;
10     }
11 
12     c = ev->data;
13 
14     ngx_log_debug0(NGX_LOG_DEBUG_CORE, ev->log, 0, "channel handler");
15 
16     for ( ;; ) {
17             //从channel[1]中读取消息
18         n = ngx_read_channel(c->fd, &ch, sizeof(ngx_channel_t), ev->log);
19         ngx_log_debug1(NGX_LOG_DEBUG_CORE, ev->log, 0, "channel: %i", n);
20 
21         if (n == NGX_ERROR) {
22             if (ngx_event_flags & NGX_USE_EPOLL_EVENT) {
23                 ngx_del_conn(c, 0);
24             }
25             ngx_close_connection(c);
26             return;
27         }
28 
29         if (ngx_event_flags & NGX_USE_EVENTPORT_EVENT) {
30             if (ngx_add_event(ev, NGX_READ_EVENT, 0) == NGX_ERROR) {
31                 return;
32             }
33         }
34 
35         if (n == NGX_AGAIN) {
36             return;
37         }
38 
39         ngx_log_debug1(NGX_LOG_DEBUG_CORE, ev->log, 0,
40                        "channel command: %d", ch.command);
41             //处理消息命令
42         switch (ch.command) {
43         case NGX_CMD_QUIT:
44             ngx_quit = 1;
45             break;
46 
47         case NGX_CMD_TERMINATE:
48             ngx_terminate = 1;
49             break;
50 
51         case NGX_CMD_REOPEN:
52             ngx_reopen = 1;
53             break;
54 
55         case NGX_CMD_OPEN_CHANNEL:
56 
57             ngx_log_debug3(NGX_LOG_DEBUG_CORE, ev->log, 0,
58                            "get channel s:%i pid:%P fd:%d",
59                            ch.slot, ch.pid, ch.fd);
60 
61             ngx_processes[ch.slot].pid = ch.pid;
62             ngx_processes[ch.slot].channel[0] = ch.fd;
63             break;
64 
65         case NGX_CMD_CLOSE_CHANNEL:
66 
67             ngx_log_debug4(NGX_LOG_DEBUG_CORE, ev->log, 0,
68                            "close channel s:%i pid:%P our:%P fd:%d",
69                            ch.slot, ch.pid, ngx_processes[ch.slot].pid,
70                            ngx_processes[ch.slot].channel[0]);
71 
72             if (close(ngx_processes[ch.slot].channel[0]) == -1) {
73                 ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
74                               "close() channel failed");
75             }
76             ngx_processes[ch.slot].channel[0] = -1;
77             break;
78         }
79     }
80 }
 

 

以上分析了nginx进程的通信机制以及工作逻辑模型,下面以图表的形式做个总结:

 

 

 

本文档也是以前研究分析的,难免会有不准确之处,希望大家一起研究探讨。

评论(0
© 2014 mamicode.com 版权所有 京ICP备13008772号-2  联系我们:gaon5@hotmail.com
迷上了代码!