Horizon: forked processes write stdout a whole line at a time (builds on #1)

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.

src/Runtime.php

@@ -13,17 +13,17 @@ 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,
         WorkerLink::class, Server\ChildJob::class, Server\ChildRoutine::class, Server\Opcache::class, Server\RoleGuard::class,
         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. */

src/Support/LineBufferFilter.php

@@ -0,0 +1,101 @@
+<?php
+
+namespace Tevkert\QueuePrefork\Support;
+
+use php_user_filter;
+
+/**
+ * A write filter that holds output until a newline, then passes each complete line on as its own
+ * write.
+ *
+ * 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 is closed or freed, at exit included, is passed on then.
+ */
+final class LineBufferFilter extends php_user_filter
+{
+    public const NAME = 'queue-prefork.line-buffer';
+
+    /** The most a partial line is held before it's passed on as is: PIPE_BUF on Linux. */
+    public const MAX_HELD = 4096;
+
+    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;
+        }
+
+        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;
+    }
+}

src/Support/LineBufferedOutput.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace Tevkert\QueuePrefork\Support;
+
+/**
+ * Makes a forked process write its stdout a whole line at a time.
+ *
+ * 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 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<int, true> 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;
+        }
+    }
+}

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);