nginx 之 多进程模型

Nginx的入口main函数在nginx.c文件中

int ngx_cdecl
main(int argc, char *const *argv)
{
// 资源初始化代码 ...

if (ngx_process == NGX_PROCESS_MASTER) {
// 启动多进程模型
ngx_master_process_cycle(cycle);

} else {
// 启动单进程模型
ngx_single_process_cycle(cycle);
}

return 0;
}

nginx的多线程框架启动入口函数nginx_master_process_cycle在文件ngx_process_cycle.c中实现：

void
ngx_master_process_cycle(ngx_cycle_t *cycle)
{
// 设置信号量
// 资源初始化

// 启动子进程(工作进程)
ngx_start_worker_processes(cycle, ccf->worker_processes, NGX_PROCESS_RESPAWN);

// 无限循环来处理服务器外部事件
for ( ;; ) {
// 等待函数开始时安装的信号发生
sigsuspend(&set);

// 终止事件处理
if (ngx_terminate) {
// ...
}
// 退出事件处理
if (ngx_quit) {
// ...
}
// 修改配置事件处理
if (ngx_reconfigure) {
// ...
}
// 重启事件处理
if (ngx_restart) {
// ...
}
if (ngx_reopen) {
// ...
}
// ...其余事件处理
}
}

static void
ngx_start_worker_processes(ngx_cycle_t *cycle, ngx_int_t n, ngx_int_t type)
{
// 生成指定个数的子进程(工作进程)
for (i = 0; i < n; i++) {
ngx_spawn_process(cycle, ngx_worker_process_cycle, NULL, "worker process", type);
}
}

ngx_spawn_process的实现位于文件ngx_process.c中

ngx_pid_t
ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data, char *name, ngx_int_t respawn)
{
pid = fork();
switch (pid) {

// 创建子进程出错
case -1:
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:
ngx_pid = ngx_getpid();
proc(cycle, data);
break;

// 主进程执行路径，什么也不做
default:
break;
}

return pid;
}

fork返回结果为0的是子进程的执行路径，子进程在进入switch后，执行ngx_spawn_process函数的第二个入参ngx_spawn_proc_pt proc，这个参数的类型是一个函数指针，定义在ngx_process.h文件中，定义为：

typedef void (*ngx_spawn_proc_pt) (ngx_cycle_t *cycle, void *data);

实际上在主线程执行函数ngx_start_worker_processes时，传入的这一函数指针的实际值为ngx_worker_process_cycle，实现了子线程的具体工作，该同样定义在ngx_process_cycle.c中，看看子线程究竟干了什么。

static void
ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data)
{
ngx_worker_process_init(cycle, 1);

// 在子进程中配置了使用多线程模型
#if (NGX_THREADS)
for (n = 0; n < ngx_threads_n; n++) {
if (ngx_create_thread((ngx_tid_t *) &ngx_threads
.tid, ngx_worker_thread_cycle, (void *) &ngx_threads
, cycle->log) != 0)
{
/* fatal */
exit(2);
}
}
#endif

// 循环处理子进程事件
for ( ;; ) {

if (ngx_exiting) { // ... }

// 具体事件处理逻辑
ngx_process_events_and_timers(cycle);

if (ngx_terminate) { // ... }
if (ngx_quit) { // ... }
if (ngx_reopen) { // ...}
}
}

子进程和主进程一样，不停的循环处理事件，差别是主进程用于处理整个服务器发生的外部事件，例如重启，配置更改，异常处理等等，而子进程是真正实现业务逻辑的地方，是不是和学校中的校长与副校长的分工很类似呢。

子进程中有一个很重要的配置就是，是否采用多线程模型来为子进程的事件处理服务，子进程的具体处理逻辑在函数ngx_process_events_and_timers中，该函数实现与ngx_events.c。

void
ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
// ngx_use_accept_mutex当然是一个信号量，作用是用于协调多个子进程对于accept操作的并发调用
if (ngx_use_accept_mutex) {
if (ngx_accept_disabled > 0) {
ngx_accept_disabled--;

} else {
if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
return;
}

if (ngx_accept_mutex_held) {
flags |= NGX_POST_EVENTS;

} else {
if (timer == NGX_TIMER_INFINITE
|| timer > ngx_accept_mutex_delay)
{
timer = ngx_accept_mutex_delay;
}
}
}
}

// 处理具体事件，这是一个宏，定义在ngx_event.h中，目的是将具体的事件处理逻辑和整个进程/线程框架分离
(void) ngx_process_events(cycle, timer, flags);

// 处理ngx_posted_accept_events是一个事件队列，其中包含了accept操作发生的事件，也就是有客户http请求进来的事件
if (ngx_posted_accept_events) {
ngx_event_process_posted(cycle, &ngx_posted_accept_events);
}

// 如果本子进程获得了监听锁，则此时可以将其释放
if (ngx_accept_mutex_held) {
ngx_shmtx_unlock(&ngx_accept_mutex);
}

// ngx_posted_events为另一个事件队列，其中包含了具体通道的IO事件
if (ngx_posted_events) {

// 如果子进程采用了多线程模型，则通知线程来处理队列中的事件，否则由子线程亲自处理
if (ngx_threaded) {
ngx_wakeup_worker_thread(cycle);

} else {
ngx_event_process_posted(cycle, &ngx_posted_events);
}
}
}

ngx_process_events的定义为：

#define ngx_process_events   ngx_event_actions.process_events

ngx_event_actions是一个结构体类型，其中定义了一些用于HTTP模型以及具体的IO模型的事件处理的函数指针，这个结构体的作用类似于Java中的接口，ngx_event_actions被各种不同的IO模型所分别实现，例如EPOLL, POLL等都有各自的版本，ngx_event_actions的定义为：

typedef struct {
ngx_int_t  (*add)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);
ngx_int_t  (*del)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);

ngx_int_t  (*enable)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);
ngx_int_t  (*disable)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);

ngx_int_t  (*add_conn)(ngx_connection_t *c);
ngx_int_t  (*del_conn)(ngx_connection_t *c, ngx_uint_t flags);

ngx_int_t  (*process_changes)(ngx_cycle_t *cycle, ngx_uint_t nowait);
ngx_int_t  (*process_events)(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);

ngx_int_t  (*init)(ngx_cycle_t *cycle, ngx_msec_t timer);
void       (*done)(ngx_cycle_t *cycle);
} ngx_event_actions_t;

处理Accept事件和IO读写事件的函数分别为：

void
ngx_event_process_posted(ngx_cycle_t *cycle,
ngx_thread_volatile ngx_event_t **posted)
{
ngx_event_t  *ev;

for ( ;; ) {

ev = (ngx_event_t *) *posted;

if (ev == NULL) {
return;
}

ngx_delete_posted_event(ev);

ev->handler(ev);
}
}

void
ngx_wakeup_worker_thread(ngx_cycle_t *cycle)
{
ngx_int_t     i;
// 唤醒空闲的线程来处理子进程中的事件
for (i = 0; i < ngx_threads_n; i++) {
if (ngx_threads[i].state == NGX_THREAD_FREE) {
ngx_cond_signal(ngx_threads[i].cv);
return;
}
}
}

上述两个函数实现在ngx_event_posted.c

这里有两个问题，线程既然被唤醒，那么是在哪里被阻塞的呢？线程被唤醒以后又是如何处理事件的？

在子进程的入口函数ngx_worker_process_cycle中，调用了ngx_create_thread来创建线程，其中传递的第二个参数ngx_worker_thread_cycle，即为线程的执行函数，

static ngx_thread_value_t
ngx_worker_thread_cycle(void *data)
{
// 为子线程安装信号
sigemptyset(&set);
sigaddset(&set, ngx_signal_value(NGX_RECONFIGURE_SIGNAL));
sigaddset(&set, ngx_signal_value(NGX_REOPEN_SIGNAL));
sigaddset(&set, ngx_signal_value(NGX_CHANGEBIN_SIGNAL));

err = ngx_thread_sigmask(SIG_BLOCK, &set, NULL);

// 资源初始化

// ngx_posted_events_mutex是守护子进程事件队列的信号量，用于协调多线程的并发访问
ngx_mutex_lock(ngx_posted_events_mutex);

// 循环等待线程被进程唤醒，以处理进程中的事件队列中的事件
for ( ;; ) {
thr->state = NGX_THREAD_FREE;

// 等待占有信号量
if (ngx_cond_wait(thr->cv, ngx_posted_events_mutex) == NGX_ERROR) {
return (ngx_thread_value_t) 1;
}

thr->state = NGX_THREAD_BUSY;

// 处理具体事件
if (ngx_event_thread_process_posted(cycle) == NGX_ERROR) {
return (ngx_thread_value_t) 1;
}
}
}

最后就剩下线程对于子进程中具体事件的处理了

ngx_int_t
ngx_event_thread_process_posted(ngx_cycle_t *cycle)
{
ngx_event_t  *ev;

for ( ;; ) {

ev = (ngx_event_t *) ngx_posted_events;

for ( ;; ) {

if (ev == NULL) {
return NGX_OK;
}

if (ngx_trylock(ev->lock) == 0) {
ev = ev->next;
continue;
}

ngx_delete_posted_event(ev);

ngx_mutex_unlock(ngx_posted_events_mutex);

ev->handler(ev);

ngx_mutex_lock(ngx_posted_events_mutex);

if (ev->locked) {
ngx_unlock(ev->lock);
}
break;
}
}
}

整个进程/线程框架非常简单明了，并与具体的事件业务逻辑完全解耦，非常值得学习。