nginx父子进程间通信

父子进程间通信

1 内部流程 — socketpair 基础

主要是通过socketpair()函数实现的,下面捋一下内部流程:
1. 话说要从ngx_start_worker_processes函数讲起:
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_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start worker processes");
    ch.command = NGX_CMD_OPEN_CHANNEL;
	/*n是由配置文件中获取,子进程个数*/
    for (i = 0; i < n; i++) {
		/*子进程创建函数,下面具体讲*/
        ngx_spawn_process(cycle, ngx_worker_process_cycle,
                          (void *) (intptr_t) i, "worker process", type);
        ch.pid = ngx_processes[ngx_process_slot].pid;
        ch.slot = ngx_process_slot;
        ch.fd = ngx_processes[ngx_process_slot].channel[0];
		
		/*父进程中执行该函数,主要像前面各个子进程的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)
{
	//...
	//...
	/*创建一对套接字讲用于父子进程间通信*/
	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_channel = ngx_processes[s].channel[1];//ngx_channel干啥用的?留个悬念
	//...
	
	/*重点来了,创建子进程*/
	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();
		/*调用ngx_worker_process_cycle函数,子进程死循环处理流程,下面再讲*/
        proc(cycle, data);
        break;
 
    default:
        break;
    }
 
	//...
	/*对子进程的相关信息进行保存*/
	ngx_processes[s].pid
	//...
	/*数组下标加1*/
	if (s == ngx_last_process) {
        ngx_last_process++;
    }
    return pid;
}
 
ngx_spawn_process()函数是在一个for循环中调用,假如有4个子进程,也就是说会进行四次
socketpair()的创建执行,每一次的创建,ngx_processes[s].channel都会有两个fd生成,
假设进行第一次循环时,ngx_processes[0].channel里面已经有ngx_processes[0].channel[0]和
ngx_processes[0].channel[1],继续往下执行,执行子进程创建执行proc(cycle, data);时,调用
static void
ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data)
{
	//...
	/*子进程初始化相关工作*/
	ngx_worker_process_init(cycle, worker);
	//...
	for( ;; )
	{
		/*子进程死循环,虽然里面内容很重要,但与本节无关*/
	}
}
static void
ngx_worker_process_init(ngx_cycle_t *cycle, ngx_int_t worker)
{
	//...
	//...
	/*for循环,进行了下异常处理*/
    for (n = 0; n < ngx_last_process; n++) {
	}
	
	/*在子进程中关闭掉channel[0],只需要用channel[1]*/
	if (close(ngx_processes[ngx_process_slot].channel[0]) == -1) {
	ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
				  "close() channel failed");
	}
	/*这个比较重要,但还没理解透*/
    if (ngx_add_channel_event(cycle, ngx_channel, NGX_READ_EVENT,
                              ngx_channel_handler)
        == NGX_ERROR)
    {
        /* fatal */
        exit(2);
    }
}
ngx_int_t
ngx_add_channel_event(ngx_cycle_t *cycle, ngx_fd_t fd, ngx_int_t event,
    ngx_event_handler_pt handler)
{
	//...
	//...
    ev = (event == NGX_READ_EVENT) ? rev : wev;
	/*将ngx_channel_handler函数挂在ev->handler上*/
    ev->handler = handler;
 
	/*如果是epoll,则执行ngx_epoll_add_connection函数,在里面可以看到
	    if (epoll_ctl(ep, EPOLL_CTL_ADD, c->fd, &ee) == -1) {
        ngx_log_error(NGX_LOG_ALERT, c->log, ngx_errno,
                      "epoll_ctl(EPOLL_CTL_ADD, %d) failed", c->fd);
        return NGX_ERROR;
    }
	将描述符第二个参数fd,而fd就是ngx_channel加入到了事件中
	ngx_channel在哪里?
	ngx_channel来自于ngx_spawn_process()中
	ngx_channel = ngx_processes[s].channel[1];
*/
    if (ngx_add_conn && (ngx_event_flags & NGX_USE_EPOLL_EVENT) == 0) {
        if (ngx_add_conn(c) == NGX_ERROR) {
            ngx_free_connection(c);
            return NGX_ERROR;
        }
 
    } else {
		/*添加事件,一旦有可读事件到来时,执行ev->handler*/
        if (ngx_add_event(ev, event, 0) == NGX_ERROR) {
            ngx_free_connection(c);
            return NGX_ERROR;
        }
    }
 
    return NGX_OK;
}
 
/*当有可读事件来的时候,触发该函数*/
static void
ngx_channel_handler(ngx_event_t *ev)
{
	//...
	ngx_connection_t  *c;
	c = ev->data; //下面的c->fd来自哪里?
	/*ev->data,*/
	//...
	for ( ;; ) {
		
		/*里面就是调用recvmsg(s, &msg, 0);读取消息*/
        n = ngx_read_channel(c->fd, &ch, sizeof(ngx_channel_t), ev->log);
		switch()
		{
			//...
			case NGX_CMD_OPEN_CHANNEL:
			//...
			/*保存各个槽位的子进程信息*/
			//...
		}
	}
	//...
}
从上面可以小结一下,感觉一下,子进程拥有套接字ngx_processes[s].channel[1],并加入
了可读事件中,一直等待着读,即等待着调用recvmsg(),那么由谁来sendmsg呢?通过哪个
套接字呢?
继续:
视线回到ngx_spawn_process()函数中,该函数带领我们一步步走进子进程的处理过程,回到
该函数调用的地方即函数ngx_start_worker_processes()中,它下面接着执行ch.pid、ch.slot
、ch.fd的赋值,用处在下面:接着调用
/*注意调用该函数是在父进程中执行*/
static void
ngx_pass_open_channel(ngx_cycle_t *cycle, ngx_channel_t *ch)
{
	/*for循环,细细体味下,假如多个子进程时,通过该循环向不同的channel[0]发送msg*/
    for (i = 0; i < ngx_last_process; i++) {
	/*一些异常处理*/
	//...
	/*该函数实际上就是调用sendmsg(s, &msg, 0);进行消息的发送*/
	/*注意参数:第一个参数ngx_processes[i].channel[0]就是要发送的fd,其实也在ch里面包含着*/
	ngx_write_channel(ngx_processes[i].channel[0],
					  ch, sizeof(ngx_channel_t), cycle->log);
	
	}
}
综上: 创建一个子进程时,父进程就会向各个channel[0]中sendmsg,子进程从channel[1]recvmsg

2 socketpair 函数可创建本地套接字

ngx_channel_t 频道是 Nginx master 进程与 worker 进程之间通信的常用工具,它是使用本机套接字实现的,即 socketpair 方法,它用于创建父子进程间使用的套接字。

#include <sys/types.h> /* See NOTES */
#include <sys/socket.h>
int socketpair(int domain, int type, int protocol, int sv[2]);
这个方法可以创建一对关联的套接字 sv[2]。
domain:表示域,在 Linux 下通常取值为 AF_UNIX;
type:取值为 SOCK_STREAM 或 SOCK_DGRAM,它表示在套接字上使用的是 TCP 还是 UDP;
protocol:必须传递 0;
sv[2]:是一个含有两个元素的整型数组,实际上就是两个套接字。
当 socketpair 返回 0 时,sv[2] 这两个套接字创建成功,否则 sockpair 返回 -1 表示失败.
当 socketpair 执行成功时,sv[2] 这两个套接字具备下列关系:
向 sv[0] 套接字写入数据,将可以从 sv[1] 套接字中读取到刚写入的数据;
同样,向 sv[1] 套接字写入数据,也可以从 sv[0] 中读取到写入的数据。

通常,在父、子进程通信前,会先调用 socketpair 方法创建这样一组套接字,在调用 fork 方法创建出子进程后,将会在父进程中关闭 sv[1] 套接字,仅使用 sv[0] 套接字用于向子进程发送数据以及接收子进程发送来的数据;
而在子进程中则关闭 sv[0] 套接字,仅使用 sv[1] 套接字既可以接收父进程发送来的数据,也可以向父进程发送数据。
ngx_channel_t 结构体是 Nginx 定义的 master 父进程与 worker 子进程间的消息格式,如下:

typedef struct {
    // 传递的 TCP 消息中的命令
    ngx_uint_t  command;
    // 进程 ID,一般是发送命令方的进程 ID
    ngx_pid_t   pid;
    // 表示发送命令方在 ngx_processes 进程数组间的序号
    ngx_int_t   slot;
    // 通信的套接字句柄
    ngx_fd_t    fd;
}ngx_channel_t;

Nginx 针对 command 成员定义了如下命令:
// 打开频道,使用频道这种方式通信前必须发送的命令
#define NGX_CMD_OPEN_CHANNEL 1
// 关闭已经打开的频道,实际上也就是关闭套接字
#define NGX_CMD_CLOSE_CHANNEL 2
// 要求接收方正常地退出进程
#define NGX_CMD_QUIT 3
// 要求接收方强制地结束进程
#define NGX_CMD_TERMINATE 4
// 要求接收方重新打开进程已经打开过的文件
#define NGX_CMD_REOPEN 5
问:master 是如何启动、停止 worker 子进程的?
答:正是通过 socketpair 产生的套接字发送命令的,即每次要派生一个子进程之前,都会先调用 socketpair 方法。
在 Nginx 派生子进程的 ngx_spawn_process 方法中,会首先派生基于 TCP 的套接字,如下:

ngx_pid_t 
ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data, 
    char *name, ngx_int_t respawn)
{
    if (respawn != NGX_PROCESS_DETACHED) {  
        /* Solaris 9 still has no AF_LOCAL */
        
        // ngx_processes[s].channel 数组正是将要用于父、子进程间通信的套接字对
        if (socketpair(AF_UNIX, SOCK_STREAM, 0, ngx_processes[s].channel) == -1)
        {
            return NGX_INVALID_PID;
        }
        
        // 将 channel 套接字对都设置为非阻塞模式
        if (ngx_nonblocking(ngx_processes[s].channel[0]) == -1) {
            ngx_close_channel(ngx_processes[s].channel, cycle->log);
            return NGX_INVALID_PID;
        }
        
        if (ngx_nonblocking(ngx_processes[s].channel[1]) == -1) {
            ngx_close_channel(ngx_processes[s].channel, cycle->log);
            return NGX_INVALID_PID;
        }
    ...
}
ngx_processes 数组定义了 Nginx 服务中所有的进程,包括 master 进程和 worker 进程,如下:
#define NGX_MAX_PROCESSES 1024

// 虽然定义了 NGX_MAX_PROCESSES 个成员,但是已经使用的元素仅与启动的进程个数有关
ngx_processes_t ngx_processes[NGX_MAX_PROCESSES];
ngx_processes 数组的类型是 ngx_processes_t,对于频道来说,这个结构体只关心它的 channel 成员:

typedef struct {
    ...
    // socketpair 创建的套接字对
    ngx_socket_t channel[2];
}ngx_processes_t;
1. ngx_write_channel:使用频道发送 ngx_channel_t 消息
ngx_int_t 
ngx_write_channel(ngx_socket_t s, ngx_channel_t *ch, size_t size,
    ngx_log_t *log)
{
    ssize_t         n;
    ngx_err_t       err;
    struct iovec    iov[1];
    struct msghdr   msg;
    
#if (NGX_HAVE_MSGHDR_MSG_CONTROL)

    union {
        struct cmsghdr  cm;
        char            space[CMSG_SPACE(sizeof(int))];
    }cmsg;
    
    if (ch->fd == -1) {
        msg.msg_control = NULL;
        msg.msg_controllen = 0;
        
    } else {
        // 辅助数据
        msg.msg_control = (caddr_t)&cmsg;
        msg.msg_controllen = sizeof(cmsg);
        
        ngx_memzero(&cmsg, sizeof(cmsg));
        
        cmsg.cm.cmsg_len = CMSG_LEN(sizeof(int));
        cmsg.cm.cmsg_level = SOL_SOCKET;
        cmsg.cm.cmsg_type = SCM_RIGHTS;
        
        /*
         * We have to use ngx_memcpy() instead of simple
         *   *(int *) CMSG_DATA(&cmsg.cm) = ch->fd;
         * because some gcc 4.4 with -O2/3/s optimization issues the warning:
         *   dereferencing type-punned pointer will break strict-aliasing rules
         *
         * Fortunately, gcc with -O1 compiles this ngx_memcpy()
         * in the same simple assignment as in the code above
         */

        ngx_memcpy(CMSG_DATA(&cmsg.cm), &ch->fd, sizeof(int));
    }   
    msg.msg_flags = 0;  
#else

    if (ch->fd == -1) {
        msg.msg_accrights = NULL;
        msg.msg_accrightslen = 0;
        
    } else {
        msg.msg_accrights = (caddr_t) &ch->fd;
        msg.msg_accrightslen = sizeof(int);
    }   
#endif

    // 指向要发送的 ch 起始地址
    iov[0].iov_base = (char *) ch;
    iov[0].iov_len = size;
    
    // msg_name 和 msg_namelen 仅用于未连接套接字(如UDP)
    msg.msg_name = NULL;
    msg.msg_namelen = 0;
    msg.msg_iov = iov;
    msg.msg_iovlen = 1;
    
    // 将该 ngx_channel_t 消息发出去
    n = sendmsg(s, &msg, 0);
    
    if (n == -1) {
        err = ngx_errno;
        if (err == NGX_EAGAIN) {
            return NGX_AGAIN;
        }
        
        return NGX_ERROR;
    }  
    return NGX_OK;
}
2. ngx_read_channel: 读取消息
ngx_int_t
ngx_read_channel(ngx_socket_t s, ngx_channel_t *ch, size_t size, ngx_log_t *log)
{
    ssize_t             n;
    ngx_err_t           err;
    struct iovec        iov[1];
    struct msghdr       msg;
#if (NGX_HAVE_MSGHDR_MSG_CONTROL)
    union {
        struct cmsghdr  cm;
        char            space[CMSG_SPACE(sizeof(int))];
    } cmsg;
#else
    int                 fd;
#endif
    
    iov[0].iov_base = (char *)ch;
    iov[0].iov_len = size;
    
    msg.msg_name = NULL;
    msg.msg_namelen = 0;
    msg.msg_iov = iov;
    msg.msg_iovlen = 1;
    
#if (NGX_HAVE_MSGHDR_MSG_CONTROL)
    msg.msg_control = (caddr_t) &cmsg;
    msg.msg_controllen = sizeof(cmsg);
#else
    msg.msg_accrights = (caddr_t) &fd;
    msg.msg_accrightslen = sizeof(int);
#endif
    // 接收命令
    n = recvmsg(s, &msg, 0);
    if (n == -1) {
        err = ngx_errno;
        if (err == NGX_EAGAIN) {
            return NGX_AGAIN;
        }
        return NGX_ERROR;
    }
    if (n == 0) {
        return NGX_ERROR;
    }  
    // 接收的数据不足
    if ((size_t) n < sizeof(ngx_channel_t)) {
        return NGX_ERROR;
    } 
#if (NGX_HAVE_MSGHDR_MSG_CONTROL)  
    // 若接收到的命令为"打开频道,使用频道这种方式通信前必须发送的命令"
    if (ch->command == NGX_CMD_OPEN_CHANNEL) {
        
        if (cmsg.cm.cmsg_len < (socklen_t) CMSG_LEN(sizeof(int))) {
            return NGX_ERROR;
        }   
        if (cmsg.cm.cmsg_level != SOL_SOCKET || cmsg.cm.cmsg_type != SCM_RIGHTS)
        {
            return NGX_ERROR;
        }      
        /* ch->fd = *(int *) CMSG_DATA(&cmsg.cm); */
        ngx_memcpy(&ch->fd, CMSG_DATA(&cmsg.cm), sizeof(int));
    } 
    // 若接收到的消息是被截断的
    if (msg.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) {
        ngx_log_error(NGX_LOG_ALERT, log, 0,
                      "recvmsg() truncated data");
    }
#else 
    if (ch->command == NGX_CMD_OPEN_CHANNEL) {
        if (msg.msg_accrightslen != sizeof(int)) {
            return NGX_ERROR;
        }    
        ch->fd = fd;
    }
#endif

    return n;
}

在 Nginx 中,目前仅存在 master 进程向 worker 进程发送消息的场景,这时对于 socketpair 方法创建的 channel[2] 套接字来说,master 进程会使用 channel[0] 套接字来发送消息,而 worker 进程会使用 channel[1] 套接字来接收消息。

  1. ngx_add_channel_event: 把接收频道消息的套接字添加到 epoll 中
    worker 进程调度 ngx_read_channel 方法接收频道消息是通过该 ngx_add_channel_event 函数将接收频道消息的套接字(对于 worker 即为channel[1])添加到 epoll 中,当接收到父进程消息时子进程会通过 epoll 的事件回调相应的 handler 方法来处理这个频道消息,如下:
ngx_int_t 
ngx_add_channel_event(ngx_cycle_t *cycle, ngx_fd_t fd, ngx_int_t event, 
    ngx_event_handler_pt handler)
{
    ngx_event_t         *ev, *rev, *wev;
    ngx_connection_t    *c;
    // 获取一个空闲连接
    c = ngx_get_connection(fd, cycle->log);
    
    if (c == NULL) {
        return NGX_ERROR;
    }
    c->pool = cycle->pool;
    
    rev = c->read;
    wev = c->write;
 
    rev->log = cycle->log;
    wev->log = cycle->log;
    rev->channel = 1;
    wev->channel = 1; 
    ev = (event == NGX_READ_EVENT) ? rev : wev; 
    // 初始化监听该 ev 事件时调用的回调函数
    ev->handler = handler; 
    // 将该接收频道消息的套接字添加到 epoll 中
    if (ngx_add_conn && (ngx_event_flags && NGX_USE_EPOLL_EVENT) == 0) {
        // 这里是同时监听该套接字的读、写事件
        if (ngx_add_conn(c) == NGX_ERROR) {
            ngx_free_connection(c);
            return NGX_ERROR;
        }
         
    } else {
        // 这里是仅监听 ev 事件
        if (ngx_add_event(ev, event, 0) == NGX_ERROR) {
            ngx_free_connection(c);
            return NGX_ERROR;
        }
    } 
    return NGX_OK;
}
4. ngx_close_channel: 关闭这个频道通信方式
void
ngx_close_channel(ngx_fd_t *fd, ngx_log_t *log)
{
    if (close(fd[0]) == -1) {
        
    }
    if (close(fd[1]) == -1) {
        
    }
}

参考资料:

  1. https://blog.csdn.net/xyyaiguozhe/article/details/14124627
  2. https://www.cnblogs.com/jimodetiantang/p/9191092.html
上一篇:how tomcat works 读书笔记(二)----------一个简单的servlet容器


下一篇:软件中的一些速度