diff --git a/etc/bootstrap/signals.php.template b/etc/bootstrap/signals.php.template new file mode 100644 index 0000000..bc279d5 --- /dev/null +++ b/etc/bootstrap/signals.php.template @@ -0,0 +1,16 @@ + $channel */ +$channel = Channel::open('{{CHANNEL}}'); + +SignalListener::listen($channel); + +unlink(__FILE__); diff --git a/etc/qa/composer-require-checker.json b/etc/qa/composer-require-checker.json index 040b1e2..b9d7347 100644 --- a/etc/qa/composer-require-checker.json +++ b/etc/qa/composer-require-checker.json @@ -2,14 +2,15 @@ "symbol-whitelist" : [ "null", "true", "false", "static", "self", "parent", - "array", "string", "int", "float", "bool", "iterable", "callable", "void", "object", "mixed", - "Safe\\json_encode", "futurePromise", "WyriHaximus\\Constants\\ComposerAutoloader\\LOCATION" + "array", "string", "int", "float", "bool", "iterable", "callable", "void", "object", "mixed" ], "php-core-extensions" : [ "Core", "date", + "pcntl", "pcre", "Phar", + "posix", "Reflection", "SPL", "standard" diff --git a/src/Runtime.php b/src/Runtime.php index d8eb053..03161de 100644 --- a/src/Runtime.php +++ b/src/Runtime.php @@ -5,15 +5,30 @@ namespace ReactParallel\Runtime; use Closure; +use parallel\Channel; use parallel\Runtime as ParallelRuntime; +use React\EventLoop\Loop; use ReactParallel\EventLoop\EventLoopBridge; +use function file_get_contents; +use function file_put_contents; +use function posix_getpid; +use function posix_kill; +use function React\Async\async; +use function str_replace; +use function sys_get_temp_dir; +use function tempnam; +use function uniqid; + use const WyriHaximus\Constants\ComposerAutoloader\LOCATION; final readonly class Runtime { private ParallelRuntime $runtime; + /** @var Channel */ + private Channel $signalChannel; + /** @api */ public static function create(EventLoopBridge $eventLoopBridge): self { @@ -22,7 +37,11 @@ public static function create(EventLoopBridge $eventLoopBridge): self public function __construct(private EventLoopBridge $eventLoopBridge, string $autoload) { - $this->runtime = new ParallelRuntime($autoload); + $channelName = '.well-known/react-parallel/runtime/signals/' . uniqid('', true); + /** @phpstan-ignore assign.propertyType */ + $this->signalChannel = Channel::make($channelName, Channel::Infinite); + $this->runtime = new ParallelRuntime($this->createBootstrapFile($autoload, $channelName)); + $this->forwardSignals(); } /** @@ -72,4 +91,31 @@ public function kill(): void { $this->runtime->kill(); } + + private function forwardSignals(): void + { + Loop::addTimer(0.0, async(function (): void { + foreach ($this->eventLoopBridge->observe($this->signalChannel) as $signal) { + /** @phpstan-ignore property.nonObject, argument.type */ + posix_kill(posix_getpid(), $signal->number); + } + })); + } + + private function createBootstrapFile(string $autoload, string $channelName): string + { + $file = tempnam(sys_get_temp_dir(), 'react_parallel_'); + /** @phpstan-ignore wyrihaximus.reactphp.blocking.function.filePutContents */ + file_put_contents( + $file, + str_replace( + ['{{AUTOLOAD}}', '{{CHANNEL}}'], + [$autoload, $channelName], + /** @phpstan-ignore wyrihaximus.reactphp.blocking.function.fileGetContents, argument.type */ + file_get_contents(__DIR__ . '/../etc/bootstrap/signals.php.template'), + ), + ); + + return $file; + } } diff --git a/src/Signal.php b/src/Signal.php new file mode 100644 index 0000000..669a883 --- /dev/null +++ b/src/Signal.php @@ -0,0 +1,40 @@ + $channel + * + * @api + */ + public static function listen(Channel $channel): void + { + pcntl_async_signals(true); + + $channel = $channel; + pcntl_signal(SIGINT, static function () use ($channel): void { + $channel->send(Signal::sigint()); + }); + pcntl_signal(SIGTERM, static function () use ($channel): void { + $channel->send(Signal::sigterm()); + }); + pcntl_signal(SIGHUP, static function () use ($channel): void { + $channel->send(Signal::sighup()); + }); + pcntl_signal(SIGQUIT, static function () use ($channel): void { + $channel->send(Signal::sigquit()); + }); + } +} diff --git a/tests/Functional/signal_forwarding_128_threads.php b/tests/Functional/signal_forwarding_128_threads.php new file mode 100644 index 0000000..15feb38 --- /dev/null +++ b/tests/Functional/signal_forwarding_128_threads.php @@ -0,0 +1,53 @@ +#!/usr/bin/env php +close(); + foreach ($parallelRuntimes as $parallelRuntime) { + $parallelRuntime->close(); + } + + Loop::stop(); +}); + +// Schedule sleep tasks on all parallel runtimes to keep them alive +foreach ($parallelRuntimes as $parallelRuntime) { + $parallelRuntime->run(static function (): void { + /** @phpstan-ignore wyrihaximus.reactphp.blocking.function.sleep */ + sleep(60); + }); +} + +echo 'Ready with ' . $threadCount . ' threads' . "\n"; + +// Run the event loop - it will stay alive until Loop::stop() is called by the signal handler +Loop::run(); + +echo 'Exited cleanly' . "\n"; diff --git a/tests/FunctionalSignalTest.php b/tests/FunctionalSignalTest.php new file mode 100644 index 0000000..6f70471 --- /dev/null +++ b/tests/FunctionalSignalTest.php @@ -0,0 +1,80 @@ + ['pipe', 'r'], + 1 => ['pipe', 'w'], + 2 => ['pipe', 'w'], + ]; + + $process = proc_open( + 'php ' . __DIR__ . '/Functional/signal_forwarding_128_threads.php', + $descriptorSpec, + $pipes, + ); + + /** @phpstan-ignore staticMethod.alreadyNarrowedType */ + self::assertNotNull($process, 'Process should start'); + + // Read stdout until we see "Ready" + $output = ''; + $startTime = microtime(true); + $status = null; + while (microtime(true) - $startTime < 120) { // 2 minute timeout + /** @phpstan-ignore argument.type */ + $status = proc_get_status($process); + if (! $status['running']) { + break; + } + + $line = fgets($pipes[1]); + if ($line === false) { + continue; + } + + $output .= $line; + if (str_contains($output, 'Ready with 128 threads')) { + break; + } + } + + self::assertStringContainsString('Ready with 128 threads', $output, 'Process should report 128 threads ready'); + + // Send SIGTERM to the process + /** @phpstan-ignore offsetAccess.notFound */ + posix_kill($status['pid'], SIGTERM); + + // Wait for the process to exit and collect remaining output + /** @phpstan-ignore argument.type */ + $exitCode = proc_close($process); + $output .= stream_get_contents($pipes[1]); + + self::assertSame(0, $exitCode, 'Process should exit cleanly'); + self::assertStringContainsString('SIGTERM received', $output, 'Output should confirm SIGTERM was received'); + self::assertStringContainsString('Exited cleanly', $output, 'Output should confirm clean exit'); + } +} diff --git a/tests/RuntimeTest.php b/tests/RuntimeTest.php index 3f9a36b..77db97c 100644 --- a/tests/RuntimeTest.php +++ b/tests/RuntimeTest.php @@ -13,10 +13,17 @@ use TheOrville\Exceptions\LatchcombException; use WyriHaximus\AsyncTestUtilities\AsyncTestCase; +use function extension_loaded; +use function pcntl_async_signals; +use function pcntl_signal; +use function posix_getpid; +use function posix_kill; use function React\Async\await; use function sleep; use function WyriHaximus\React\timedPromise; +use const SIGTERM; + final class RuntimeTest extends AsyncTestCase { #[Test] @@ -95,4 +102,93 @@ public function weKilledTheThread(): void await($promise); } + + #[Test] + public function signalChannelIsConfigured(): void + { + $runtime = Runtime::create(new EventLoopBridge()); + + try { + $result = $runtime->run(static fn (): bool => extension_loaded('pcntl')); + + self::assertTrue($result); + } finally { + $runtime->close(); + } + } + + #[Test] + public function signalsAreForwardedToMainProcess(): void + { + $signalReceived = false; + + pcntl_signal(SIGTERM, static function () use (&$signalReceived): void { + $signalReceived = true; + }); + pcntl_async_signals(true); + + $runtime = Runtime::create(new EventLoopBridge()); + + try { + $mainPid = posix_getpid(); + $mainPid = $mainPid; + $runtime->run(static function (int $mainPid): void { + posix_kill($mainPid, SIGTERM); + }, [$mainPid]); + } finally { + $runtime->close(); + } + + self::assertTrue($signalReceived); + } + + #[Test] + public function signalsAreForwardedViaLoopAddSignal(): void + { + $signalReceived = false; + + $runtime = Runtime::create(new EventLoopBridge()); + + try { + Loop::addSignal(SIGTERM, static function () use (&$signalReceived): void { + $signalReceived = true; + }); + + $mainPid = posix_getpid(); + $mainPid = $mainPid; + $runtime->run(static function (int $mainPid): void { + posix_kill($mainPid, SIGTERM); + }, [$mainPid]); + } finally { + $runtime->close(); + } + + self::assertTrue($signalReceived); + } + + #[Test] + public function workerSignalListenerForwardsSignal(): void + { + $signalReceived = false; + + pcntl_signal(SIGTERM, static function () use (&$signalReceived): void { + $signalReceived = true; + }); + pcntl_async_signals(true); + + $runtime = Runtime::create(new EventLoopBridge()); + + try { + // The worker thread's SignalListener catches SIGTERM and forwards it via channel + // The Runtime's forwardSignals() then sends it to the main process via posix_kill + $runtime->run(static function (): void { + // Signal ourselves (the worker thread) - SignalListener should catch it + posix_kill(posix_getpid(), SIGTERM); + }); + } finally { + $runtime->close(); + } + + self::assertTrue($signalReceived); + } } diff --git a/tests/SignalListenerTest.php b/tests/SignalListenerTest.php new file mode 100644 index 0000000..df16e1f --- /dev/null +++ b/tests/SignalListenerTest.php @@ -0,0 +1,76 @@ + $channel */ + $channel = Channel::make('test-signals', Channel::Infinite); + + SignalListener::listen($channel); + + pcntl_async_signals(false); + + /** @phpstan-ignore staticMethod.alreadyNarrowedType */ + self::assertNotFalse(pcntl_signal_get_handler(SIGINT)); + /** @phpstan-ignore staticMethod.alreadyNarrowedType */ + self::assertNotFalse(pcntl_signal_get_handler(SIGTERM)); + /** @phpstan-ignore staticMethod.alreadyNarrowedType */ + self::assertNotFalse(pcntl_signal_get_handler(SIGHUP)); + /** @phpstan-ignore staticMethod.alreadyNarrowedType */ + self::assertNotFalse(pcntl_signal_get_handler(SIGQUIT)); + } + + /** @return array */ + public static function signals(): iterable + { + return [ + 'SIGINT' => [SIGINT, 'SIGINT'], + 'SIGTERM' => [SIGTERM, 'SIGTERM'], + 'SIGHUP' => [SIGHUP, 'SIGHUP'], + 'SIGQUIT' => [SIGQUIT, 'SIGQUIT'], + ]; + } + + #[Test] + #[DataProvider('signals')] + public function listenSendsSignalToChannel(int $signalNumber, string $signalName): void + { + /** @var Channel $channel */ + $channel = Channel::make('test-signals-' . $signalName, Channel::Infinite); + + SignalListener::listen($channel); + + pcntl_async_signals(true); + + posix_kill(posix_getpid(), $signalNumber); + + $signal = $channel->recv(); + /** @phpstan-ignore staticMethod.alreadyNarrowedType */ + self::assertInstanceOf(Signal::class, $signal); + self::assertSame($signalNumber, $signal->number); + self::assertSame($signalName, $signal->name); + } +} diff --git a/tests/SignalTest.php b/tests/SignalTest.php new file mode 100644 index 0000000..3362b87 --- /dev/null +++ b/tests/SignalTest.php @@ -0,0 +1,40 @@ + */ + public static function signalFactoryMethods(): iterable + { + return [ + 'sigint' => [Signal::class . '::sigint', SIGINT, 'SIGINT'], + 'sigterm' => [Signal::class . '::sigterm', SIGTERM, 'SIGTERM'], + 'sighup' => [Signal::class . '::sighup', SIGHUP, 'SIGHUP'], + 'sigquit' => [Signal::class . '::sigquit', SIGQUIT, 'SIGQUIT'], + ]; + } + + #[Test] + #[DataProvider('signalFactoryMethods')] + public function createsSignalWithCorrectNumberAndName(callable $factory, int $expectedNumber, string $expectedName): void + { + /** @var Signal $signal */ + $signal = $factory(); + + self::assertSame($expectedNumber, $signal->number); + self::assertSame($expectedName, $signal->name); + } +}