From 16b7815eac0fd9147c0bb5235e1bc9f4b888a16c Mon Sep 17 00:00:00 2001 From: Dan Krieger Date: Sun, 8 Nov 2026 22:32:23 -0800 Subject: [PATCH 1/2] Horizon: forked processes write their output a whole line at a time Stock Horizon gives each worker a pipe that its supervisor reads and relays, so lines from different workers never mix. Forked processes write straight to the stdout they inherited, which they all share, and queue:work writes each job line in three or four pieces, so pieces from different workers interleave mid-line in the container log. A write filter on STDOUT and STDERR now holds output until a newline and passes each complete line on as its own write. A pipe delivers a write of up to PIPE_BUF in one piece, so each line reaches the log whole. A partial line is held up to 4096 bytes and passed on when the stream closes. It filters the streams themselves, which are what Symfony's ConsoleOutput writes to outside Windows, and is attached in Wiring::child() so it is in place before the command writes anything. --- src/Runtime.php | 6 +- src/Support/LineBufferFilter.php | 57 +++++++++++++++++ src/Support/LineBufferedOutput.php | 47 ++++++++++++++ src/Wiring.php | 4 ++ tests/Unit/LineBufferedOutputTest.php | 90 +++++++++++++++++++++++++++ 5 files changed, 201 insertions(+), 3 deletions(-) create mode 100644 src/Support/LineBufferFilter.php create mode 100644 src/Support/LineBufferedOutput.php create mode 100644 tests/Unit/LineBufferedOutputTest.php diff --git a/src/Runtime.php b/src/Runtime.php index bd5e6d4..72650fc 100644 --- a/src/Runtime.php +++ b/src/Runtime.php @@ -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\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..00c67ff --- /dev/null +++ b/src/Support/LineBufferFilter.php @@ -0,0 +1,57 @@ +held .= $bucket->data; + $consumed += $bucket->datalen; + } + + $passed = false; + + while (($end = strpos($this->held, "\n")) !== false) { + $this->pass($out, substr($this->held, 0, $end + 1)); + $this->held = (string) substr($this->held, $end + 1); + $passed = true; + } + + if ($this->held !== '' && ($closing || strlen($this->held) >= self::MAX_HELD)) { + $this->pass($out, $this->held); + $this->held = ''; + $passed = true; + } + + 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): void + { + stream_bucket_append($out, stream_bucket_new($this->stream, $data)); + } +} diff --git a/src/Support/LineBufferedOutput.php b/src/Support/LineBufferedOutput.php new file mode 100644 index 0000000..6a79aff --- /dev/null +++ b/src/Support/LineBufferedOutput.php @@ -0,0 +1,47 @@ + stream ids already filtered in this process */ + private static array $filtered = []; + + public static function attachToStdio(): void + { + if (\defined('STDOUT')) { + self::attachTo(\STDOUT); + } + + if (\defined('STDERR')) { + self::attachTo(\STDERR); + } + } + + /** @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..fe5e714 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,9 @@ private static function child(Application $app): void RoleGuard::shim(72); } + // Forked processes share the stdout they inherited; keep each one's lines whole. + LineBufferedOutput::attachToStdio(); + 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..7599b51 --- /dev/null +++ b/tests/Unit/LineBufferedOutputTest.php @@ -0,0 +1,90 @@ +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)); + } + + /** @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; + } +} From d251cd765bde473e9b934dd71722d2870f5959ee Mon Sep 17 00:00:00 2001 From: Ando Hakobyan <337683099+tevkert@users.noreply.github.com> Date: Mon, 9 Nov 2026 14:22:49 +0400 Subject: [PATCH 2/2] Line buffering: keep the exit status and fatal-error output, filter only shared stdout - Only STDOUT is filtered. After an engine fatal error (a memory or time limit) PHP stops calling user filters, and a filtered stream drops everything written to it, so with STDERR filtered Laravel's report of the error was lost. Job lines go to STDOUT; STDERR is written as stock. - A partial line still held when a stream was freed without fclose() (STDOUT at exit) made the process exit with status 255 and lost the line: PHP invalidates the stream before the filter's last flush, so stream_bucket_new() threw. The filter now makes that last bucket against a temporary memory stream, which still reaches the stream being freed, and drops rather than throws if even that fails. Output written later in shutdown (shutdown functions, destructors) stays line-buffered. - A process forked after the filter was attached (a job's own pcntl_fork()) no longer repeats the part of a line its parent held: the filter drops what it inherited when its pid changes. - The filter is attached only where stdout is shared: forked Horizon processes and shared-log workers. A normal-mode queue:work worker has its stub's own stdio and keeps stock output behaviour. - One write with many lines is split with an offset instead of copying the rest of the buffer once per line. - LineBufferFilter extends php_user_filter, PHP's own class, so PreloadTest took it for a class with a framework parent and left it out of Runtime::CLASSES; a forked child then autoloaded it from whichever release was on disk. It is preloaded now, and the test counts php_user_filter as no parent. - Comments: the console kernel reroutes Symfony's command events itself once the app has booted (outside unit tests), so CommandStarting needs no opt-in; stock Horizon relays each worker's pipe but doesn't promise whole lines either. - README: the line-at-a-time stdout under "Limitations and differences from stock". - Tests, in real PHP processes: the exit status and held tail, STDERR output after a memory-limit fatal error, output from shutdown functions and destructors, a stream closed before exit, a child forked while its parent holds part of a line, and four forked writers sharing one pipe, which mangles lines without the filter; plus many lines in one write. The process tests end themselves after 30 s rather than hang CI. Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + src/Runtime.php | 8 +- src/Support/LineBufferFilter.php | 70 +++++++++-- src/Support/LineBufferedOutput.php | 21 ++-- src/Wiring.php | 7 +- tests/Unit/LineBufferedOutputTest.php | 175 ++++++++++++++++++++++++++ tests/Unit/PreloadTest.php | 2 +- 7 files changed, 252 insertions(+), 32 deletions(-) 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 72650fc..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\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, + 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 index 00c67ff..40da9bb 100644 --- a/src/Support/LineBufferFilter.php +++ b/src/Support/LineBufferFilter.php @@ -8,13 +8,14 @@ * A write filter that holds output until a newline, then passes each complete line on as its own * write. * - * Forked workers share their parent's stdout instead of each having a pipe to a supervisor, and - * `queue:work` writes every job line in three or four pieces. Without this, pieces from different - * workers interleave mid-line. A pipe delivers a write of up to PIPE_BUF (4096 bytes on Linux) in one - * piece, so a line written whole reaches the log whole. + * Forked Horizon processes share the master's stdout, and shared-log workers one log file, and + * `queue:work` writes every job line in three or four pieces, so pieces from different processes + * interleave mid-line. A pipe delivers a write of up to PIPE_BUF (4096 bytes on Linux) in one piece, + * and an append-mode file takes each write at its end, so a line written whole, up to that size, + * reaches the log whole. * * A partial line longer than MAX_HELD is passed on rather than held without bound, and whatever is - * held when the stream closes is passed on then. + * held when the stream is closed or freed, at exit included, is passed on then. */ final class LineBufferFilter extends php_user_filter { @@ -25,33 +26,76 @@ final class LineBufferFilter extends php_user_filter private string $held = ''; + private int $pid = 0; + + public function onCreate(): bool + { + $this->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; + } - while (($end = strpos($this->held, "\n")) !== false) { - $this->pass($out, substr($this->held, 0, $end + 1)); - $this->held = (string) substr($this->held, $end + 1); - $passed = true; + if ($start > 0) { + $this->held = substr($this->held, $start); } if ($this->held !== '' && ($closing || strlen($this->held) >= self::MAX_HELD)) { - $this->pass($out, $this->held); + $passed = $this->pass($out, $this->held) || $passed; $this->held = ''; - $passed = true; } 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): void + private function pass($out, string $data): bool { - stream_bucket_append($out, stream_bucket_new($this->stream, $data)); + 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 index 6a79aff..834933e 100644 --- a/src/Support/LineBufferedOutput.php +++ b/src/Support/LineBufferedOutput.php @@ -3,30 +3,27 @@ namespace Tevkert\QueuePrefork\Support; /** - * Makes a forked process write its output a whole line at a time. + * Makes a forked process write its stdout a whole line at a time. * - * Stock Horizon gives each worker a pipe and its supervisor relays what arrives, so lines from - * different workers never mix. A forked worker writes straight to the stdout it inherited, which every - * other forked worker shares; {@see LineBufferFilter} keeps their lines apart. + * Stock Horizon gives each worker a pipe of its own, which its supervisor reads and relays. A forked + * Horizon process writes straight to the stdout it inherited, which every forked process in the tree + * shares, as shared-log workers share one log file; {@see LineBufferFilter} keeps their lines apart. * - * It filters the STDOUT and STDERR streams themselves, because that is what Symfony's ConsoleOutput - * writes to outside Windows, and it needs no console event: Laravel only dispatches CommandStarting - * once something has opted in to rerouting Symfony's command events. + * It filters the STDOUT stream itself, because that is what Symfony's ConsoleOutput writes to outside + * Windows, so it covers whatever command runs without depending on a console event. STDERR stays as + * is: after an engine fatal error (a memory or time limit) PHP stops calling user filters, and a + * filtered stream then drops everything written to it, Laravel's report of that error included. */ final class LineBufferedOutput { /** @var array stream ids already filtered in this process */ private static array $filtered = []; - public static function attachToStdio(): void + public static function attachToStdout(): void { if (\defined('STDOUT')) { self::attachTo(\STDOUT); } - - if (\defined('STDERR')) { - self::attachTo(\STDERR); - } } /** @param resource $stream */ diff --git a/src/Wiring.php b/src/Wiring.php index fe5e714..a07fba9 100644 --- a/src/Wiring.php +++ b/src/Wiring.php @@ -79,8 +79,11 @@ private static function child(Application $app): void RoleGuard::shim(72); } - // Forked processes share the stdout they inherited; keep each one's lines whole. - LineBufferedOutput::attachToStdio(); + 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 index 7599b51..49e8ba2 100644 --- a/tests/Unit/LineBufferedOutputTest.php +++ b/tests/Unit/LineBufferedOutputTest.php @@ -65,6 +65,181 @@ public function test_a_console_output_on_a_filtered_stream_writes_whole_lines_an $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 { 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]; } }