PHP simulates supervisor process management

                                                       

Recommended: "PHP Video Tutorial"

Preface

Simulating supervisor process management DEMO (simple implementation)

That’s right, it’s about making wheels! The purpose is to learn!

Screenshot:

I have implemented the function of a Copy subprocess in the picture. I think it would be very useful if used in AMQP when adding or removing consumers.

Implementation

1. Start the sub-process in the main process loop to execute the command
2. Enter 127.0.0.1:7865 on the web to obtain the sub-process Status
3. The socket receives the request message, performs the corresponding operation, and returns to the web page.
4. Recycle the child process to prevent it from being called a zombie process.

Insufficiency: Unable to continue to monitor the error page. Since the response from the socket is loaded through the include function, the tail -f command cannot appear in the loaded page, otherwise the stream will fall into an infinite loop~. I think there should be a solution (I wrote the socket multi-process mode to imitate the mode in which fpm starts a sub-process to handle the request after receiving it, but there are problems with the execution. So I posted the code in the hope of getting everyone's advice).
Extension: Since the process can be well managed (expected), you can customize some of your own needs, such as: (1) Customize AMQP consumer process management services. (2) Simulate crontab timing service.

Knowledge points

In the process of code implementation, there are many details worth learning.
1. In the while() loop, the non-blocking mode of stream is enabled. So you cannot use sleep(1) in a loop, but use stream_select($read, $write, $except, 1) to block the stream internally.
For blocking and non-blocking modes, please refer here
2. There are many functions that can execute external programs, but they are all slightly different. proc_open is used here, which is a very powerful function. Before this, I have used pcntl_exec to execute external programs, but I need to pcntl_fork first. However, other methods such as exec and shell_exec cannot manage child processes.
3. When restarting or stopping a child process, it only changes the state of the child process in the memory of the main process first, and does not really operate on the child process. At the same place init() handles the child process. This can prevent some weird phenomena caused by the context when the child process is started.

Code

Because there are too many codes, if you have better suggestions for my solution, you can see it here on github.

Main process code: Process.php

<?php require_once __DIR__ . &#39;/Consumer.php&#39;;require_once __DIR__ . &#39;/StreamConnection.php&#39;;require_once __DIR__ . &#39;/Http.php&#39;;class Process{
    /** 
     * 待启动的消费者数组
     */
    protected $consumers = array();
    protected $childPids = array();

    const PPID_FILE = __DIR__ . &#39;/process&#39;;
    protected $serializerConsumer;

    public function __construct()
    {
        $this->consumers = $this->getConsumers();
    }

    // 这里是个DEMO，实际可以用读取配置文件的方式。
    public function getConsumers()
    {
        $consumer = new Consumer([
            'program' => 'test',
            'command' => '/usr/bin/php test.php',
            'directory' => __DIR__,
            'logfile' => __DIR__ . '/test.log',
            'uniqid' => uniqid(),
            'auto_restart' => false,
        ]);
        return [
            $consumer->uniqid => $consumer,
        ];
    }

    public function run()
    {
        if (empty($this->consumers)) {
            // consumer empty
            return;
        }
        if ($this->_notifyMaster()) {
            // master alive
            return;
        }

        $pid = pcntl_fork();
        if ($pid  0) {
            exit;
        }
        if (!posix_setsid()) {
            exit;
        }

        $stream = new StreamConnection('tcp://0.0.0.0:7865');
        @cli_set_process_title('AMQP Master Process');
        // 将主进程ID写入文件
        file_put_contents(self::PPID_FILE, getmypid());
        // master进程继续
        while (true) {
            $this->init();
            pcntl_signal_dispatch();
            $this->waitpid();
            // 如果子进程被全部回收，则主进程退出
            // if (empty($this->childPids)) {
            //     $stream->close($stream->getSocket());
            //     break;
            // }
            $stream->accept(function ($uniqid, $action) {
                $this->handle($uniqid, $action);
                return $this->display();
            });
        }
    }

    protected function init()
    {
        foreach ($this->consumers as &$c) {
            switch ($c->state) {
                case Consumer::RUNNING:
                case Consumer::STOP:
                    break;
                case Consumer::NOMINAL:
                case Consumer::STARTING:
                    $this->fork($c);
                    break;
                case Consumer::STOPING:
                    if ($c->pid && posix_kill($c->pid, SIGTERM)) {
                        $this->reset($c, Consumer::STOP);
                    }
                    break;
                case Consumer::RESTART:
                    if (empty($c->pid)) {
                        $this->fork($c);
                        break;
                    }
                    if (posix_kill($c->pid, SIGTERM)) {
                        $this->reset($c, Consumer::STOP);
                        $this->fork($c);
                    }
                    break;
                default:
                    break;
            }
        }
    }

    protected function reset(Consumer $c, $state)
    {
        $c->pid = '';
        $c->uptime = '';
        $c->state = $state;
        $c->process = null;
    }

    protected function waitpid()
    {
        foreach ($this->childPids as $uniqid => $pid) {
            $result = pcntl_waitpid($pid, $status, WNOHANG);
            if ($result == $pid || $result == -1) {
                unset($this->childPids[$uniqid]);
                $c = &$this->consumers[$uniqid];
                $state = pcntl_wifexited($status) ? Consumer::EXITED : Consumer::STOP;
                $this->reset($c, $state);
            }
        }
    }


    /**
     * 父进程存活情况下，只会通知父进程信息，否则可能产生多个守护进程
     */
    private function _notifyMaster()
    {
        $ppid = file_get_contents(self::PPID_FILE );
        $isAlive = $this->checkProcessAlive($ppid);
        if (!$isAlive) return false;
        return true;
    }

    public function checkProcessAlive($pid)
    {
        if (empty($pid)) return false;
        $pidinfo = `ps co pid {$pid} | xargs`;
        $pidinfo = trim($pidinfo);
        $pattern = "/.*?PID.*?(\d+).*?/";
        preg_match($pattern, $pidinfo, $matches);
        return empty($matches) ? false : ($matches[1] == $pid ? true : false);
    }

    /**
     * fork一个新的子进程
     */
    protected function fork(Consumer $c)
    {
        $descriptorspec = [2 => ['file', $c->logfile, 'a'],];
        $process = proc_open('exec ' . $c->command, $descriptorspec, $pipes, $c->directory);
        if ($process) {
            $ret = proc_get_status($process);
            if ($ret['running']) {
                $c->state = Consumer::RUNNING;
                $c->pid = $ret['pid'];
                $c->process = $process;
                $c->uptime = date('m-d H:i');
                $this->childPids[$c->uniqid] = $ret['pid'];
            } else {
                $c->state = Consumer::EXITED;
                proc_close($process);
            }
        } else {
            $c->state = Consumer::ERROR;
        }
        return $c;
    }

    public function display()
    {
        $location = 'http://127.0.0.1:7865';
        $basePath = Http::$basePath;
        $scriptName = isset($_SERVER['SCRIPT_NAME']) &&
            !empty($_SERVER['SCRIPT_NAME']) &&
            $_SERVER['SCRIPT_NAME'] != '/' ? $_SERVER['SCRIPT_NAME'] : '/index.php';
        if ($scriptName == '/index.html') {
            return Http::status_301($location);
        }

        $sourcePath = $basePath . $scriptName;
        if (!is_file($sourcePath)) {
            return Http::status_404();
        }

        ob_start();
        include $sourcePath;
        $response = ob_get_contents();
        ob_clean();

        return Http::status_200($response);
    }

    public function handle($uniqid, $action)
    {
        if (!empty($uniqid) && !isset($this->consumers[$uniqid])) {
            return;
        }
        switch ($action) {
            case 'refresh':
                break;
            case 'restartall':
                $this->killall(true);
                break;
            case 'stopall':
                $this->killall();
                break;
            case 'stop':
                $c = &$this->consumers[$uniqid];
                if ($c->state != Consumer::RUNNING) break;
                $c->state = Consumer::STOPING;
                break;
            case 'start':
                $c = &$this->consumers[$uniqid];
                if ($c->state == Consumer::RUNNING) break;
                $c->state = Consumer::STARTING;
                break;
            case 'restart':
                $c = &$this->consumers[$uniqid];
                $c->state = Consumer::RESTART;
                break;
            case 'copy':
                $c = $this->consumers[$uniqid];
                $newC = clone $c;
                $newC->uniqid = uniqid('C');
                $newC->state = Consumer::NOMINAL;
                $newC->pid = '';
                $this->consumers[$newC->uniqid] = $newC;
                break;
            default:
                break;
        }
    }

    protected function killall($restart = false)
    {
        foreach ($this->consumers as &$c) {
            $c->state = $restart ? Consumer::RESTART : Consumer::STOPING;
        }
    }}$cli = new Process();$cli->run();

Consumer consumer object

<?php require_once __DIR__ . &#39;/BaseObject.php&#39;;class Consumer extends BaseObject{
    /** 开启多少个消费者 */
    public $numprocs = 1;
    /** 当前配置的唯一标志 */
    public $program;
    /** 执行的命令 */
    public $command;
    /** 当前工作的目录 */
    public $directory;

    /** 通过 $qos $queueName $duplicate 生成的 $queue */
    public $queue;
    /** 程序执行日志记录 */
    public $logfile = &#39;&#39;;
    /** 消费进程的唯一ID */
    public $uniqid;
    /** 进程IDpid */
    public $pid;
    /** 进程状态 */
    public $state = self::NOMINAL;
    /** 自启动 */
    public $auto_restart = false;

    public $process;
    /** 启动时间 */
    public $uptime;

    const RUNNING = &#39;running&#39;;
    const STOP = &#39;stoped&#39;;
    const NOMINAL = &#39;nominal&#39;;
    const RESTART = &#39;restart&#39;;
    const STOPING = &#39;stoping&#39;;
    const STARTING = &#39;stating&#39;;
    const ERROR = &#39;error&#39;;
    const BLOCKED = &#39;blocked&#39;;
    const EXITED = &#39;exited&#39;;
    const FATEL = &#39;fatel&#39;;}

stream related code: StreamConnection.php

<?php class StreamConnection{
    protected $socket;
    protected $timeout = 2; //s
    protected $client;

    public function __construct($host)
    {
        $this->socket = $this->connect($host);
    }

    public function connect($host)
    {
        $socket = stream_socket_server($host, $errno, $errstr);
        if (!$socket) {
            exit('stream error');
        }
        stream_set_timeout($socket, $this->timeout);
        stream_set_chunk_size($socket, 1024);
        stream_set_blocking($socket, false);
        $this->client = [$socket];
        return $socket;
    }

    public function accept(Closure $callback)
    {
        $read = $this->client;
        if (stream_select($read, $write, $except, 1) socket, $read)) {
            $cs = stream_socket_accept($this->socket);
            $this->client[] = $cs;
        }
        foreach ($read as $s) {
            if ($s == $this->socket) continue;
            $header = fread($s, 1024);
            if (empty($header)) {
                $index = array_search($s, $this->client);
                if ($index)
                    unset($this->client[$index]);
                $this->close($s);
                continue;
            }
            Http::parse_http($header);
            $uniqid = isset($_GET['uniqid']) ? $_GET['uniqid'] : '';
            $action = isset($_GET['action']) ? $_GET['action'] : '';
            $response = $callback($uniqid, $action);
            $this->write($s, $response);
            $index = array_search($s, $this->client);
            if ($index)
                unset($this->client[$index]);
            $this->close($s);
        }
    }

    public function write($socket, $response)
    {
        $ret = fwrite($socket, $response, strlen($response));
    }

    public function close($socket)
    {
        $flag = fclose($socket);
    }

    public function getSocket()
    {
        return $this->socket;
    }}

Http response Code: Http.php

<?php class Http{

    public static $basePath = __DIR__ . &#39;/views&#39;;
    public static $max_age = 120; //秒

    /*
    *  函数:     parse_http
    *  描述:     解析http协议
    */
    public static function parse_http($http)
    {
        // 初始化
        $_POST = $_GET = $_COOKIE = $_REQUEST = $_SESSION = $_FILES =  array();
        $GLOBALS[&#39;HTTP_RAW_POST_DATA&#39;] = &#39;&#39;;
        // 需要设置的变量名
        $_SERVER = array(
            &#39;QUERY_STRING&#39; => '',
            'REQUEST_METHOD' => '',
            'REQUEST_URI' => '',
            'SERVER_PROTOCOL' => '',
            'SERVER_SOFTWARE' => '',
            'SERVER_NAME' => '',
            'HTTP_HOST' => '',
            'HTTP_USER_AGENT' => '',
            'HTTP_ACCEPT' => '',
            'HTTP_ACCEPT_LANGUAGE' => '',
            'HTTP_ACCEPT_ENCODING' => '',
            'HTTP_COOKIE' => '',
            'HTTP_CONNECTION' => '',
            'REMOTE_ADDR' => '',
            'REMOTE_PORT' => '0',
            'SCRIPT_NAME' => '',
            'HTTP_REFERER' => '',
            'CONTENT_TYPE' => '',
            'HTTP_IF_NONE_MATCH' => '',
        );

        // 将header分割成数组
        list($http_header, $http_body) = explode("\r\n\r\n", $http, 2);
        $header_data = explode("\r\n", $http_header);

        list($_SERVER['REQUEST_METHOD'], $_SERVER['REQUEST_URI'], $_SERVER['SERVER_PROTOCOL']) = explode(' ', $header_data[0]);

        unset($header_data[0]);
        foreach ($header_data as $content) {
            // \r\n\r\n
            if (empty($content)) {
                continue;
            }
            list($key, $value) = explode(':', $content, 2);
            $key = strtolower($key);
            $value = trim($value);
            switch ($key) {
                case 'host':
                    $_SERVER['HTTP_HOST'] = $value;
                    $tmp = explode(':', $value);
                    $_SERVER['SERVER_NAME'] = $tmp[0];
                    if (isset($tmp[1])) {
                        $_SERVER['SERVER_PORT'] = $tmp[1];
                    }
                    break;
                case 'cookie':
                    $_SERVER['HTTP_COOKIE'] = $value;
                    parse_str(str_replace('; ', '&', $_SERVER['HTTP_COOKIE']), $_COOKIE);
                    break;
                case 'user-agent':
                    $_SERVER['HTTP_USER_AGENT'] = $value;
                    break;
                case 'accept':
                    $_SERVER['HTTP_ACCEPT'] = $value;
                    break;
                case 'accept-language':
                    $_SERVER['HTTP_ACCEPT_LANGUAGE'] = $value;
                    break;
                case 'accept-encoding':
                    $_SERVER['HTTP_ACCEPT_ENCODING'] = $value;
                    break;
                case 'connection':
                    $_SERVER['HTTP_CONNECTION'] = $value;
                    break;
                case 'referer':
                    $_SERVER['HTTP_REFERER'] = $value;
                    break;
                case 'if-modified-since':
                    $_SERVER['HTTP_IF_MODIFIED_SINCE'] = $value;
                    break;
                case 'if-none-match':
                    $_SERVER['HTTP_IF_NONE_MATCH'] = $value;
                    break;
                case 'content-type':
                    if (!preg_match('/boundary="?(\S+)"?/', $value, $match)) {
                        $_SERVER['CONTENT_TYPE'] = $value;
                    } else {
                        $_SERVER['CONTENT_TYPE'] = 'multipart/form-data';
                        $http_post_boundary = '--' . $match[1];
                    }
                    break;
            }
        }

        // script_name
        $_SERVER['SCRIPT_NAME'] = parse_url($_SERVER['REQUEST_URI'], PHP_URL_PATH);

        // QUERY_STRING
        $_SERVER['QUERY_STRING'] = parse_url($_SERVER['REQUEST_URI'], PHP_URL_QUERY);
        if ($_SERVER['QUERY_STRING']) {
            // $GET
            parse_str($_SERVER['QUERY_STRING'], $_GET);
        } else {
            $_SERVER['QUERY_STRING'] = '';
        }

        // REQUEST
        $_REQUEST = array_merge($_GET, $_POST);

        return array('get' => $_GET, 'post' => $_POST, 'cookie' => $_COOKIE, 'server' => $_SERVER, 'files' => $_FILES);
    }

    public static function status_404()
    {
        return <p>Script to be executed: test.php</p><pre class="brush:php;toolbar:false"><?php while(true) {
    file_put_contents(__DIR__  .  &#39;/test.log&#39;, date(&#39;Y-m-d H:i:s&#39;));
    sleep(1);}

View page in the current directory:
|- Process.php
|- Http. php
|- StreamConnection.php
|- Consumer.php
|- BaseObject.php
|- views/

For more programming-related knowledge, please visit: Programming teaching! !

The above is the detailed content of PHP simulates supervisor process management. For more information, please follow other related articles on the PHP Chinese website!