Compare commits

...

16 Commits

Author SHA1 Message Date
as2252258 3fe9ab06c3 Invalidate changed PHP files before worker reload 2026-07-08 14:34:38 +08:00
as2252258 a61e83ff6a Fix FileWatcher timer sentinel handling 2026-07-08 13:49:46 +08:00
as2252258 55d50b7bff Handle SIGINT after Swoole server start 2026-07-08 12:13:48 +08:00
as2252258 f6219a5c5d Avoid creating Swoole event loop for SIGINT 2026-07-08 12:04:04 +08:00
as2252258 acc966c3df eee 2026-07-08 11:58:04 +08:00
as2252258 614e943845 Fix server SIGINT handling under Swoole 2026-07-08 11:50:02 +08:00
as2252258 27ffa05515 eee 2026-07-08 11:39:27 +08:00
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
as2252258 908c8719d3 eee 2026-06-24 20:17:54 +08:00
as2252258 69b845d924 eee 2026-06-12 23:57:18 +08:00
as2252258 d9c8344a29 eee 2026-04-17 16:31:59 +08:00
12 changed files with 808 additions and 239 deletions
+12 -5
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,20 @@ 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); $this->terminateProcesses();
if ($this->server !== null) {
$this->server->shutdown();
}
return true; return true;
} }
/** /**
* @param array $service * @param array $service
* @param int $daemon * @param int $daemon
@@ -112,7 +119,7 @@ class AsyncServer extends Component implements ServerInterface
$settings[Constant::OPTION_DAEMONIZE] = (bool)$daemon; $settings[Constant::OPTION_DAEMONIZE] = (bool)$daemon;
$settings[Constant::OPTION_ENABLE_REUSE_PORT] = true; $settings[Constant::OPTION_ENABLE_REUSE_PORT] = true;
$settings[Constant::OPTION_PID_FILE] = storage('.swoole.pid'); $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'); $settings[Constant::OPTION_LOG_FILE] = storage('system.log');
} }
return $settings; return $settings;
+187 -67
View File
@@ -2,6 +2,7 @@
namespace Kiri\Server\Abstracts; namespace Kiri\Server\Abstracts;
use Kiri\Di\HotReloadState;
use Kiri\Di\Inject\Container; use Kiri\Di\Inject\Container;
use Kiri\Error\StdoutLogger; use Kiri\Error\StdoutLogger;
use Kiri\Events\EventProvider; use Kiri\Events\EventProvider;
@@ -9,13 +10,19 @@ use Kiri\Router\Router;
use Kiri\Server\Events\OnWorkerStart; use Kiri\Server\Events\OnWorkerStart;
use Kiri\Server\Processes\AbstractProcess; use Kiri\Server\Processes\AbstractProcess;
use Kiri\Server\Processes\OnProcessInterface; use Kiri\Server\Processes\OnProcessInterface;
use Kiri\Server\ServerInterface;
use Psr\Log\LoggerInterface; use Psr\Log\LoggerInterface;
use Swoole\Coroutine;
use Swoole\Event; use Swoole\Event;
use Swoole\Process; use Swoole\Process;
use Swoole\Timer;
use const SIGKILL;
use const SIGTERM;
use const SIGUSR1;
class FileWatcher extends AbstractProcess implements OnProcessInterface class FileWatcher extends AbstractProcess implements OnProcessInterface
{ {
use ReloadWorkers;
protected mixed $pipe; protected mixed $pipe;
#[Container(LoggerInterface::class)] #[Container(LoggerInterface::class)]
@@ -29,35 +36,40 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
protected bool $reloading = false; protected bool $reloading = false;
private array $watchPaths = []; private array $watchPaths = [];
private array $excludePatterns = []; private array $excludePatterns = [];
private array $extensions = ['php']; private array $extensions = ['php'];
private string $strategy; private string $strategy;
private bool $running = false; private bool $running = false;
private bool $scanning = false;
private $inotifyFd = null; private mixed $inotifyFd = null;
private array $inotifyWatchMap = []; private array $inotifyWatchMap = [];
private $fswatchProcess = null; private mixed $fswatchProcess = null;
private array $fswatchPipes = []; private array $fswatchPipes = [];
private int $pollInterval = 2; private int $pollInterval = 2;
private array $fileSnapshot = []; private array $fileSnapshot = [];
private $pollTimer = null; private int $pollTimer = -1;
private $debounceTimer = null; private int $debounceTimer = -1;
private int $debounceMs = 200; private int $debounceMs = 200;
private int $minReloadIntervalMs = 1000;
private int $lastReloadAtMs = 0;
private const STRATEGY_INOTIFY = 'inotify'; private const string STRATEGY_INOTIFY = 'inotify';
private const STRATEGY_FSWATCH = 'fswatch'; private const string STRATEGY_FSWATCH = 'fswatch';
private const STRATEGY_POLL = 'poll'; private const string STRATEGY_POLL = 'poll';
public function __construct() public function __construct()
{ {
$this->watchPaths = config('servers.reload.listen', []); $this->watchPaths = config('servers.reload.listen', []);
$this->excludePatterns = config('servers.reload.scan.skip_patterns', []); $this->excludePatterns = config('servers.reload.scan.skip_patterns', []);
$this->extensions = config('servers.reload.scan.extensions', ['php']); $this->extensions = config('servers.reload.scan.extensions', ['php']);
di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']); di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']);
$this->strategy = $this->detectBestStrategy(); $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();
} }
public function getName(): string public function getName(): string
@@ -107,7 +119,7 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
private function isFswatchAvailable(): bool private function isFswatchAvailable(): bool
{ {
$output = []; $output = [];
$returnVar = 0; $returnVar = 0;
exec('which fswatch 2>/dev/null', $output, $returnVar); exec('which fswatch 2>/dev/null', $output, $returnVar);
return $returnVar === 0 && !empty($output[0]); return $returnVar === 0 && !empty($output[0]);
@@ -127,6 +139,10 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
public function process(Process|null $process): void public function process(Process|null $process): void
{ {
if (class_exists('\\Swoole\\Coroutine')) {
Coroutine::set(['enable_deadlock_check' => false]);
}
if ($this->running) { if ($this->running) {
return; return;
} }
@@ -158,34 +174,66 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
$this->inotifyFd = null; $this->inotifyFd = null;
} }
if (isset($this->fswatchPipes[1]) && is_resource($this->fswatchPipes[1])) { foreach ($this->fswatchPipes as $pipe) {
@Event::del($this->fswatchPipes[1]); if (is_resource($pipe)) {
@fclose($this->fswatchPipes[1]); @Event::del($pipe);
@fclose($pipe);
}
}
$this->fswatchPipes = [];
$this->stopFswatchProcess();
if ($this->pollTimer > 0) {
Timer::clear($this->pollTimer);
$this->pollTimer = -1;
} }
if (isset($this->fswatchPipes[2]) && is_resource($this->fswatchPipes[2])) { if ($this->debounceTimer > 0) {
@fclose($this->fswatchPipes[2]); Timer::clear($this->debounceTimer);
} $this->debounceTimer = -1;
if ($this->fswatchProcess && is_resource($this->fswatchProcess)) {
@proc_terminate($this->fswatchProcess);
@proc_close($this->fswatchProcess);
$this->fswatchProcess = null;
}
if ($this->pollTimer) {
\Swoole\Timer::clear($this->pollTimer);
$this->pollTimer = null;
}
if ($this->debounceTimer) {
\Swoole\Timer::clear($this->debounceTimer);
$this->debounceTimer = null;
} }
@Event::exit(); @Event::exit();
} }
private function stopFswatchProcess(): void
{
if (!$this->fswatchProcess || !is_resource($this->fswatchProcess)) {
$this->fswatchProcess = null;
return;
}
$status = @proc_get_status($this->fswatchProcess);
if (($status['running'] ?? false) === true) {
@proc_terminate($this->fswatchProcess, SIGTERM);
for ($i = 0; $i < 10; $i++) {
usleep(100000);
$status = @proc_get_status($this->fswatchProcess);
if (($status['running'] ?? false) !== true) {
break;
}
}
}
$status = @proc_get_status($this->fswatchProcess);
if (($status['running'] ?? false) === true) {
@proc_terminate($this->fswatchProcess, SIGKILL);
for ($i = 0; $i < 5; $i++) {
usleep(100000);
$status = @proc_get_status($this->fswatchProcess);
if (($status['running'] ?? false) !== true) {
break;
}
}
}
$status = @proc_get_status($this->fswatchProcess);
if (($status['running'] ?? false) !== true) {
@proc_close($this->fswatchProcess);
}
$this->fswatchProcess = null;
}
private function isExcluded(string $path): bool private function isExcluded(string $path): bool
{ {
foreach ($this->excludePatterns as $pattern) { foreach ($this->excludePatterns as $pattern) {
@@ -209,20 +257,26 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
private function triggerCallback(array $changedFiles): void private function triggerCallback(array $changedFiles): void
{ {
$changedFiles = array_values(array_unique(array_filter($changedFiles, function ($file) { $changedFiles = array_filter($changedFiles, function ($file) {
return !$this->isExcluded($file) && $this->hasValidExtension($file); return !$this->isExcluded($file) && $this->hasValidExtension($file);
}))); })
|> array_unique(...)
|> array_values(...);
if (empty($changedFiles)) { if (empty($changedFiles)) {
return; return;
} }
if ($this->debounceTimer) { if ($this->debounceTimer > 0) {
\Swoole\Timer::clear($this->debounceTimer); Timer::clear($this->debounceTimer);
} }
$this->debounceTimer = \Swoole\Timer::after($this->debounceMs, function () use ($changedFiles) { $now = intdiv(hrtime(true), 1000000);
$this->debounceTimer = null; $elapsed = $this->lastReloadAtMs > 0 ? $now - $this->lastReloadAtMs : $this->minReloadIntervalMs;
$delay = max($this->debounceMs, $this->minReloadIntervalMs - $elapsed);
$this->debounceTimer = Timer::after($delay, function () use ($changedFiles) {
$this->debounceTimer = -1;
$this->reload($changedFiles); $this->reload($changedFiles);
}); });
} }
@@ -236,23 +290,69 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
$this->reloading = true; $this->reloading = true;
try { try {
$preview = implode(', ', array_slice($changedFiles, 0, 3)); $this->lastReloadAtMs = intdiv(hrtime(true), 1000000);
$this->invalidateChangedFiles($changedFiles);
$preview = implode(', ', array_slice($changedFiles, 0, 3));
if (count($changedFiles) > 3) { if (count($changedFiles) > 3) {
$preview .= ' ...'; $preview .= ' ...';
} }
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()) {
var_dump($server::class, get_class_methods($server)); di(StdoutLogger::class)->println('reload server failed: master pid not found or signal failed');
if (method_exists($server, 'reload')) {
$server->reload();
} }
} finally { } finally {
$this->reloading = false; $this->reloading = false;
} }
} }
private function invalidateChangedFiles(array $changedFiles): void
{
if (!function_exists('opcache_invalidate')) {
return;
}
foreach ($changedFiles as $file) {
if (!is_string($file) || !str_ends_with($file, '.php')) {
continue;
}
$path = realpath($file) ?: $file;
if (is_file($path)) {
@opcache_invalidate($path, true);
}
}
}
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 isProcessAlive(int $pid): bool
{
return $pid > 1 && @Process::kill($pid, 0);
}
private function startInotify(): void private function startInotify(): void
{ {
$this->inotifyFd = inotify_init(); $this->inotifyFd = inotify_init();
@@ -263,6 +363,10 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
} }
Event::add($this->inotifyFd, function ($fd) { Event::add($this->inotifyFd, function ($fd) {
if (!$this->running) {
return;
}
$events = inotify_read($fd); $events = inotify_read($fd);
if ($events === false) { if ($events === false) {
return; return;
@@ -270,9 +374,9 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
$changedFiles = []; $changedFiles = [];
foreach ($events as $event) { foreach ($events as $event) {
$wd = $event['wd']; $wd = $event['wd'];
$dir = $this->inotifyWatchMap[$wd] ?? ''; $dir = $this->inotifyWatchMap[$wd] ?? '';
$name = $event['name'] ?? ''; $name = $event['name'] ?? '';
$filePath = $name === '' ? $dir : $dir . '/' . $name; $filePath = $name === '' ? $dir : $dir . '/' . $name;
if (($event['mask'] & IN_CREATE) && $filePath !== '' && is_dir($filePath) && !$this->isExcluded($filePath)) { if (($event['mask'] & IN_CREATE) && $filePath !== '' && is_dir($filePath) && !$this->isExcluded($filePath)) {
@@ -296,7 +400,7 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
if (is_file($path)) { if (is_file($path)) {
if ($this->hasValidExtension($path)) { if ($this->hasValidExtension($path)) {
$wd = inotify_add_watch($this->inotifyFd, $path, IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVE | IN_CLOSE_WRITE); $wd = inotify_add_watch($this->inotifyFd, $path, IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVE | IN_CLOSE_WRITE);
$this->inotifyWatchMap[$wd] = dirname($path); $this->inotifyWatchMap[$wd] = dirname($path);
} }
return; return;
@@ -306,7 +410,7 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
return; return;
} }
$wd = inotify_add_watch($this->inotifyFd, $path, IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVE | IN_CLOSE_WRITE); $wd = inotify_add_watch($this->inotifyFd, $path, IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVE | IN_CLOSE_WRITE);
$this->inotifyWatchMap[$wd] = $path; $this->inotifyWatchMap[$wd] = $path;
$items = scandir($path); $items = scandir($path);
@@ -352,12 +456,16 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
stream_set_blocking($this->fswatchPipes[1], false); stream_set_blocking($this->fswatchPipes[1], false);
Event::add($this->fswatchPipes[1], function ($pipe) { Event::add($this->fswatchPipes[1], function ($pipe) {
if (!$this->running) {
return;
}
$data = fread($pipe, 8192); $data = fread($pipe, 8192);
if ($data === false || $data === '') { if ($data === false || $data === '') {
return; return;
} }
$files = explode("\0", trim($data, "\0")); $files = explode("\0", trim($data, "\0"));
$changedFiles = array_values(array_filter($files, function ($file) { $changedFiles = array_values(array_filter($files, function ($file) {
return $file !== ''; return $file !== '';
})); }));
@@ -370,15 +478,27 @@ class FileWatcher extends AbstractProcess implements OnProcessInterface
{ {
$this->buildFileSnapshot(); $this->buildFileSnapshot();
$intervalMs = $this->pollInterval * 1000; $intervalMs = $this->pollInterval * 1000;
$this->pollTimer = \Swoole\Timer::tick($intervalMs, function () { $this->pollTimer = Timer::tick($intervalMs, function () {
$newSnapshot = []; if (!$this->running || $this->scanning) {
$changedFiles = $this->compareSnapshots($newSnapshot); return;
if (!empty($changedFiles)) {
$this->triggerCallback($changedFiles);
} }
$this->fileSnapshot = $newSnapshot; $this->scanning = true;
try {
$newSnapshot = [];
$changedFiles = $this->compareSnapshots($newSnapshot);
if (!$this->running) {
return;
}
if (!empty($changedFiles)) {
$this->triggerCallback($changedFiles);
}
$this->fileSnapshot = $newSnapshot;
} finally {
$this->scanning = false;
}
}); });
} }
+166 -128
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;
@@ -16,156 +15,195 @@ use Kiri\Server\Processes\AbstractProcess;
class HotReload extends AbstractProcess class HotReload extends AbstractProcess
{ {
use ReloadWorkers;
/**
* @var mixed
*/
protected mixed $pipe;
/** /**
* @var LoggerInterface|StdoutLogger * @var mixed
*/ */
#[Container(LoggerInterface::class)] protected mixed $pipe;
public StdoutLogger|LoggerInterface $logger;
/** /**
* @var bool * @var LoggerInterface|StdoutLogger
*/ */
protected bool $enable_coroutine = false; #[Container(LoggerInterface::class)]
public StdoutLogger|LoggerInterface $logger;
/** /**
* @var array * @var bool
*/ */
protected array $watches = []; protected bool $enable_coroutine = false;
/** /**
* @var bool * @var array
*/ */
protected bool $enable_queue = false; protected array $watches = [];
/** /**
* @var bool * @var bool
*/ */
protected bool $reloading = false; protected bool $enable_queue = false;
/** /**
* @throws \Exception * @var bool
*/ */
public function __construct() protected bool $reloading = false;
{
di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']);
}
/** /**
* @return string * @throws \Exception
*/ */
public function getName(): string public function __construct()
{ {
return 'hotReload'; di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']);
} }
/** /**
* @return void * @return string
*/ */
public function onSigterm(): void public function getName(): string
{ {
// TODO: Implement onSigterm() method. return 'hotReload';
$this->stop(); }
}
/**
* @param ?Process $process
*/
public function process(Process|null $process): void
{
$this->pipe = inotify_init();
$this->addListen();
Event::add($this->pipe, function () use ($process) {
$read = inotify_read($this->pipe);
if (count($read) > 0) {
$this->reload();
}
});
Event::cycle(function (): void {
if ($this->isStop()) {
Event::exit();
}
});
Event::wait();
}
/** /**
* @return void * @return void
*/ */
public function reload(): void public function onSigterm(): void
{ {
if ($this->reloading) { $this->stop();
return; }
}
$this->reloading = true;
di(StdoutLogger::class)->println('reloading server[' . \config('site.id', 'system-service') . '], please waite.');
$this->clear();
di(ServerInterface::class)->reload();
$this->addListen();
$this->reloading = false;
}
/** public function stop(): void
* @return void {
*/ parent::stop();
protected function addListen(): void if (isset($this->pipe) && is_resource($this->pipe)) {
{ @Event::del($this->pipe);
foreach (config('servers.reload.listen') as $value) { @fclose($this->pipe);
$this->readDirectory($value); }
} @Event::exit();
} }
/**
* @param ?Process $process
*/
public function process(Process|null $process): void
{
$this->pipe = inotify_init();
$this->addListen();
Event::add($this->pipe, function () use ($process) {
$read = inotify_read($this->pipe);
if (count($read) > 0) {
$this->reload();
}
});
Event::cycle(function (): void {
if ($this->isStop()) {
Event::exit();
}
});
Event::wait();
}
/** /**
* @return void * @return void
*/ */
protected function clear(): void public function reload(): void
{ {
$this->watches = []; if ($this->reloading) {
} return;
}
$this->reloading = true;
/** di(StdoutLogger::class)->println('reloading server[' . \config('site.id', 'system-service') . '], please waite.');
* @param string $directory
* @return void $this->clear();
*/ if (!$this->reloadWorkers()) {
public function readFile(string $directory): void di(StdoutLogger::class)->println('reload server failed: master pid not found or signal failed');
{ }
if (str_ends_with($directory, '.php') === true) { $this->addListen();
inotify_add_watch($this->pipe, $directory, IN_MODIFY | IN_MOVE | IN_CREATE | IN_DELETE);
} $this->reloading = false;
} }
/** private function isMasterProcessCommand(string $command): bool
* @param string $directory {
* @return void $prefix = '[' . config('site.id', 'system-service') . '].Master[';
*/ return str_contains($command, $prefix);
public function readDirectory(string $directory): void }
{
foreach (glob($directory . '/*') as $data) { private function isSameAppProcess(int $pid): bool
if (is_dir($data)) { {
$this->readDirectory($data); $cwd = @readlink('/proc/' . $pid . '/cwd');
} else { if ($cwd === false || $cwd === '') {
$this->readFile($data); 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);
}
protected function addListen(): void
{
foreach (config('servers.reload.listen') as $value) {
$this->readDirectory($value);
}
}
/**
* @return void
*/
protected function clear(): void
{
$this->watches = [];
}
/**
* @param string $directory
* @return void
*/
public function readFile(string $directory): void
{
if (str_ends_with($directory, '.php') === true) {
inotify_add_watch($this->pipe, $directory, IN_MODIFY | IN_MOVE | IN_CREATE | IN_DELETE);
}
}
/**
* @param string $directory
* @return void
*/
public function readDirectory(string $directory): void
{
foreach (glob($directory . '/*') as $data) {
if (is_dir($data)) {
$this->readDirectory($data);
} else {
$this->readFile($data);
}
}
}
} }
+87
View File
@@ -0,0 +1,87 @@
<?php
namespace Kiri\Server\Abstracts;
use Swoole\Process;
trait ReloadWorkers
{
/**
* @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 ($candidates
|> array_filter(...)
|> array_unique(...)
|> array_values(...) 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 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;
}
}
+88 -1
View File
@@ -42,15 +42,99 @@ 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 {
if (!$this->shouldHandleSigint()) {
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);
} }
} }
private function shouldHandleSigint(): bool
{
$currentPid = getmypid();
if ($this->isMasterProcess($currentPid)) {
return true;
}
$masterPid = $this->getVerifiedPidFileMasterPid();
return $masterPid !== null && $masterPid === $currentPid;
}
private function getVerifiedPidFileMasterPid(): ?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 || !$this->isProcessAlive($pid) || !$this->isSameAppProcess($pid) || !$this->isMasterProcess($pid)) {
return null;
}
return $pid;
}
private function isMasterProcess(int $pid): bool
{
$command = $this->getProcessCommand($pid);
if ($command === '') {
return false;
}
$prefix = '[' . config('site.id', 'system-service') . '].Master[';
return str_contains($command, $prefix);
}
private function getProcessCommand(int $pid): string
{
$command = @file_get_contents('/proc/' . $pid . '/cmdline');
if (!is_string($command) || $command === '') {
return '';
}
return str_replace("\0", ' ', $command);
}
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);
}
/** /**
* @param $signal * @param $signal
* @param $callback * @param $callback
@@ -58,6 +142,9 @@ trait TraitServer
*/ */
private function onPcntlSignal($signal, $callback): void private function onPcntlSignal($signal, $callback): void
{ {
if (function_exists('pcntl_async_signals')) {
pcntl_async_signals(true);
}
\pcntl_signal($signal, $callback); \pcntl_signal($signal, $callback);
} }
+11 -6
View File
@@ -10,7 +10,9 @@ use Kiri\Server\Abstracts\Server;
use Kiri\Server\Events\OnBeforeShutdown; use Kiri\Server\Events\OnBeforeShutdown;
use Kiri\Server\Events\OnShutdown; use Kiri\Server\Events\OnShutdown;
use Kiri\Server\Events\OnStart; use Kiri\Server\Events\OnStart;
use Swoole\Server as SServer; use Swoole\Server as SServer;
use Swoole\Process;
use const SIGINT;
/** /**
@@ -32,11 +34,14 @@ class OnServer extends Server
* @param SServer $server * @param SServer $server
* @throws * @throws
*/ */
public function onStart(SServer $server): void public function onStart(SServer $server): void
{ {
Kiri::setProcessName(sprintf('Master[%d]', $server->master_pid)); Kiri::setProcessName(sprintf('Master[%d]', $server->master_pid));
Process::signal(SIGINT, static function () use ($server): void {
event(new OnStart($server)); $server->shutdown();
});
event(new OnStart($server));
} }
+1 -1
View File
@@ -74,7 +74,7 @@ class OnServerWorker extends Kiri\Server\Abstracts\Server
try { try {
$this->dispatch->dispatch(new OnBeforeWorkerStart($server, $workerId)); $this->dispatch->dispatch(new OnBeforeWorkerStart($server, $workerId));
if ($workerId < $server->setting['worker_num']) { if ($workerId < $server->setting['worker_num']) {
CoordinatorManager::utility(Coordinator::WORKER_START)->waite(); CoordinatorManager::utility(Coordinator::WORKER_START)->wait();
$this->dispatch->dispatch(new OnWorkerStart($server, $workerId)); $this->dispatch->dispatch(new OnWorkerStart($server, $workerId));
} else { } else {
+20 -7
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;
/** /**
@@ -16,6 +17,8 @@ abstract class AbstractProcess implements OnProcessInterface
private bool $stop = false; private bool $stop = false;
private bool $stopping = false;
public Process $process; public Process $process;
@@ -131,9 +134,11 @@ abstract class AbstractProcess implements OnProcessInterface
$array['deadlock_check_disable_trace'] = false; $array['deadlock_check_disable_trace'] = false;
$array['exit_condition'] = [$this, 'exit_condition']; $array['exit_condition'] = [$this, 'exit_condition'];
Coroutine::set($array); Coroutine::set($array);
$process::signal(SIGINT, static function (): void {});
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, static function (): void {});
} }
return $this; return $this;
} }
@@ -151,9 +156,7 @@ abstract class AbstractProcess implements OnProcessInterface
*/ */
public function pointWaitSignal($signal): void public function pointWaitSignal($signal): void
{ {
$this->stop = true; $this->requestStop();
$this->onSigterm();
} }
@@ -162,10 +165,20 @@ abstract class AbstractProcess implements OnProcessInterface
*/ */
public function coroutineWaitSignal(): void public function coroutineWaitSignal(): void
{ {
$value = Coroutine::waitSignal(SIGTERM); Coroutine::waitSignal(SIGTERM);
if ($value) { $this->requestStop();
$this->stop = true; }
private function requestStop(): void
{
if ($this->stopping) {
return;
} }
$this->stopping = true;
$this->stop = true;
$this->onSigterm(); $this->onSigterm();
} }
+38 -1
View File
@@ -5,6 +5,8 @@ namespace Kiri\Server\Processes;
use Exception; use Exception;
use Kiri; use Kiri;
use Swoole\Process; use Swoole\Process;
use const SIGKILL;
use const SIGTERM;
trait TraitProcess trait TraitProcess
{ {
@@ -66,4 +68,39 @@ trait TraitProcess
{ {
return $this->_process; return $this->_process;
} }
}
/**
* @return void
*/
private function terminateProcesses(): void
{
$pids = [];
foreach ($this->_process as $process) {
$pid = (int)($process->pid ?? 0);
if ($pid > 1 && $pid !== getmypid()) {
$pids[] = $pid;
}
}
$pids = array_values(array_unique($pids));
foreach ($pids as $pid) {
if (@Process::kill($pid, 0)) {
@Process::kill($pid, SIGTERM);
}
}
for ($i = 0; $i < 10; $i++) {
$alive = array_values(array_filter($pids, static fn(int $pid): bool => @Process::kill($pid, 0)));
if ($alive === []) {
return;
}
usleep(100000);
}
foreach ($pids as $pid) {
if (@Process::kill($pid, 0)) {
@Process::kill($pid, SIGKILL);
}
}
}
}
+21 -17
View File
@@ -125,22 +125,26 @@ class ServerCommand extends Command
} }
/** /**
* @param InputInterface $input * @param InputInterface $input
* @return int * @return int
* @throws * @throws
*/ */
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;
$this->asyncServer->addProcess([FileWatcher::class]); if ($hotReload) {
} else { $this->asyncServer->addProcess([FileWatcher::class]);
di(Router::class)->scan_build_route(); }
} if (!$hotReload) {
$this->asyncServer->initCoreServers(config('servers.server', []), (int)$input->getOption('daemon')); // 非热更模式可在 Master 进程预扫描,Worker 通过 fork 继承扫描结果。
$this->asyncServer->start(); // 热更模式必须让新 Worker 自己加载业务类,避免继承旧类定义后无法重新声明。
return 1; 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 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,180 @@ 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) || !$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);
}
} }
+1 -1
View File
@@ -9,7 +9,7 @@
} }
], ],
"require": { "require": {
"php": ">=8.4", "php": ">=8.5",
"ext-json": "*", "ext-json": "*",
"composer-runtime-api": "^2.0", "composer-runtime-api": "^2.0",
"psr/http-server-middleware": "^1.0", "psr/http-server-middleware": "^1.0",