diff --git a/README.md b/README.md index 383918b..837ba60 100644 --- a/README.md +++ b/README.md @@ -265,6 +265,7 @@ No job failed, and every run's last hour was within a point of its median. Prefo - **`queue:work` in a terminal:** Ctrl-Z suspends only the stub, so the worker keeps taking jobs until `fg` or Ctrl-C. Under a process manager this never applies. - **Tested Horizon minors only.** The package overrides a few Horizon internals (`AddSupervisor`, `SupervisorFactory`, `Supervisor` and `ProcessPool`), so a newer minor runs stock until a release raises the limit; `allow_untested_horizon` lifts the check at your own risk. - **The forked Horizon master is Linux-only.** On macOS only supervisors and workers are forked. +- **Forked Horizon processes and shared-log workers write stdout a line at a time.** They share one stdout or log file, so each process holds a line until its newline (up to 4 KB) and writes it in one piece, which keeps lines from different workers apart. A partial line shows up once it's finished, and output that doesn't go through PHP's `STDOUT` stream (`echo`, `dump()`, subprocesses) isn't held, so it can come before it. Stderr is written as in stock. - **Shared-memory extensions are shared by the whole tree,** as under PHP-FPM: APCu with `apc.enable_cli`, Relay and sysvshm. - **Entry scripts must read arguments from `$_SERVER['argv']`,** not `getopt()`. Laravel's `artisan` does. - **Pid reuse (Horizon).** Signals go through each process's parent, which owns the pid until it reaps it. Only if a fork server has died are its children signalled directly, after a pid and start-time check, which leaves a small window that pidfds close when FFI is on. Stock Horizon signals pids from Redis with no check at all. diff --git a/src/Runtime.php b/src/Runtime.php index bd5e6d4..d578e84 100644 --- a/src/Runtime.php +++ b/src/Runtime.php @@ -13,7 +13,7 @@ final class Runtime /** Bumped whenever the template/fork-server/child protocol changes (checked by children). */ public const PROTOCOL = 2; - /** The package's classes that don't extend the framework's: what preload() loads. */ + /** The package's classes that don't extend the framework's (PHP's own count as none): what preload() loads. */ public const CLASSES = [ Client\Broker::class, Client\Direct::class, Client\Report::class, Handoff::class, Link::class, Master::class, MasterHandoff::class, Preflight::class, QueueWorkerLink::class, Runtime::class, Settings::class, Wiring::class, @@ -21,9 +21,9 @@ final class Runtime Server\SharedLog::class, Server\Signals::class, Server\Template::class, Server\UnitServer::class, Server\WorkServer::class, Server\WorkerContext::class, Support\Audit::class, Support\Channel::class, Support\CommandLine::class, Support\DotEnv::class, Support\EarlyConfig::class, Support\Fd::class, - Support\Identity::class, Support\LogTargets::class, Support\Native::class, Support\Process::class, - Support\RuntimeDir::class, Support\StatusFiles::class, Support\Switches::class, Support\TemplateKey::class, - Support\WorkTemplates::class, + Support\Identity::class, Support\LineBufferFilter::class, Support\LineBufferedOutput::class, + Support\LogTargets::class, Support\Native::class, Support\Process::class, Support\RuntimeDir::class, + Support\StatusFiles::class, Support\Switches::class, Support\TemplateKey::class, Support\WorkTemplates::class, ]; /** template | fork-server | unit-loop | worker | horizon-master, or null in an ordinary process. */ diff --git a/src/Support/LineBufferFilter.php b/src/Support/LineBufferFilter.php new file mode 100644 index 0000000..40da9bb --- /dev/null +++ b/src/Support/LineBufferFilter.php @@ -0,0 +1,101 @@ +pid = getmypid(); + + return true; + } + + public function filter($in, $out, &$consumed, bool $closing): int + { + if ($this->pid !== getmypid()) { + // A process forked after this filter was attached (a job's own pcntl_fork()) inherits what + // the parent held, and the parent passes that on itself. + $this->pid = getmypid(); + $this->held = ''; + } + + while ($bucket = stream_bucket_make_writeable($in)) { + $this->held .= $bucket->data; + $consumed += $bucket->datalen; + } + + $passed = false; + $start = 0; + + while (($end = strpos($this->held, "\n", $start)) !== false) { + $passed = $this->pass($out, substr($this->held, $start, $end + 1 - $start)) || $passed; + $start = $end + 1; + } + + if ($start > 0) { + $this->held = substr($this->held, $start); + } + + if ($this->held !== '' && ($closing || strlen($this->held) >= self::MAX_HELD)) { + $passed = $this->pass($out, $this->held) || $passed; + $this->held = ''; + } + + return $passed ? PSFS_PASS_ON : PSFS_FEED_ME; + } + + /** One bucket per line: the stream writes each bucket with its own write call. */ + private function pass($out, string $data): bool + { + if (is_resource($this->stream)) { + stream_bucket_append($out, stream_bucket_new($this->stream, $data)); + + return true; + } + + // A stream freed without fclose() (STDOUT at exit, or any stream whose last reference + // goes) is invalidated before this last flush. A bucket only needs some stream to be made + // against, and still goes to the stream being freed. If even that fails, drop what's left + // rather than throw, which at exit would turn the status into 255. + $stream = @fopen('php://memory', 'r'); + + if ($stream === false) { + return false; + } + + try { + stream_bucket_append($out, stream_bucket_new($stream, $data)); + } catch (\Throwable) { + return false; + } finally { + fclose($stream); + } + + return true; + } +} diff --git a/src/Support/LineBufferedOutput.php b/src/Support/LineBufferedOutput.php new file mode 100644 index 0000000..834933e --- /dev/null +++ b/src/Support/LineBufferedOutput.php @@ -0,0 +1,44 @@ + stream ids already filtered in this process */ + private static array $filtered = []; + + public static function attachToStdout(): void + { + if (\defined('STDOUT')) { + self::attachTo(\STDOUT); + } + } + + /** @param resource $stream */ + public static function attachTo($stream): void + { + if (! is_resource($stream) || isset(self::$filtered[(int) $stream])) { + return; + } + + if (! in_array(LineBufferFilter::NAME, stream_get_filters(), true)) { + stream_filter_register(LineBufferFilter::NAME, LineBufferFilter::class); + } + + if (@stream_filter_append($stream, LineBufferFilter::NAME, STREAM_FILTER_WRITE) !== false) { + self::$filtered[(int) $stream] = true; + } + } +} diff --git a/src/Wiring.php b/src/Wiring.php index 5c01774..a07fba9 100644 --- a/src/Wiring.php +++ b/src/Wiring.php @@ -14,6 +14,7 @@ use Tevkert\QueuePrefork\Horizon\PreforkHorizonCommand; use Tevkert\QueuePrefork\Horizon\PreforkSupervisorFactory; use Tevkert\QueuePrefork\Server\RoleGuard; +use Tevkert\QueuePrefork\Support\LineBufferedOutput; use Tevkert\QueuePrefork\Support\StatusFiles; use Tevkert\QueuePrefork\Support\Switches; use Throwable; @@ -78,6 +79,12 @@ private static function child(Application $app): void RoleGuard::shim(72); } + if (Runtime::$output !== 'normal') { + // Forked Horizon processes share the master's stdout, and shared-log workers one log file: + // keep each process's lines whole. A normal-mode queue:work worker has its stub's own stdio. + LineBufferedOutput::attachToStdout(); + } + if (Runtime::$role === 'horizon-master') { self::hostedMaster($app); diff --git a/tests/Unit/LineBufferedOutputTest.php b/tests/Unit/LineBufferedOutputTest.php new file mode 100644 index 0000000..49e8ba2 --- /dev/null +++ b/tests/Unit/LineBufferedOutputTest.php @@ -0,0 +1,265 @@ +filteredPair(); + + fwrite($writer, ' 2026-11-09 06:05:32 Laravel\Scout\Jobs\MakeSearchable'); + fwrite($writer, ' .........'); + $this->assertSame('', $this->readAvailable($reader)); + + fwrite($writer, " RUNNING\n"); + $this->assertSame(" 2026-11-09 06:05:32 Laravel\\Scout\\Jobs\\MakeSearchable ......... RUNNING\n", $this->readAvailable($reader)); + } + + public function test_complete_lines_pass_and_the_rest_waits(): void + { + [$writer, $reader] = $this->filteredPair(); + + fwrite($writer, "one\ntwo\nthr"); + $this->assertSame("one\ntwo\n", $this->readAvailable($reader)); + + fwrite($writer, "ee\n"); + $this->assertSame("three\n", $this->readAvailable($reader)); + } + + public function test_a_partial_line_is_passed_on_when_the_stream_closes(): void + { + [$writer, $reader] = $this->filteredPair(); + + fwrite($writer, 'no newline'); + fclose($writer); + + $this->assertSame('no newline', $this->readAvailable($reader)); + } + + public function test_a_partial_line_is_not_held_past_the_limit(): void + { + [$writer, $reader] = $this->filteredPair(); + + fwrite($writer, str_repeat('x', LineBufferFilter::MAX_HELD)); + + $this->assertSame(LineBufferFilter::MAX_HELD, strlen($this->readAvailable($reader))); + } + + public function test_a_console_output_on_a_filtered_stream_writes_whole_lines_and_attaching_twice_is_harmless(): void + { + [$writer, $reader] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + LineBufferedOutput::attachTo($writer); + LineBufferedOutput::attachTo($writer); + $output = new StreamOutput($writer); + + $output->write('half'); + $this->assertSame('', $this->readAvailable($reader)); + + $output->writeln(' and half'); + $this->assertSame("half and half\n", $this->readAvailable($reader)); + } + + public function test_many_lines_in_one_write_each_pass_whole(): void + { + // A memory stream, not a socket: hundreds of small writes into a socket nobody reads yet can fill + // its send buffer and block. + $stream = fopen('php://memory', 'w+'); + LineBufferedOutput::attachTo($stream); + $lines = ''; + + for ($i = 0; $i < 2000; $i++) { + $lines .= "line {$i}\n"; + } + + fwrite($stream, $lines.'partial'); + rewind($stream); + + $this->assertSame($lines, stream_get_contents($stream)); + } + + public function test_a_held_tail_is_passed_on_at_exit_and_the_exit_status_is_kept(): void + { + $result = $this->runPhp(' + LineBufferedOutput::attachToStdout(); + fwrite(STDOUT, "whole\ntail"); + exit(3); + '); + + $this->assertSame([3, "whole\ntail", ''], $result); + } + + public function test_stderr_is_left_alone_so_output_after_an_engine_fatal_error_survives(): void + { + // After a fatal error PHP no longer calls user filters, and a filtered stream drops what is + // written to it; Laravel reports such an error on STDERR from a shutdown function. + [$status, $out, $err] = $this->runPhp(' + LineBufferedOutput::attachToStdout(); + register_shutdown_function(fn () => fwrite(STDERR, "reported on stderr\n")); + fwrite(STDERR, "stderr part"); + fwrite(STDOUT, "before\n"); + ini_set("memory_limit", "16M"); + $big = str_repeat("x", 64 * 1024 * 1024); + '); + + $this->assertSame([255, "before\n"], [$status, $out]); + $this->assertStringContainsString('Allowed memory size', $err); + $this->assertStringStartsWith('stderr part', $err); + $this->assertStringEndsWith("reported on stderr\n", $err); + } + + public function test_output_from_later_shutdown_code_stays_line_buffered_and_its_tail_is_passed_on(): void + { + $result = $this->runPhp(' + LineBufferedOutput::attachToStdout(); + $GLOBALS["object"] = new class { + public function __destruct() { fwrite(STDOUT, "destructor"); fwrite(STDOUT, " line\ndestructor tail"); } + }; + register_shutdown_function(fn () => fwrite(STDOUT, "shutdown")); + fwrite(STDOUT, "tail "); + '); + + $this->assertSame([0, "tail shutdowndestructor line\ndestructor tail", ''], $result); + } + + public function test_a_child_forked_after_attaching_does_not_repeat_what_its_parent_held(): void + { + if (! function_exists('pcntl_fork')) { + $this->markTestSkipped('needs ext-pcntl'); + } + + // A job that forks (as the Concurrency fork driver does) while its process holds part of a line. + [$status, $out, $err] = $this->runPhp(' + LineBufferedOutput::attachToStdout(); + fwrite(STDOUT, "parent start"); + $children = []; + + foreach (["silent", "talking"] as $kind) { + $pid = pcntl_fork(); + + if ($pid === 0) { + if ($kind === "talking") { + fwrite(STDOUT, "child line\nchild tail"); + } + + exit(0); + } + + $children[] = $pid; + } + + foreach ($children as $pid) { + pcntl_waitpid($pid, $status); + } + + fwrite(STDOUT, " parent end\n"); + '); + + // The children are done before the parent finishes its line; the silent one writes nothing. + $this->assertSame([0, "child line\nchild tailparent start parent end\n", ''], [$status, $out, $err]); + } + + public function test_a_stream_closed_before_exit_leaves_the_exit_status_alone(): void + { + $result = $this->runPhp(' + LineBufferedOutput::attachToStdout(); + fwrite(STDOUT, "whole\ntail"); + fclose(STDOUT); + '); + + $this->assertSame([0, "whole\ntail", ''], $result); + } + + public function test_forked_writers_sharing_a_pipe_keep_their_lines_whole(): void + { + if (! function_exists('pcntl_fork')) { + $this->markTestSkipped('needs ext-pcntl'); + } + + // As in a forked Horizon tree: each child filters the stdout it inherited and writes every line + // in pieces, with pauses that let the others write in between. + [$status, $out, $err] = $this->runPhp(' + $children = []; + + for ($w = 1; $w <= 4; $w++) { + $pid = pcntl_fork(); + + if ($pid === 0) { + pcntl_alarm(30); // a writer that hangs ends rather than holding the pipe open + LineBufferedOutput::attachToStdout(); + + for ($n = 1; $n <= 100; $n++) { + fwrite(STDOUT, " worker {$w}"); + usleep(random_int(0, 300)); + fwrite(STDOUT, " line {$n} ........."); + usleep(random_int(0, 300)); + fwrite(STDOUT, " DONE\n"); + } + + exit(0); + } + + $children[] = $pid; + } + + foreach ($children as $pid) { + pcntl_waitpid($pid, $status); + } + '); + + $this->assertSame([0, ''], [$status, $err]); + $lines = explode("\n", rtrim($out, "\n")); + $this->assertCount(400, $lines); + + foreach ($lines as $line) { + $this->assertMatchesRegularExpression('/^ worker [1-4] line \d+ \.{9} DONE$/', $line); + } + } + + /** @return array{0: int, 1: string, 2: string} exit status, stdout and stderr of a PHP process running $code */ + private function runPhp(string $code): array + { + // The alarm makes a hang fail the test instead of using up the CI job's time. + $script = 'require $argv[1]; use Tevkert\QueuePrefork\Support\LineBufferedOutput; ' + .'if (function_exists("pcntl_alarm")) { pcntl_alarm(30); } '.$code; + $process = proc_open( + [PHP_BINARY, '-d', 'display_errors=stderr', '-r', $script, dirname(__DIR__, 2).'/vendor/autoload.php'], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], + $pipes, + ); + $out = stream_get_contents($pipes[1]); + $err = stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + + return [proc_close($process), $out, $err]; + } + + /** @return array{0: resource, 1: resource} */ + private function filteredPair(): array + { + [$writer, $reader] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + LineBufferedOutput::attachTo($writer); + + return [$writer, $reader]; + } + + /** @param resource $reader */ + private function readAvailable($reader): string + { + stream_set_blocking($reader, false); + usleep(20_000); + $data = ''; + + while (($chunk = fread($reader, 65536)) !== false && $chunk !== '') { + $data .= $chunk; + } + + return $data; + } +} diff --git a/tests/Unit/PreloadTest.php b/tests/Unit/PreloadTest.php index 0891ae3..af756ea 100644 --- a/tests/Unit/PreloadTest.php +++ b/tests/Unit/PreloadTest.php @@ -17,7 +17,7 @@ public function test_preload_lists_every_class_without_a_framework_parent(): voi foreach ($files as $file) { $source = $file->getExtension() === 'php' ? (string) file_get_contents($file->getPathname()) : ''; - if (preg_match('/^namespace (\S+);/m', $source, $ns) && preg_match('/^(?:final |abstract )?class (\w+)(?! extends)\b/m', $source, $class)) { + if (preg_match('/^namespace (\S+);/m', $source, $ns) && preg_match('/^(?:final |abstract )?class (\w+)(?! extends (?!php_user_filter\b))\b/m', $source, $class)) { $expected[] = $ns[1].'\\'.$class[1]; } }