PHP网络处理模块FPM源码分析
发布:smiling 来源: PHP粉丝网 添加日期:2023-09-16 16:20:10 浏览: 评论:0
一个请求从浏览器到达PHP脚本执行中间有个必要模块是网络处理模块,FPM是这个模块的一部分,配合fastcgi协议实现对请求的从监听到转发到PHP处理,并将结果返回这条流程。
FPM采用多进程模型,就是创建一个master进程,在master进程中创建并监听socket,然后fork多个子进程,然后子进程各自accept请求,子进程在启动后阻塞在accept上,有请求到达后开始读取请求 数据,读取完成后开始处理然后再返回,在这期间是不会接收其它请求的,也就是说fpm的子进程同时只能响应 一个请求,只有把这个请求处理完成后才会accept下一个请求,这是一种同步阻塞的模型。master进程负责管理子进程,监听子进程的状态,控制子进程的数量。master进程与worker进程之间通过共享变量同步信息。
从main函数开始
- int main(int argc, char *argv[])
- {
- zend_signal_startup();
- // 将全局变量sapi_module设置为cgi_sapi_module
- sapi_startup(&cgi_sapi_module);
- fcgi_init();
- // 获取命令行参数,其中php-fpm -D、-i等参数都是在这里被解析出来的
- // ...
- cgi_sapi_module.startup(&cgi_sapi_module);
- fpm_init(argc, argv, fpm_config ? fpm_config : CGIG(fpm_config), fpm_prefix, fpm_pid, test_conf, php_allow_to_run_as_root, force_daemon, force_stderr);
- // master进程会在这一步死循环,后面的流程都是子进程在执行。
- fcgi_fd = fpm_run(&max_requests);
- fcgi_fd = fpm_run(&max_requests);
- request = fpm_init_request(fcgi_fd);
- // accept请求
- // ....
- }
main()函数展现了这个fpm运行完整的框架,可见整个fpm主要分为三个部分:
1、运行前的fpm_init();
2、运行函数fpm_run();
3、子进程accept请求处理。
FPM中的事件监听机制
在详细了解fpm工作过程前,我们要先了解fpm中的事件机制。在fpm中事件的监听默认使用kqueue来实现,关于kqueue的介绍可以看看我之前整理的这篇文章kqueue用法简介。
- // fpm中的事件结构体
- struct fpm_event_s {
- // 事件的句柄
- int fd;
- // 下一次触发的事件
- struct timeval timeout;
- // 频率:多久执行一次
- struct timeval frequency;
- // 事件触发时调用的函数
- void (*callback)(struct fpm_event_s *, short, void *);
- void *arg; // 调用callback时的参数
- // FPM_EV_READ:读;FPM_EV_TIMEOUT:;FPM_EV_PERSIST:;FPM_EV_EDGE:;
- int flags;
- int index; // 在fd句柄数组中的索引
- // 事件的类型 FPM_EV_READ:读;FPM_EV_TIMEOUT:计时器;FPM_EV_PERSIST:;FPM_EV_EDGE:;
- short which;
- };
- // 事件队列
- typedef struct fpm_event_queue_s {
- struct fpm_event_queue_s *prev;
- struct fpm_event_queue_s *next;
- struct fpm_event_s *ev;
- } fpm_event_queue;
以fpm_run()中master进程注册的一个sp[0]的可读事件为例:
- void fpm_event_loop(int err)
- {
- static struct fpm_event_s signal_fd_event;
- // 创建一个事件:管道sp[0]可读时触发
- fpm_event_set(&signal_fd_event, fpm_signals_get_fd(), FPM_EV_READ, &fpm_got_signal, NULL);
- // 将事件添加进queue
- fpm_event_add(&signal_fd_event, 0);
- // 处理定时器等逻辑
- // 以阻塞的方式获取事件
- // module->wait()是一个接口定义的方法签名,下面展示kqueue的实现
- ret = module->wait(fpm_event_queue_fd, timeout);
- }
- int fpm_event_add(struct fpm_event_s *ev, unsigned long int frequency)
- {
- // ...
- // 如果事件是触发事件则之间添加进queue中
- // 对于定时器事件先根据事件的frequency设置事件的触发频率和下一次触发的事件
- if (fpm_event_queue_add(&fpm_event_queue_timer, ev) != 0) {
- return -1;
- }
- return 0;
- }
- static int fpm_event_queue_add(struct fpm_event_queue_s **queue, struct fpm_event_s *ev)
- {
- // ...
- // 构建并将当前事件插入事件队列queue中
- if (*queue == fpm_event_queue_fd && module->add) {
- // module->add(ev)是一个接口定义的方法签名,下面展示kqueue的实现
- module->add(ev);
- }
- return 0;
- }
- // kqueue关于添加事件到kqueue的实现
- static int fpm_event_kqueue_add(struct fpm_event_s *ev) /* {{{ */
- {
- struct kevent k;
- int flags = EV_ADD;
- if (ev->flags & FPM_EV_EDGE) {
- flags = flags | EV_CLEAR;
- }
- EV_SET(&k, ev->fd, EVFILT_READ, flags, 0, 0, (void *)ev);
- if (kevent(kfd, &k, 1, NULL, 0, NULL) < 0) {
- zlog(ZLOG_ERROR, "kevent: unable to add event");
- return -1;
- }
- /* mark the event as registered */
- ev->index = ev->fd;
- return 0;
- }
FPM中关于kqueue的实现
- // kqueue关于从kqueue中监听事件的实现
- static int fpm_event_kqueue_wait(struct fpm_event_queue_s *queue, unsigned long int timeout) /* {{{ */
- {
- struct timespec t;
- int ret, i;
- /* ensure we have a clean kevents before calling kevent() */
- memset(kevents, 0, sizeof(struct kevent) * nkevents);
- /* convert ms to timespec struct */
- t.tv_sec = timeout / 1000;
- t.tv_nsec = (timeout % 1000) * 1000 * 1000;
- /* wait for incoming event or timeout */
- ret = kevent(kfd, NULL, 0, kevents, nkevents, &t);
- if (ret == -1) {
- /* trigger error unless signal interrupt */
- if (errno != EINTR) {
- zlog(ZLOG_WARNING, "epoll_wait() returns %d", errno);
- return -1;
- }
- }
- /* fire triggered events */
- for (i = 0; i < ret; i++) {
- if (kevents[i].udata) {
- struct fpm_event_s *ev = (struct fpm_event_s *)kevents[i].udata;
- fpm_event_fire(ev);
- /* sanity check */
- if (fpm_globals.parent_pid != getpid()) {
- return -2;
- }
- }
- }
- return ret;
- }
fpm_init
fpm_init()负责启动前的初始化工作,包括注册各个模块的销毁时用于清理变量的callback。下面只介绍几个重要的init。
fpm_conf_init_main
负责解析php-fpm.conf配置文件,分配worker pool内存结构并保存到全局变量fpm_worker_all_pools中,各worker pool配置解析到 fpm_worker_pool_s->config 中。
所谓worker pool 是fpm可以同时监听多个端口,每个端口对应一个worker pool。
fpm_scoreboard_init_main
为每个worker pool分配一个fpm_scoreboard_s结构的内存空间scoreboard,用于记录worker进程运行信息。
- // fpm_scoreboard_s 结构
- struct fpm_scoreboard_s {
- union {
- atomic_t lock;
- char dummy[16];
- };
- char pool[32];
- int pm; // 进程的管理方式 static、dynamic、ondemand
- time_t start_epoch;
- int idle; // 空闲的worker进程数
- int active; // 繁忙的worker进程数
- int active_max; // 最大繁忙进程数
- unsigned long int requests;
- unsigned int max_children_reached;
- int lq;
- int lq_max;
- unsigned int lq_len;
- unsigned int nprocs;
- int free_proc;
- unsigned long int slow_rq;
- struct fpm_scoreboard_proc_s *procs[];
- };
fpm_signals_init_main
fpm注册自己的信号量,并设置监听函数的处理逻辑。
- int fpm_signals_init_main() /* {{{ */
- {
- struct sigaction act;
- // 创建一个全双工套接字
- // 全双工的套接字是一个可以读、写的socket通道[0]和[1],每个进程固定一个管道。
- // 写数据时:管道不满不会被阻塞;读数据时:管道里没有数据会阻塞(可设置)
- // 向sp[0]写入数据时,sp[0]的读取将会被阻塞,sp[1]的写管道会被阻塞,sp[1]中此时读取sp[0]的数据
- if (0 > socketpair(AF_UNIX, SOCK_STREAM, 0, sp)) {
- zlog(ZLOG_SYSERROR, "failed to init signals: socketpair()");
- return -1;
- }
- if (0 > fd_set_blocked(sp[0], 0) || 0 > fd_set_blocked(sp[1], 0)) {
- zlog(ZLOG_SYSERROR, "failed to init signals: fd_set_blocked()");
- return -1;
- }
- if (0 > fcntl(sp[0], F_SETFD, FD_CLOEXEC) || 0 > fcntl(sp[1], F_SETFD, FD_CLOEXEC)) {
- zlog(ZLOG_SYSERROR, "falied to init signals: fcntl(F_SETFD, FD_CLOEXEC)");
- return -1;
- }
- memset(&act, 0, sizeof(act));
- act.sa_handler = sig_handler; // 监听到信号调用这个函数
- sigfillset(&act.sa_mask);
- if (0 > sigaction(SIGTERM, &act, 0) ||
- 0 > sigaction(SIGINT, &act, 0) ||
- 0 > sigaction(SIGUSR1, &act, 0) ||
- 0 > sigaction(SIGUSR2, &act, 0) ||
- 0 > sigaction(SIGCHLD, &act, 0) ||
- 0 > sigaction(SIGQUIT, &act, 0)) {
- zlog(ZLOG_SYSERROR, "failed to init signals: sigaction()");
- return -1;
- }
- return 0;
- }
- // 所有信号共用同一个处理函数
- static void sig_handler(int signo) /* {{{ */
- {
- static const char sig_chars[NSIG + 1] = {
- [SIGTERM] = 'T',
- [SIGINT] = 'I',
- [SIGUSR1] = '1',
- [SIGUSR2] = '2',
- [SIGQUIT] = 'Q',
- [SIGCHLD] = 'C'
- };
- char s;
- int saved_errno;
- if (fpm_globals.parent_pid != getpid()) {
- return;
- }
- saved_errno = errno;
- s = sig_chars[signo];
- zend_quiet_write(sp[1], &s, sizeof(s)); // 将信息对应的字节写进管道sp[1]端,此时sp[1]端的读数据会阻塞;数据可以从sp[0]端读取
- errno = saved_errno;
- }
fpm_sockets_init_main
每个worker pool 开启一个socket套接字。
fpm_event_init_main
这里启动master的事件管理器。用于管理IO、定时事件,其中IO事件通过kqueue、epoll、 poll、select等管理,定时事件就是定时器,一定时间后触发某个事件。同样,我们以kqueue的实现为例看下源码。
- int fpm_event_init_main()
- {
- // ...
- if (module->init(max) < 0) {
- zlog(ZLOG_ERROR, "Unable to initialize the event module %s", module->name);
- return -1;
- }
- // ...
- }
- // max用于指定kqueue事件数组的大小
- static int fpm_event_kqueue_init(int max) /* {{{ */
- {
- if (max < 1) {
- return 0;
- }
- kfd = kqueue();
- if (kfd < 0) {
- zlog(ZLOG_ERROR, "kqueue: unable to initialize");
- return -1;
- }
- kevents = malloc(sizeof(struct kevent) * max);
- if (!kevents) {
- zlog(ZLOG_ERROR, "epoll: unable to allocate %d events", max);
- return -1;
- }
- memset(kevents, 0, sizeof(struct kevent) * max);
- nkevents = max;
- return 0;
- }
fpm_run
fpm_init到此结束,下面进入fpm_run阶段,在这个阶段master进程会根据配置fork出多个子进程然后master进程会进入fpm_event_loop(0)函数,并在这个函数内部死循环,也就是说master进程将不再执行后面的代码,后面的逻辑全部是子进程执行的操作。
master进程在fpm_event_loop里通过管道sp来监听子进程的各个事件,同时也要处理自身产生的一些事件、定时器等任务,来响应的管理子进程。内部的逻辑在介绍事件监听机制时已经详细说过。
- int fpm_run(int *max_requests) /* {{{ */
- {
- struct fpm_worker_pool_s *wp;
- /* create initial children in all pools */
- for (wp = fpm_worker_all_pools; wp; wp = wp->next) {
- int is_parent;
- is_parent = fpm_children_create_initial(wp);
- if (!is_parent) {
- goto run_child;
- }
- }
- /* run event loop forever */
- fpm_event_loop(0);
- run_child: /* only workers reach this point */
- fpm_cleanups_run(FPM_CLEANUP_CHILD);
- *max_requests = fpm_globals.max_requests;
- return fpm_globals.listening_socket;
- }
子进程处理请求
回到main函数,fpm_run后面的逻辑都是子进程在运行。首先会初始化一个fpm的request结构的变量,然后子进程会阻塞在fcgi_accept_request(request)函数上等待请求。关于fcgi_accept_request函数就是死循环一个socket编程的accept函数来接收请求,并将请求数据全部取出。
- ...
- // 初始化request
- request = fpm_init_request(fcgi_fd);
- zend_first_try {
- // accept接收请求
- while (EXPECTED(fcgi_accept_request(request) >= 0)) {
- init_request_info();
- fpm_request_info();
- if (UNEXPECTED(php_request_startup() == FAILURE)) {
- // ...
- }
- if (UNEXPECTED(fpm_status_handle_request())) {
- goto fastcgi_request_done;
- }
- ...
- // 打开配置文件中DOCUMENT_ROOT设置的脚本
- if (UNEXPECTED(php_fopen_primary_script(&file_handle) == FAILURE)) {
- ...
- }
- fpm_request_executing();
- // 执行脚本
- php_execute_script(&file_handle);
- ...
- }
- // 销毁请求request
- fcgi_destroy_request(request);
- // fcgi退出
- fcgi_shutdown();
- if (cgi_sapi_module.php_ini_path_override) {
- free(cgi_sapi_module.php_ini_path_override);
- }
- if (cgi_sapi_module.ini_entries) {
- free(cgi_sapi_module.ini_entries);
- }
- } zend_catch {
- ...
- } zend_end_try();
Tags: PHP网络处理模块 FPM源码分析
- 上一篇:使用PHP抓取微博数据实现demo及原理解析
- 下一篇:最后一页
推荐文章
热门文章
最新评论文章
- 写给考虑创业的年轻程序员(10)
- PHP新手上路(一)(7)
- 惹恼程序员的十件事(5)
- PHP邮件发送例子,已测试成功(5)
- 致初学者:PHP比ASP优秀的七个理由(4)
- PHP会被淘汰吗?(4)
- PHP新手上路(四)(4)
- 如何去学习PHP?(2)
- 简单入门级php分页代码(2)
- php中邮箱email 电话等格式的验证(2)