Compare commits

...
6 Commits
Author SHA1 Message Date
as2252258 1f443958e8 eee 2026-07-08 10:46:00 +08:00
as2252258 77e9333651 eee 2026-07-08 10:45:21 +08:00
as2252258 56ebbd2a2e eee 2026-07-08 10:08:10 +08:00
as2252258 7f36bb0d53 eee 2026-07-08 09:42:56 +08:00
as2252258 0adfe14a1f eee 2026-07-07 22:59:28 +08:00
as2252258 3ca39d2205 eee 2026-07-03 14:25:26 +08:00
7 changed files with 429 additions and 24 deletions
+10 -5
View File
@@ -9,7 +9,6 @@ use Kiri\Exception\NotFindClassException;
use Kiri\Server\Config as SConfig;
use Kiri\Server\Constant;
use Kiri\Server\Events\OnServerBeforeStart;
use Kiri\Server\Events\OnShutdown;
use Kiri\Server\Handler\OnServer;
use Kiri\Server\Processes\TraitProcess;
use Kiri\Server\ServerInterface;
@@ -32,6 +31,8 @@ class AsyncServer extends Component implements ServerInterface
*/
private ?Server $server = null;
private bool $shuttingDown = false;
/**
* @param array $service
@@ -61,14 +62,18 @@ class AsyncServer extends Component implements ServerInterface
*/
public function shutdown(): bool
{
$this->server->shutdown();
if ($this->shuttingDown) {
return true;
}
$this->shuttingDown = true;
$this->dispatch->dispatch(new OnShutdown);
if ($this->server !== null) {
$this->server->shutdown();
}
return true;
}
/**
* @param array $service
* @param int $daemon
@@ -112,7 +117,7 @@ class AsyncServer extends Component implements ServerInterface
$settings[Constant::OPTION_DAEMONIZE] = (bool)$daemon;
$settings[Constant::OPTION_ENABLE_REUSE_PORT] = true;
$settings[Constant::OPTION_PID_FILE] = storage('.swoole.pid');
if (!isset($settings[Constant::OPTION_PID_FILE])) {
if (!isset($settings[Constant::OPTION_LOG_FILE])) {
$settings[Constant::OPTION_LOG_FILE] = storage('system.log');
}
return $settings;
+118 -5
View File
@@ -10,10 +10,10 @@ use Kiri\Router\Router;
use Kiri\Server\Events\OnWorkerStart;
use Kiri\Server\Processes\AbstractProcess;
use Kiri\Server\Processes\OnProcessInterface;
use Kiri\Server\ServerInterface;
use Psr\Log\LoggerInterface;
use Swoole\Event;
use Swoole\Process;
use const SIGUSR1;
class FileWatcher extends AbstractProcess implements OnProcessInterface
{
@@ -45,6 +45,8 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
private $pollTimer = null;
private $debounceTimer = null;
private int $debounceMs = 200;
private int $minReloadIntervalMs = 1000;
private int $lastReloadAtMs = 0;
private const STRATEGY_INOTIFY = 'inotify';
private const STRATEGY_FSWATCH = 'fswatch';
@@ -58,6 +60,8 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']);
$this->debounceMs = (int)config('servers.reload.debounce_ms', $this->debounceMs);
$this->minReloadIntervalMs = (int)config('servers.reload.min_interval_ms', $this->minReloadIntervalMs);
$this->strategy = $this->detectBestStrategy();
}
@@ -222,7 +226,11 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
\Swoole\Timer::clear($this->debounceTimer);
}
$this->debounceTimer = \Swoole\Timer::after($this->debounceMs, function () use ($changedFiles) {
$now = intdiv(hrtime(true), 1000000);
$elapsed = $this->lastReloadAtMs > 0 ? $now - $this->lastReloadAtMs : $this->minReloadIntervalMs;
$delay = max($this->debounceMs, $this->minReloadIntervalMs - $elapsed);
$this->debounceTimer = \Swoole\Timer::after($delay, function () use ($changedFiles) {
$this->debounceTimer = null;
$this->reload($changedFiles);
});
@@ -237,6 +245,7 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
$this->reloading = true;
try {
$this->lastReloadAtMs = intdiv(hrtime(true), 1000000);
$preview = implode(', ', array_slice($changedFiles, 0, 3));
if (count($changedFiles) > 3) {
$preview .= ' ...';
@@ -245,15 +254,119 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
di(HotReloadState::class)->store($changedFiles);
di(StdoutLogger::class)->println('detected file changes, reloading server: ' . $preview);
$server = di(ServerInterface::class);
if (method_exists($server, 'reload')) {
$server->reload();
if (!$this->reloadWorkers()) {
di(StdoutLogger::class)->println('reload server failed: master pid not found or signal failed');
}
} finally {
$this->reloading = false;
}
}
private function reloadWorkers(): bool
{
$pid = $this->getMasterPid();
if ($pid === null) {
return false;
}
return Process::kill($pid, SIGUSR1);
}
private function getMasterPid(): ?int
{
$processes = $this->listProcesses();
$candidates = [];
$pidFile = function_exists('storage') ? storage('.swoole.pid') : '';
if (is_file($pidFile)) {
$candidates[] = (int)trim((string)file_get_contents($pidFile));
}
foreach ($processes as $process) {
if ($this->isMasterProcessCommand($process['command'])) {
$candidates[] = (int)$process['pid'];
}
}
foreach (array_values(array_unique(array_filter($candidates))) as $pid) {
if ($this->isProcessAlive($pid) && $this->isMasterPid($pid, $processes)) {
return $pid;
}
}
return null;
}
/**
* @param int $pid
* @param array $processes
* @return bool
*/
private function isMasterPid(int $pid, array $processes): bool
{
foreach ($processes as $process) {
if ((int)$process['pid'] === $pid) {
return $this->isMasterProcessCommand($process['command']) && $this->isSameAppProcess($pid);
}
}
return false;
}
private function isMasterProcessCommand(string $command): bool
{
$prefix = '[' . config('site.id', 'system-service') . '].Master[';
return str_contains($command, $prefix);
}
private function isSameAppProcess(int $pid): bool
{
$cwd = @readlink('/proc/' . $pid . '/cwd');
if ($cwd === false || $cwd === '') {
return false;
}
return $this->normalizePath($cwd) === $this->normalizePath(APP_PATH);
}
private function normalizePath(string $path): string
{
$real = realpath($path) ?: $path;
return rtrim(str_replace('\\', '/', $real), '/');
}
private function listProcesses(): array
{
$lines = [];
exec('ps -eo pid=,ppid=,args=', $lines);
$processes = [];
foreach ($lines as $line) {
$line = trim($line);
if ($line === '') {
continue;
}
$parts = preg_split('/\s+/', $line, 3);
if (count($parts) < 3) {
continue;
}
$processes[] = [
'pid' => (int)$parts[0],
'ppid' => (int)$parts[1],
'command' => $parts[2],
];
}
return $processes;
}
private function isProcessAlive(int $pid): bool
{
return $pid > 1 && @Process::kill($pid, 0);
}
private function startInotify(): void
{
$this->inotifyFd = inotify_init();
+102 -2
View File
@@ -7,7 +7,6 @@ use Kiri\Error\StdoutLogger;
use Kiri\Events\EventProvider;
use Kiri\Router\Router;
use Kiri\Server\Events\OnWorkerStart;
use Kiri\Server\ServerInterface;
use Psr\Log\LoggerInterface;
use Swoole\Event;
use Swoole\Process;
@@ -116,7 +115,9 @@ class HotReload extends AbstractProcess
di(StdoutLogger::class)->println('reloading server[' . \config('site.id', 'system-service') . '], please waite.');
$this->clear();
di(ServerInterface::class)->reload();
if (!$this->reloadWorkers()) {
di(StdoutLogger::class)->println('reload server failed: master pid not found or signal failed');
}
$this->addListen();
$this->reloading = false;
@@ -126,6 +127,105 @@ class HotReload extends AbstractProcess
/**
* @return void
*/
private function reloadWorkers(): bool
{
$pid = $this->getMasterPid();
if ($pid === null) {
return false;
}
return Process::kill($pid, SIGUSR1);
}
private function getMasterPid(): ?int
{
$processes = $this->listProcesses();
$candidates = [];
$pidFile = function_exists('storage') ? storage('.swoole.pid') : '';
if (is_file($pidFile)) {
$candidates[] = (int)trim((string)file_get_contents($pidFile));
}
foreach ($processes as $process) {
if ($this->isMasterProcessCommand($process['command'])) {
$candidates[] = (int)$process['pid'];
}
}
foreach (array_values(array_unique(array_filter($candidates))) as $pid) {
if ($this->isProcessAlive($pid) && $this->isMasterPid($pid, $processes)) {
return $pid;
}
}
return null;
}
private function isMasterPid(int $pid, array $processes): bool
{
foreach ($processes as $process) {
if ((int)$process['pid'] === $pid) {
return $this->isMasterProcessCommand($process['command']) && $this->isSameAppProcess($pid);
}
}
return false;
}
private function isMasterProcessCommand(string $command): bool
{
$prefix = '[' . config('site.id', 'system-service') . '].Master[';
return str_contains($command, $prefix);
}
private function isSameAppProcess(int $pid): bool
{
$cwd = @readlink('/proc/' . $pid . '/cwd');
if ($cwd === false || $cwd === '') {
return false;
}
return $this->normalizePath($cwd) === $this->normalizePath(APP_PATH);
}
private function normalizePath(string $path): string
{
$real = realpath($path) ?: $path;
return rtrim(str_replace('\\', '/', $real), '/');
}
private function listProcesses(): array
{
$lines = [];
exec('ps -eo pid=,ppid=,args=', $lines);
$processes = [];
foreach ($lines as $line) {
$line = trim($line);
if ($line === '') {
continue;
}
$parts = preg_split('/\s+/', $line, 3);
if (count($parts) < 3) {
continue;
}
$processes[] = [
'pid' => (int)$parts[0],
'ppid' => (int)$parts[1],
'command' => $parts[2],
];
}
return $processes;
}
private function isProcessAlive(int $pid): bool
{
return $pid > 1 && @Process::kill($pid, 0);
}
protected function addListen(): void
{
foreach (config('servers.reload.listen') as $value) {
+12 -1
View File
@@ -42,7 +42,19 @@ trait TraitServer
*/
public function onSigint($no, array $signInfo): void
{
static $handling = false;
if ($handling) {
return;
}
$handling = true;
try {
$pidFile = function_exists('storage') ? storage('.swoole.pid') : '';
$masterPid = is_file($pidFile) ? (int)trim((string)file_get_contents($pidFile)) : 0;
if ($masterPid > 1 && $masterPid !== getmypid()) {
return;
}
Kiri::getLogger()->alert('Pid ' . getmypid() . ' get signo ' . $no);
$this->shutdown();
} catch (\Throwable $exception) {
@@ -50,7 +62,6 @@ trait TraitServer
}
}
/**
* @param $signal
* @param $callback
+3 -1
View File
@@ -6,6 +6,7 @@ namespace Kiri\Server\Processes;
use Swoole\Coroutine;
use Swoole\Process;
use const SIGHUP;
use const SIGINT;
use const SIGTERM;
/**
@@ -133,7 +134,8 @@ abstract class AbstractProcess implements OnProcessInterface
Coroutine::set($array);
Coroutine::create(fn() => $this->coroutineWaitSignal());
} else {
$process::signal(SIGTERM | SIGINT | SIGUSR1, [$this, 'pointWaitSignal']);
$process::signal(SIGTERM, [$this, 'pointWaitSignal']);
$process::signal(SIGINT, [$this, 'pointWaitSignal']);
}
return $this;
}
+6 -3
View File
@@ -133,12 +133,15 @@ class ServerCommand extends Command
protected function start(InputInterface $input): int
{
$this->asyncServer->addProcess(config('process', []));
if (\config('servers.reload.hot', false) === true) {
$hotReload = \config('servers.reload.hot', false) === true;
if ($hotReload) {
$this->asyncServer->addProcess([FileWatcher::class]);
}
// Master 进程在 fork 前完成首次扫描,Worker 启动时不再重复全量扫描
// 避免每个 Worker 独立执行 opcache_compile_file + invalidateClasses 造成 OOM
if (!$hotReload) {
// 非热更模式可在 Master 进程预扫描,Worker 通过 fork 继承扫描结果。
// 热更模式必须让新 Worker 自己加载业务类,避免继承旧类定义后无法重新声明。
di(Router::class)->scan_build_route();
}
$this->asyncServer->initCoreServers(config('servers.server', []), (int)$input->getOption('daemon'));
$this->asyncServer->start();
return 1;
+176 -5
View File
@@ -31,9 +31,13 @@ class State extends Component
*/
public function isRunner(): bool
{
if ($this->getPidFilePid() !== null || $this->getNamedServerPids() !== []) {
return true;
}
$ports = $this->sortService($this->servers);
foreach ($ports as $config) {
if (checkPortIsAlready($config['port'])) {
if (checkPortIsAlready($config->port)) {
return true;
}
}
@@ -47,13 +51,180 @@ class State extends Component
*/
public function exit($port): void
{
if (!($pid = checkPortIsAlready($port))) {
$pids = $this->collectStopPids((int)$port);
if ($pids === []) {
return;
}
while (checkPortIsAlready($port)) {
Process::kill($pid, 0) && Process::kill($pid, SIGTERM);
usleep(300);
$this->terminatePids($pids, SIGTERM, 30);
$remaining = array_values(array_filter($pids, [$this, 'isProcessAlive']));
if ($remaining !== []) {
$this->terminatePids($remaining, SIGKILL, 10);
}
if (!checkPortIsAlready((int)$port)) {
@unlink(storage('.swoole.pid'));
}
}
private function collectStopPids(int $port): array
{
$pids = [];
$pidFilePid = $this->getPidFilePid();
if ($pidFilePid !== null) {
$pids[] = $pidFilePid;
$pids = array_merge($pids, $this->getDescendantPids($pidFilePid));
}
$pids = array_merge($pids, $this->getNamedServerPids());
$portPid = checkPortIsAlready($port);
if ($portPid !== false) {
$pids[] = (int)$portPid;
$pids = array_merge($pids, $this->getDescendantPids((int)$portPid));
}
$currentPid = getmypid();
$pids = array_values(array_unique(array_filter(array_map('intval', $pids), function (int $pid) use ($currentPid) {
return $pid > 1 && $pid !== $currentPid;
})));
rsort($pids);
return $pids;
}
private function terminatePids(array $pids, int $signal, int $rounds): void
{
for ($i = 0; $i < $rounds; $i++) {
$alive = false;
foreach ($pids as $pid) {
if (!$this->isProcessAlive($pid)) {
continue;
}
$alive = true;
@Process::kill($pid, $signal);
}
if (!$alive) {
return;
}
usleep(100000);
}
}
private function getPidFilePid(): ?int
{
$pidFile = storage('.swoole.pid');
if (!is_file($pidFile)) {
return null;
}
$pid = (int)trim((string)file_get_contents($pidFile));
if ($pid <= 1 || !$this->isProcessAlive($pid) || !$this->isSameAppProcess($pid)) {
return null;
}
return $pid;
}
private function getDescendantPids(int $parentPid): array
{
$processes = $this->listProcesses();
$children = [];
foreach ($processes as $process) {
$children[(int)$process['ppid']][] = (int)$process['pid'];
}
$result = [];
$stack = [$parentPid];
while ($stack !== []) {
$pid = array_pop($stack);
foreach ($children[$pid] ?? [] as $childPid) {
$result[] = $childPid;
$stack[] = $childPid;
}
}
return $result;
}
private function getNamedServerPids(): array
{
$siteId = '[' . config('site.id', 'system-service') . ']';
$pids = [];
foreach ($this->listProcesses() as $process) {
$command = $process['command'];
if (!str_contains($command, $siteId)) {
continue;
}
if (!preg_match('/\\]\\..+\\[[0-9]+\\]/', $command)) {
continue;
}
if (!$this->isSameAppProcess((int)$process['pid'])) {
continue;
}
$pids[] = (int)$process['pid'];
}
return $pids;
}
private function listProcesses(): array
{
$lines = [];
exec('ps -eo pid=,ppid=,args=', $lines);
$processes = [];
foreach ($lines as $line) {
$line = trim($line);
if ($line === '') {
continue;
}
$parts = preg_split('/\s+/', $line, 3);
if (count($parts) < 3) {
continue;
}
$processes[] = [
'pid' => (int)$parts[0],
'ppid' => (int)$parts[1],
'command' => $parts[2],
];
}
return $processes;
}
private function isSameAppProcess(int $pid): bool
{
$cwd = @readlink('/proc/' . $pid . '/cwd');
if ($cwd === false || $cwd === '') {
return false;
}
return $this->normalizePath($cwd) === $this->normalizePath(APP_PATH);
}
private function normalizePath(string $path): string
{
$real = realpath($path) ?: $path;
return rtrim(str_replace('\\', '/', $real), '/');
}
private function isProcessAlive(int $pid): bool
{
return $pid > 1 && @Process::kill($pid, 0);
}
}