Compare commits

...

3 Commits

Author SHA1 Message Date
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 242 additions and 21 deletions
+9 -4
View File
@@ -9,7 +9,6 @@ use Kiri\Exception\NotFindClassException;
use Kiri\Server\Config as SConfig; use Kiri\Server\Config as SConfig;
use Kiri\Server\Constant; use Kiri\Server\Constant;
use Kiri\Server\Events\OnServerBeforeStart; use Kiri\Server\Events\OnServerBeforeStart;
use Kiri\Server\Events\OnShutdown;
use Kiri\Server\Handler\OnServer; use Kiri\Server\Handler\OnServer;
use Kiri\Server\Processes\TraitProcess; use Kiri\Server\Processes\TraitProcess;
use Kiri\Server\ServerInterface; use Kiri\Server\ServerInterface;
@@ -32,6 +31,8 @@ class AsyncServer extends Component implements ServerInterface
*/ */
private ?Server $server = null; private ?Server $server = null;
private bool $shuttingDown = false;
/** /**
* @param array $service * @param array $service
@@ -61,14 +62,18 @@ class AsyncServer extends Component implements ServerInterface
*/ */
public function shutdown(): bool 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; return true;
} }
/** /**
* @param array $service * @param array $service
* @param int $daemon * @param int $daemon
+26 -3
View File
@@ -245,15 +245,38 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
di(HotReloadState::class)->store($changedFiles); di(HotReloadState::class)->store($changedFiles);
di(StdoutLogger::class)->println('detected file changes, reloading server: ' . $preview); di(StdoutLogger::class)->println('detected file changes, reloading server: ' . $preview);
$server = di(ServerInterface::class); if (!$this->reloadWorkers()) {
if (method_exists($server, 'reload')) { di(StdoutLogger::class)->println('reload server failed: master pid not found or signal failed');
$server->reload();
} }
} finally { } finally {
$this->reloading = false; $this->reloading = false;
} }
} }
private function reloadWorkers(): bool
{
$pid = $this->getMasterPid();
if ($pid === null) {
return false;
}
return Process::kill($pid, SIGUSR1);
}
private function getMasterPid(): ?int
{
$pidFile = function_exists('storage') ? storage('.swoole.pid') : '';
if (!is_file($pidFile)) {
return null;
}
$pid = (int)trim((string)file_get_contents($pidFile));
if ($pid <= 1 || !Process::kill($pid, 0)) {
return null;
}
return $pid;
}
private function startInotify(): void private function startInotify(): void
{ {
$this->inotifyFd = inotify_init(); $this->inotifyFd = inotify_init();
+27 -2
View File
@@ -7,7 +7,6 @@ use Kiri\Error\StdoutLogger;
use Kiri\Events\EventProvider; use Kiri\Events\EventProvider;
use Kiri\Router\Router; use Kiri\Router\Router;
use Kiri\Server\Events\OnWorkerStart; use Kiri\Server\Events\OnWorkerStart;
use Kiri\Server\ServerInterface;
use Psr\Log\LoggerInterface; use Psr\Log\LoggerInterface;
use Swoole\Event; use Swoole\Event;
use Swoole\Process; use Swoole\Process;
@@ -116,7 +115,9 @@ class HotReload extends AbstractProcess
di(StdoutLogger::class)->println('reloading server[' . \config('site.id', 'system-service') . '], please waite.'); di(StdoutLogger::class)->println('reloading server[' . \config('site.id', 'system-service') . '], please waite.');
$this->clear(); $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->addListen();
$this->reloading = false; $this->reloading = false;
@@ -126,6 +127,30 @@ class HotReload extends AbstractProcess
/** /**
* @return void * @return void
*/ */
private function reloadWorkers(): bool
{
$pid = $this->getMasterPid();
if ($pid === null) {
return false;
}
return Process::kill($pid, SIGUSR1);
}
private function getMasterPid(): ?int
{
$pidFile = function_exists('storage') ? storage('.swoole.pid') : '';
if (!is_file($pidFile)) {
return null;
}
$pid = (int)trim((string)file_get_contents($pidFile));
if ($pid <= 1 || !Process::kill($pid, 0)) {
return null;
}
return $pid;
}
protected function addListen(): void protected function addListen(): void
{ {
foreach (config('servers.reload.listen') as $value) { foreach (config('servers.reload.listen') as $value) {
+13 -2
View File
@@ -42,15 +42,26 @@ trait TraitServer
*/ */
public function onSigint($no, array $signInfo): void public function onSigint($no, array $signInfo): void
{ {
static $handling = false;
if ($handling) {
return;
}
$handling = true;
try { 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); Kiri::getLogger()->alert('Pid ' . getmypid() . ' get signo ' . $no);
$this->shutdown(); $this->shutdown();
} catch (\Throwable $exception) { } catch (\Throwable $exception) {
\Kiri::getLogger()->json_log($exception); \Kiri::getLogger()->json_log($exception);
} }
} }
/** /**
* @param $signal * @param $signal
* @param $callback * @param $callback
+3 -1
View File
@@ -6,6 +6,7 @@ namespace Kiri\Server\Processes;
use Swoole\Coroutine; use Swoole\Coroutine;
use Swoole\Process; use Swoole\Process;
use const SIGHUP; use const SIGHUP;
use const SIGINT;
use const SIGTERM; use const SIGTERM;
/** /**
@@ -133,7 +134,8 @@ abstract class AbstractProcess implements OnProcessInterface
Coroutine::set($array); Coroutine::set($array);
Coroutine::create(fn() => $this->coroutineWaitSignal()); Coroutine::create(fn() => $this->coroutineWaitSignal());
} else { } else {
$process::signal(SIGTERM | SIGINT | SIGUSR1, [$this, 'pointWaitSignal']); $process::signal(SIGTERM, [$this, 'pointWaitSignal']);
$process::signal(SIGINT, [$this, 'pointWaitSignal']);
} }
return $this; return $this;
} }
+7 -4
View File
@@ -133,12 +133,15 @@ class ServerCommand extends Command
protected function start(InputInterface $input): int protected function start(InputInterface $input): int
{ {
$this->asyncServer->addProcess(config('process', [])); $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]); $this->asyncServer->addProcess([FileWatcher::class]);
} }
// Master 进程在 fork 前完成首次扫描,Worker 启动时不再重复全量扫描 if (!$hotReload) {
// 避免每个 Worker 独立执行 opcache_compile_file + invalidateClasses 造成 OOM // 非热更模式可在 Master 进程预扫描,Worker 通过 fork 继承扫描结果。
di(Router::class)->scan_build_route(); // 热更模式必须让新 Worker 自己加载业务类,避免继承旧类定义后无法重新声明。
di(Router::class)->scan_build_route();
}
$this->asyncServer->initCoreServers(config('servers.server', []), (int)$input->getOption('daemon')); $this->asyncServer->initCoreServers(config('servers.server', []), (int)$input->getOption('daemon'));
$this->asyncServer->start(); $this->asyncServer->start();
return 1; return 1;
+157 -5
View File
@@ -31,9 +31,13 @@ class State extends Component
*/ */
public function isRunner(): bool public function isRunner(): bool
{ {
if ($this->getPidFilePid() !== null || $this->getNamedServerPids() !== []) {
return true;
}
$ports = $this->sortService($this->servers); $ports = $this->sortService($this->servers);
foreach ($ports as $config) { foreach ($ports as $config) {
if (checkPortIsAlready($config['port'])) { if (checkPortIsAlready($config->port)) {
return true; return true;
} }
} }
@@ -47,13 +51,161 @@ class State extends Component
*/ */
public function exit($port): void public function exit($port): void
{ {
if (!($pid = checkPortIsAlready($port))) { $pids = $this->collectStopPids((int)$port);
if ($pids === []) {
return; return;
} }
while (checkPortIsAlready($port)) {
Process::kill($pid, 0) && Process::kill($pid, SIGTERM); $this->terminatePids($pids, SIGTERM, 30);
usleep(300);
$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)) {
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;
}
$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 isProcessAlive(int $pid): bool
{
return $pid > 1 && @Process::kill($pid, 0);
}
} }