diff --git a/Blacklight/libraries/Runners/BackfillRunner.php b/Blacklight/libraries/Runners/BackfillRunner.php index 0a9a02058..96142772f 100644 --- a/Blacklight/libraries/Runners/BackfillRunner.php +++ b/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); diff --git a/Blacklight/libraries/Runners/BaseRunner.php b/Blacklight/libraries/Runners/BaseRunner.php index 285e927d1..4f12a8ec4 100644 --- a/Blacklight/libraries/Runners/BaseRunner.php +++ b/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 + } + } } diff --git a/Blacklight/libraries/Runners/BinariesRunner.php b/Blacklight/libraries/Runners/BinariesRunner.php index ff758b2f0..333645e9e 100644 --- a/Blacklight/libraries/Runners/BinariesRunner.php +++ b/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); diff --git a/Blacklight/libraries/Runners/PostProcessRunner.php b/Blacklight/libraries/Runners/PostProcessRunner.php index 0008b281f..02b1ae22e 100644 --- a/Blacklight/libraries/Runners/PostProcessRunner.php +++ b/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); diff --git a/Blacklight/libraries/Runners/ReleasesRunner.php b/Blacklight/libraries/Runners/ReleasesRunner.php index deb65d095..09c54262d 100644 --- a/Blacklight/libraries/Runners/ReleasesRunner.php +++ b/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) { diff --git a/config/nntmux.php b/config/nntmux.php index c1430e709..4dc9b1f26 100644 --- a/config/nntmux.php +++ b/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),