Worker.php

<?php
namespace TphpWorker;

use TphpWorker\Server\TcpServer;
use TphpWorker\Event\EventLoop;
use TphpWorker\Process\SignalHandler;
use TphpWorker\Process\ProcessManager;

class Worker {
    // 服务器列表:array<int, TcpServer>
    public array $servers = [];

    // 事件循环
    public EventLoop $loop;

    // 信号处理器
    public SignalHandler $signalHandler;

    // 进程管理器
    public ProcessManager $processManager;

    // 是否运行中
    public bool $running = false;

    // 工作进程数(所有 server 的最大值,用于决定是否启用多进程)
    public int $maxProcessCount = 1;

    // 是否为子进程(由 fork 设置)
    public bool $isChild = false;

    // 子进程的工作进程索引
    public int $workerIndex = 0;

    public function __construct() {
        $this->loop = new EventLoop();
        $this->signalHandler = new SignalHandler($this->loop);
        $this->processManager = new ProcessManager();
    }

    // 添加服务器
    public function addServer(TcpServer $server): void {
        $this->servers[] = $server;
        // 更新最大进程数
        if ($server->processCount > $this->maxProcessCount) {
            $this->maxProcessCount = $server->processCount;
        }
    }

    // 启动所有服务器并进入事件循环
    // 若 maxProcessCount > 1 且 pcntl 可用,则 fork 子进程
    public function runAll(): void {
        if (count($this->servers) === 0) {
            echo sprintf("Worker: no servers registered\n");
            return;
        }

        echo sprintf("Worker: starting %d server(s), max processes=%d\n",
            count($this->servers), $this->maxProcessCount);

        // 判断是否需要多进程
        $needFork = $this->maxProcessCount > 1 && $this->processManager->pcntlAvailable;

        if ($needFork) {
            $this->runMultiProcess();
        } else {
            if ($this->maxProcessCount > 1 && !$this->processManager->pcntlAvailable) {
                echo sprintf("Worker: pcntl not available, running in single process mode\n");
            }
            $this->runSingleProcess();
        }
    }

    // 单进程模式运行
    public function runSingleProcess(): void {
        // 启动所有服务器
        $this->startAllServers();

        // 设置信号处理(周期性检查停止标志)
        $this->signalHandler->setupPeriodicCheck(1.0);

        // 进入事件循环
        $this->running = true;
        echo sprintf("Worker: event loop started (single process)\n");
        $this->loop->run();

        // 事件循环退出,清理
        $this->running = false;
        $this->signalHandler->cleanup();
        $this->stopAll();
        echo sprintf("Worker: stopped\n");
    }

    // 多进程模式运行
    public function runMultiProcess(): void {
        $processCount = $this->maxProcessCount;

        echo sprintf("Worker: forking %d worker process(es)\n", $processCount);

        // fork 子进程
        // 注:闭包内直接使用 $this(TinyPHP 自动绑定),避免 use ($worker) 导致类型推断失败
        $this->processManager->fork($processCount, function(int $index) {
            // 子进程内执行
            $this->isChild = true;
            $this->workerIndex = $index;

            // 子进程创建自己的 EventLoop(避免与父进程共享)
            $this->loop = new EventLoop();
            $this->signalHandler = new SignalHandler($this->loop);

            // 启动所有服务器(在子进程的 EventLoop 中)
            $this->startAllServers();

            // 设置信号处理
            $this->signalHandler->setupPeriodicCheck(1.0);

            // 进入事件循环
            $this->running = true;
            echo sprintf("Worker[%d]: event loop started\n", $index);
            $this->loop->run();

            // 事件循环退出,清理
            $this->running = false;
            $this->signalHandler->cleanup();
            $this->stopAll();
            echo sprintf("Worker[%d]: stopped\n", $index);
        });

        // 父进程:监控子进程
        if ($this->processManager->isParent) {
            echo sprintf("Worker: parent process monitoring %d children\n",
                $this->processManager->childCount());

            // 设置信号处理(父进程也检查停止标志)
            $this->signalHandler->setupPeriodicCheck(1.0);

            // 父进程等待子进程退出
            // 子进程异常退出时,重新 fork
            $this->processManager->wait(function(int $workerIndex, int $status) {
                echo sprintf("Worker: child %d exited abnormally (status=%d), reforking\n",
                    $workerIndex, $status);
                // 重新 fork 一个子进程
                $this->processManager->fork(1, function(int $index) use ($workerIndex) {
                    $this->isChild = true;
                    $this->workerIndex = $workerIndex;

                    $this->loop = new EventLoop();
                    $this->signalHandler = new SignalHandler($this->loop);

                    $this->startAllServers();
                    $this->signalHandler->setupPeriodicCheck(1.0);

                    $this->running = true;
                    echo sprintf("Worker[%d]: event loop restarted\n", $workerIndex);
                    $this->loop->run();

                    $this->running = false;
                    $this->signalHandler->cleanup();
                    $this->stopAll();
                });
            });
        }
    }

    // 启动所有服务器(注册到 EventLoop)
    public function startAllServers(): void {
        $serverCount = count($this->servers);
        for ($i = 0; $i < $serverCount; $i++) {
            $server = $this->servers[$i];
            echo "Worker: starting server on " . $server->address
                . " (processes=" . $server->processCount . ")\n";
            $server->start($this->loop);
        }
    }

    // 停止所有服务器
    public function stopAll(): void {
        $serverCount = count($this->servers);
        for ($i = 0; $i < $serverCount; $i++) {
            $server = $this->servers[$i];
            if ($server->started) {
                $server->stop();
            }
        }
    }

    // 请求停止(可由信号处理或外部调用)
    public function stop(): void {
        echo sprintf("Worker: stop requested\n");
        $this->signalHandler->requestStop();
        $this->loop->stop();

        // 如果是父进程,终止所有子进程
        if (!$this->isChild && $this->processManager->isParent) {
            $this->processManager->terminateAll();
        }
    }
}