PHP-FPM源码分析

Posted 2021-04-05 SegmentFault

来源：SegmentFault 思否

---

一个请求从浏览器到达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函数来接收请求，并将请求数据全部取出。

...
 

  
    
    
  

   
     
     
   
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
    
 
   
   
     
     
   
// 初始化requestrequest = 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();
  
    
    
  

---

点击左下角阅读原文，到  SegmentFault 思否社区  和文章作者展开更多互动和交流。

- END -