Add realtime output and related switch

Blacklight/libraries/Runners/BackfillRunner.php

@@ -19,15 +19,29 @@ class BackfillRunner extends BaseRunner
         $work = DB::select($select);
 
         $maxProcesses = (int) Settings::settingValue('backfillthreads');
-        $pool = $this->createPool($maxProcesses);
 
         $count = count($work);
-        if ($count > 0) {
-            $this->headerStart('backfill', $count, $maxProcesses);
-        } else {
+        if ($count === 0) {
             $this->headerNone();
+
+            return;
         }
 
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($work as $group) {
+                $commands[] = PHP_BINARY.' misc/update/backfill.php '.$group->name.(isset($group->max) ? (' '.$group->max) : '');
+            }
+            $this->runStreamingCommands($commands, $maxProcesses, 'backfill');
+
+            return;
+        }
+
+        $pool = $this->createPool($maxProcesses);
+
+        $this->headerStart('backfill', $count, $maxProcesses);
+
         $taskNum = $count;
         foreach ($work as $group) {
             $pool->add(function () use ($group) {
@@ -118,6 +132,17 @@ class BackfillRunner extends BaseRunner
             $queues[$i] = sprintf('get_range  backfill  %s  %s  %s  %s', $groupName, $data[0]->our_first - $i * $maxMessages - $maxMessages, $data[0]->our_first - $i * $maxMessages - 1, $i + 1);
         }
 
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($queues as $queue) {
+                $commands[] = $this->buildDnrCommand($queue);
+            }
+            $this->runStreamingCommands($commands, $threads, 'safe_backfill');
+
+            return;
+        }
+
         $pool = $this->createPool($threads);
 
         $this->headerStart('safe_backfill', count($queues), $threads);

Blacklight/libraries/Runners/BaseRunner.php

@@ -62,4 +62,67 @@ abstract class BaseRunner
             $this->colorCli->header('No work to do!');
         }
     }
+
+    /**
+     * Run multiple shell commands concurrently and stream their output in real-time.
+     * Uses Symfony Process start() with a small event loop to enforce max concurrency.
+     */
+    protected function runStreamingCommands(array $commands, int $maxProcesses, string $desc): void
+    {
+        $maxProcesses = max(1, (int) $maxProcesses);
+        $running = [];
+        $queue = $commands;
+        $total = \count($commands);
+        $started = 0;
+        $finished = 0;
+
+        $this->headerStart('postprocess: '.$desc, $total, $maxProcesses);
+
+        $startNext = function () use (&$queue, &$running, &$started) {
+            if (empty($queue)) {
+                return;
+            }
+            $cmd = array_shift($queue);
+            $proc = Process::fromShellCommandline($cmd);
+            $proc->setTimeout((int) config('nntmux.multiprocessing_max_child_time', 1800));
+            $proc->start(function ($type, $buffer) {
+                // Stream both STDOUT and STDERR
+                echo $buffer;
+            });
+            $running[spl_object_id($proc)] = $proc;
+            $started++;
+        };
+
+        // Prime initial processes
+        for ($i = 0; $i < $maxProcesses && ! empty($queue); $i++) {
+            $startNext();
+        }
+
+        // Event loop
+        while (! empty($running)) {
+            foreach ($running as $key => $proc) {
+                if (! $proc->isRunning()) {
+                    // Print any remaining buffered output
+                    $out = $proc->getIncrementalOutput();
+                    $err = $proc->getIncrementalErrorOutput();
+                    if ($out !== '') {
+                        echo $out;
+                    }
+                    if ($err !== '') {
+                        echo $err;
+                    }
+                    unset($running[$key]);
+                    $finished++;
+                    if (config('nntmux.echocli')) {
+                        $this->colorCli->primary('Finished task #'.($total - $finished + 1).' for '.$desc);
+                    }
+                    // Start next from queue if available
+                    if (! empty($queue)) {
+                        $startNext();
+                    }
+                }
+            }
+            usleep(100000); // 100ms
+        }
+    }
 }

Blacklight/libraries/Runners/BinariesRunner.php

@@ -18,15 +18,29 @@ class BinariesRunner extends BaseRunner
         );
 
         $maxProcesses = (int) Settings::settingValue('binarythreads');
-        $pool = $this->createPool($maxProcesses);
 
         $count = count($work);
-        if ($count > 0) {
-            $this->headerStart('binaries', $count, $maxProcesses);
-        } else {
+        if ($count === 0) {
             $this->headerNone();
+
+            return;
         }
 
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($work as $group) {
+                $commands[] = PHP_BINARY.' misc/update/update_binaries.php '.$group->name.' '.$group->max;
+            }
+            $this->runStreamingCommands($commands, $maxProcesses, 'binaries');
+
+            return;
+        }
+
+        $pool = $this->createPool($maxProcesses);
+
+        $this->headerStart('binaries', $count, $maxProcesses);
+
         $taskNum = $count;
         foreach ($work as $group) {
             $pool->add(function () use ($group) {
@@ -99,6 +113,17 @@ class BinariesRunner extends BaseRunner
             }
         }
 
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($queues as $queue) {
+                $commands[] = $this->buildDnrCommand($queue);
+            }
+            $this->runStreamingCommands($commands, $maxProcesses, 'safe_binaries');
+
+            return;
+        }
+
         $pool = $this->createPool($maxProcesses);
         $this->headerStart('safe_binaries', count($queues), $maxProcesses);
 

Blacklight/libraries/Runners/PostProcessRunner.php

@@ -17,6 +17,19 @@ class PostProcessRunner extends BaseRunner
             return;
         }
 
+        // If streaming is enabled, run commands with real-time output
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($releases as $release) {
+                // id may already be a single GUID bucket char; if not, take first char defensively
+                $char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
+                $commands[] = PHP_BINARY.' misc/update/postprocess.php '.$type.$char;
+            }
+            $this->runStreamingCommands($commands, $maxProcesses, $desc);
+
+            return;
+        }
+
         $pool = $this->createPool($maxProcesses);
         $count = count($releases);
         $this->headerStart('postprocess: '.$desc, $count, $maxProcesses);

Blacklight/libraries/Runners/ReleasesRunner.php

@@ -29,14 +29,27 @@ class ReleasesRunner extends BaseRunner
             }
         }
 
+        $count = count($uGroups);
+        if ($count === 0) {
+            $this->headerNone();
+
+            return;
+        }
+
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($uGroups as $group) {
+                $commands[] = $this->buildDnrCommand('releases  '.$group['id']);
+            }
+            $this->runStreamingCommands($commands, $maxProcesses, 'releases');
+
+            return;
+        }
+
         $pool = $this->createPool($maxProcesses);
 
-        $count = count($uGroups);
-        if ($count > 0) {
-            $this->headerStart('releases', $count, $maxProcesses);
-        } else {
-            $this->headerNone();
-        }
+        $this->headerStart('releases', $count, $maxProcesses);
 
         $taskNum = $count;
         foreach ($uGroups as $group) {
@@ -61,8 +74,26 @@ class ReleasesRunner extends BaseRunner
         $groups = DB::select('SELECT id , name FROM usenet_groups WHERE (active = 1 OR backfill = 1)');
         $maxProcesses = (int) Settings::settingValue('releasethreads');
 
+        $count = count($groups);
+        if ($count === 0) {
+            $this->headerNone();
+
+            return;
+        }
+
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($groups as $group) {
+                $commands[] = $this->buildDnrCommand('update_per_group  '.$group->id);
+            }
+            $this->runStreamingCommands($commands, $maxProcesses, 'update_per_group');
+
+            return;
+        }
+
         $pool = $this->createPool($maxProcesses);
-        $this->headerStart('update_per_group', count($groups), $maxProcesses);
+        $this->headerStart('update_per_group', $count, $maxProcesses);
 
         foreach ($groups as $group) {
             $pool->add(function () use ($group) {
@@ -107,14 +138,27 @@ class ReleasesRunner extends BaseRunner
             }
         }
 
+        $count = count($queues);
+        if ($count === 0) {
+            $this->headerNone();
+
+            return;
+        }
+
+        // Streaming mode
+        if ((bool) config('nntmux.stream_fork_output', false) === true) {
+            $commands = [];
+            foreach ($queues as $queue) {
+                $commands[] = PHP_BINARY.' misc/update/tmux/bin/groupfixrelnames.php "'.$queue.'" true';
+            }
+            $this->runStreamingCommands($commands, $maxThreads, 'fixRelNames_'.$mode);
+
+            return;
+        }
+
         $pool = $this->createPool($maxThreads);
 
-        $count = count($queues);
-        if ($count > 0) {
-            $this->headerStart('fixRelNames_'.$mode, $count, $maxThreads);
-        } else {
-            $this->headerNone();
-        }
+        $this->headerStart('fixRelNames_'.$mode, $count, $maxThreads);
 
         $taskNum = $count;
         foreach ($queues as $queue) {

config/nntmux.php

@@ -15,6 +15,7 @@ return [
     'admin_password' => env('ADMIN_PASS', 'admin'),
     'admin_email' => env('ADMIN_EMAIL', 'admin@example.com'),
     'multiprocessing_max_child_time' => env('NN_MULTIPROCESSING_MAX_CHILD_TIME', 1800),
+    'stream_fork_output' => env('STREAM_FORK_OUTPUT', false),
     'purge_inactive_users' => env('PURGE_INACTIVE_USERS', false),
     'purge_inactive_users_days' => env('PURGE_INACTIVE_USERS_DAYS', 180),
     'elasticsearch_enabled' => env('ELASTICSEARCH_ENABLED', false),