[thrift]PHP版多进程服务器

<?php
/**
 * 多进程形式的server.
 * @package thrift.server
 * @author flynetcn
 */
namespace Thrift\Server;

use Thrift\Server\TServer;
use Thrift\Transport\TTransport;
use Thrift\Exception\TException;
use Thrift\Exception\TTransportException;

class TMultiProcessServer extends TServer
{
	/**
	 * 捕获的信号编号
	 */
	static $catchQuitSignal = 0;

	/**
	 * worker进程数量
	 */
	private $workProcessNum = 4;

	/**
	 * 每个worker进程处理的最大请求数
	 */
	private $maxWorkRequestNum = 2000;

	/**
	 * 当前worker进程已处理的请求数
	 */
	private $currentWorkRequestNum = 0;
	
	/**
	 * 当前连接调用次数
	 */
	private $currentConnectCallNum = 0;

	/**
	 * 发送超时
	 */
	private $sendTimeoutSec = 1;

	/**
	 * 接收超时
	 */
	private $recvTimeoutSec = 1;

	/**
	 * 当前进程pid
	 */
	private $pid = 0;

	/**
	 * Flag for the main serving loop
	 */
	private $stop_ = false;

	/**
	 * List of children.
	 */
	protected $childrens = array();

	/**
	 * 服务器日志文件
	 */
	protected static $logFiles;
	protected static $pidFile;

	/**
	 * run
	 */ 
	public function serve($daemon=false, array $config=array())
	{
		if (isset($config['workProcessNum'])) {
			$this->workProcessNum = intval($config['workProcessNum']);
		}
		if ($this->workProcessNum < 1) {
			self::log(1, "child workProcessNum can not be less than 1");
			throw new TException('child workProcessNum can not be less than 1');
		}
		if (isset($config['maxWorkRequestNum'])) {
			$this->maxWorkRequestNum = intval($config['maxWorkRequestNum']);
		}
		if ($this->maxWorkRequestNum < 1) {
			self::log(1, "child maxWorkRequestNum can not be less than 1");
			throw new TException('child maxWorkRequestNum can not be less than 1');
		}
		if (isset($config['sendTimeoutSec'])) {
			$this->sendTimeoutSec = intval($config['sendTimeoutSec']);
		}
		if (isset($config['recvTimeoutSec'])) {
			$this->recvTimeoutSec = intval($config['recvTimeoutSec']);
		}
		if ($daemon) {
			$this->daemon();
			$this->registerSignalHandler();
			self::$logFiles = isset($config['logFiles']) && is_array($config['logFiles']) ? $config['logFiles'] : array();
			self::$pidFile = isset($config['pidFile']) ? $config['pidFile'] : '';
			declare(ticks=3);
		}
		$this->pid = posix_getpid();
		self::createPidFile($this->pid);
		self::log(0, "manage process({$this->pid}) has started");
		$this->transport_->listen();
		while (!$this->stop_) {
			while ($this->workProcessNum > 0) {
				try {
					$pid = pcntl_fork();
					if ($pid > 0) {
						$this->handleParent($pid, $this->workProcessNum);
					} else if ($pid === 0) {
						$this->pid = posix_getpid();
						$this->handleChild($this->workProcessNum);
					} else {
						self::log(1, "Failed to fork");
						throw new TException('Failed to fork');
					}
					$this->workProcessNum--;
				} catch (Exception $e) {
				}
			}
			$this->collectChildren();
			sleep(2);
			if (\Thrift\Server\TMultiProcessServer::$catchQuitSignal) {
				$this->stop();
			}
		}
	}
	
	public function getCurrentWorkRequestNum()
	{
		return $this->currentWorkRequestNum;
	}
	
	public function getCurrentConnectCallNum()
	{
		return $this->currentConnectCallNum;
	}

	/**
	 * Code run by the parent
	 *
	 * @param int $pid
	 * @param int $num 进程编号
	 * @return void
	 */
	private function handleParent($pid, $num)
	{
		$this->childrens[$pid] = $num;
	}

	/**
	 * Code run by the child.
	 *
	 * @param int $num 进程编号
	 * @return void
	 */
	private function handleChild($num)
	{
		self::log(0, "child process($this->pid) has started");
		$this->childrens = array();
		while (!$this->stop_) {
			try {
				$transport = $this->transport_->accept();
				if ($transport != null) {
					$transport->setSendTimeout($this->sendTimeoutSec * 1000);
					$transport->setRecvTimeout($this->recvTimeoutSec * 1000);
					$this->currentWorkRequestNum++;
					$this->currentConnectCallNum = 0;
					$inputTransport = $this->inputTransportFactory_->getTransport($transport);
					$outputTransport = $this->outputTransportFactory_->getTransport($transport);
					$inputProtocol = $this->inputProtocolFactory_->getProtocol($inputTransport);
					$outputProtocol = $this->outputProtocolFactory_->getProtocol($outputTransport);
					while ($this->processor_->process($inputProtocol, $outputProtocol)) {
						$this->currentConnectCallNum++;
					}
					@$transport->close();
				}
			} catch (TTransportException $e) {
			} catch (Exception $e) {
				self::log(1, $e->getMessage().'('.$e->getCode().')');
			}
			if (\Thrift\Server\TMultiProcessServer::$catchQuitSignal) {
				$this->stop();
			}
			if ($this->currentWorkRequestNum >= $this->maxWorkRequestNum) {
				self::log(0, "child process($this->pid) has processe {$this->currentWorkRequestNum} requests will be exit");
				$this->stop();
				break;
			}
		}
		exit(0);
	}

	/**
	 * Collects any children we may have
	 *
	 * @return void
	 */
	private function collectChildren()
	{
		foreach ($this->childrens as $pid => $num) {
			if (pcntl_waitpid($pid, $status, WNOHANG) > 0) {
				unset($this->childrens[$pid]);
				$this->workProcessNum++;
			}
		}
	}

	/**
	 * @return void
	 */
	public function stop()
	{
		$this->transport_->close();
		$this->stop_ = true;
		foreach ($this->childrens as $pid => $num) {
			if (!posix_kill($pid, SIGTERM)) {
			}
		}
	}

	/**
	 * 附加信号处理
	 */
	public static function sig_handler($signo)
	{
		switch ($signo) {
			case SIGTERM:
			case SIGHUP:
			case SIGQUIT:
			case SIGTSTP:
				$pid = posix_getpid();
				self::log(0, "process($pid) catch signo: $signo");
				\Thrift\Server\TMultiProcessServer::$catchQuitSignal = $signo;
				break;
			default:
		}
	}

	/**
	 * 附加信号处理
	 */
	private function registerSignalHandler()
	{
		pcntl_signal(SIGTERM, '\Thrift\Server\TMultiProcessServer::sig_handler');
		pcntl_signal(SIGHUP, '\Thrift\Server\TMultiProcessServer::sig_handler');
		pcntl_signal(SIGQUIT, '\Thrift\Server\TMultiProcessServer::sig_handler');
		pcntl_signal(SIGTSTP, '\Thrift\Server\TMultiProcessServer::sig_handler');
		declare(ticks=3);
	}

	/**
	 * 附加守护进程方式
	 */
	private function daemon()
	{
		if (!function_exists('posix_setsid')) {
			return;
		}
		if (($pid1 = pcntl_fork()) != 0) {
			exit;
		}
		posix_setsid();
		if (($pid2 = pcntl_fork()) != 0) {
			exit;
		}
	}

	public static function log($type, $msg)
	{
		static $fds;
		$msg = date('Y-m-d H:i:s')." $type {$msg}\n";
		if (isset(self::$logFiles[$type]) && self::$logFiles[$type]) {
			if (file_exists(self::$logFiles[$type])) {
				if (empty($fds[$type])) {
					$fds[$type] = fopen(self::$logFiles[$type], 'a');
				}
				if (!$fds[$type]) {
					$fds[$type] = fopen('php://stdout', 'w');
					fwrite($fds[$type], date('Y-m-d H:i:s')." WARNING fopen(".self::$logFiles[$type].") failed\n");
				}
			} else {
				if (!is_dir(dirname(self::$logFiles[$type])) && !mkdir(dirname(self::$logFiles[$type]), 0755, true)) {
					$fds[$type] = fopen('php://stdout', 'w');
					fwrite($fds[$type], date('Y-m-d H:i:s')." WARNING mkdir(".self::$logFiles[$type].") failed\n");
				} elseif (!($fds[$type] = fopen(self::$logFiles[$type], 'a'))) {
					$fds[$type] = fopen('php://stdout', 'w');
					fwrite($fds[$type], date('Y-m-d H:i:s')." WARNING fopen(".self::$logFiles[$type].") failed\n");
				}
			}
		} else {
			$fds[$type] = fopen('php://stdout', 'w');
		}
		$ret = fwrite($fds[$type], $msg);
		if (!$ret && self::$logFiles[$type]) {
			fclose($fds[$type]);
			$fds[$type] = fopen(self::$logFiles[$type], 'a');
			$ret = fwrite($fds[$type], $msg);
		}
		return true;
	}

	public static function createPidFile($pid=0)
	{
		if (!$pid) {
			$pid = posix_getpid();
		}
		if (file_exists(self::$pidFile)) {
			$fd = fopen(self::$pidFile, 'w');
			if (!$fd) {
				self::log(1, "fopen(".self::$pidFile.") failed");
				return false;
			}
		} else {
			if (!is_dir(dirname(self::$pidFile)) && !mkdir(dirname(self::$pidFile), 0755, true)) {
				self::log(1, "mkdir(".self::$pidFile.") failed");
				return false;
			} elseif (!($fd = fopen(self::$pidFile, 'w'))) {
				self::log(1, "fopen(".self::$pidFile.") failed");
				return false;
			}
		}
		if (!fwrite($fd, "$pid")) {
			self::log(1, "fwrite(".self::$pidFile.",$pid) failed");
			return false;
		}
		fclose($fd);
		return  true;
	}
}