From d929eef6b7612f3811eb27375ae67f2fb891b976 Mon Sep 17 00:00:00 2001 From: Maxence Lange Date: Thu, 10 Sep 2026 10:14:47 -0100 Subject: [PATCH 1/3] fix(jobs-worker): delay next check by one second Signed-off-by: Maxence Lange --- lib/private/BackgroundJob/JobList.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/private/BackgroundJob/JobList.php b/lib/private/BackgroundJob/JobList.php index 2c63ef1777164..9345346fb235c 100644 --- a/lib/private/BackgroundJob/JobList.php +++ b/lib/private/BackgroundJob/JobList.php @@ -246,7 +246,7 @@ public function getNext(bool $onlyTimeSensitive = false, ?array $jobClasses = nu $update = $this->connection->getQueryBuilder(); $update->update('jobs') ->set('reserved_at', $update->createNamedParameter($this->timeFactory->getTime())) - ->set('last_checked', $update->createNamedParameter($this->timeFactory->getTime())) + ->set('last_checked', $update->createNamedParameter($this->timeFactory->getTime() + 1)) ->where($update->expr()->eq('id', $update->createParameter('jobid'))) ->andWhere($update->expr()->eq('reserved_at', $update->createParameter('reserved_at'))) ->andWhere($update->expr()->eq('last_checked', $update->createParameter('last_checked'))); From bfc45c84fd968f34a9ad917e5346276cd9faf550 Mon Sep 17 00:00:00 2001 From: Maxence Lange Date: Thu, 10 Sep 2026 15:42:19 -0100 Subject: [PATCH 2/3] fix(jobs-worker): validate freshly available job Signed-off-by: Maxence Lange --- lib/public/BackgroundJob/TimedJob.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/public/BackgroundJob/TimedJob.php b/lib/public/BackgroundJob/TimedJob.php index 9c2de7fb9fd47..47a65b91428d3 100644 --- a/lib/public/BackgroundJob/TimedJob.php +++ b/lib/public/BackgroundJob/TimedJob.php @@ -80,7 +80,7 @@ public function setTimeSensitivity(int $sensitivity): void { */ #[\Override] final public function start(IJobList $jobList): void { - if (($this->time->getTime() - $this->lastRun) > $this->interval) { + if (($this->time->getTime() - $this->lastRun) >= $this->interval) { if ($this->interval >= 12 * 60 * 60 && $this->isTimeSensitive()) { Server::get(LoggerInterface::class)->debug('TimedJob ' . get_class($this) . ' has a configured interval of ' . $this->interval . ' seconds, but is also marked as time sensitive. Please consider marking it as time insensitive to allow more sensitive jobs to run when needed.'); } From 57b6544912a875693faf1c74402983077008f399 Mon Sep 17 00:00:00 2001 From: Maxence Lange Date: Thu, 10 Sep 2026 16:10:13 -0100 Subject: [PATCH 3/3] feat(jobs-worker): multi-threading Signed-off-by: Maxence Lange --- core/Command/Background/JobWorker.php | 123 ++++++++++++++++++++++---- 1 file changed, 106 insertions(+), 17 deletions(-) diff --git a/core/Command/Background/JobWorker.php b/core/Command/Background/JobWorker.php index ad30e3e9f7bf2..1b8de611e2aec 100644 --- a/core/Command/Background/JobWorker.php +++ b/core/Command/Background/JobWorker.php @@ -9,19 +9,26 @@ namespace OC\Core\Command\Background; +use DateTimeImmutable; use OC\BackgroundJob\JobClassesRegistry; use OC\BackgroundJob\JobRuns; use OC\Core\Command\InterruptedException; use OCP\BackgroundJob\IJobList; use OCP\Files\ISetupManager; +use OCP\IDBConnection; use OCP\ITempManager; use Psr\Log\LoggerInterface; +use Symfony\Component\Console\Exception\InvalidOptionException; use Symfony\Component\Console\Input\InputArgument; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\OutputInterface; class JobWorker extends JobBase { + private int $forkCount = 0; + private ?int $stopAfterSeconds; + private ?int $startTime; + public function __construct( protected IJobList $jobList, protected LoggerInterface $logger, @@ -29,6 +36,7 @@ public function __construct( private ISetupManager $setupManager, private readonly JobRuns $jobRuns, private readonly JobClassesRegistry $jobClassesRegistry, + private readonly IDBConnection $connection, ) { parent::__construct($jobList, $logger); } @@ -58,6 +66,13 @@ protected function configure(): void { 'Interval in seconds in which the worker should repeat already processed jobs (set to 0 for no repeat)', 1 ) + ->addOption( + 'thread', + 'j', + InputOption::VALUE_REQUIRED, + 'create multiple thread', + 1 + ) ->addOption( 'stop_after', 't', @@ -69,13 +84,13 @@ protected function configure(): void { #[\Override] protected function execute(InputInterface $input, OutputInterface $output): int { - $startTime = time(); + $this->startTime = time(); $stopAfterOptionValue = $input->getOption('stop_after'); - $stopAfterSeconds = $stopAfterOptionValue === null + $this->stopAfterSeconds = $stopAfterOptionValue === null ? null : $this->parseStopAfter($stopAfterOptionValue); - if ($stopAfterSeconds !== null) { - $output->writeln('Background job worker will stop after ' . $stopAfterSeconds . ' seconds'); + if ($this->stopAfterSeconds !== null) { + $output->writeln('Background job worker will stop after ' . $this->stopAfterSeconds . ' seconds'); } $jobClasses = $input->getArgument('job-classes'); @@ -91,18 +106,75 @@ protected function execute(InputInterface $input, OutputInterface $output): int } } + $multiThread = $input->getOption('thread') ?? 0; + if ($multiThread > 1) { + if (!extension_loaded('posix')) { + throw new InvalidOptionException('posix extension is required to use --thread'); + } + + while (true) { + usleep(10000); // not needed but still better to slightly desync + $pid = pcntl_fork(); + // work around as the parent database connection is inherited by the child. + // when child process is over, parent process database connection will drop. + // The drop can happen anytime, even in the middle of a running request. + // work around is to close the connection as soon as possible after forking. + $this->connection->close(); + + if ($pid === -1) { + // TODO: manage issue while forking + } elseif ($pid === 0) { + $color = $this->createRandomColor(); + $this->runWorker($input, $output, $jobClasses, " "); + exit(); + } else { + // main process, counting forks + $this->forkCount++; + while (true) { + // Handle canceling of the process + try { + $this->abortIfInterrupted(); + } catch (InterruptedException) { + return 0; + } + + if (pcntl_waitpid(0, $status, WNOHANG) !== 0) { + $this->forkCount--; + } + if ($this->forkCount < $multiThread) { + break; + } + usleep(50000); + } + } + } + } else { + $this->runWorker($input, $output, $jobClasses); + } + + $this->waitForChild(); + return 0; + } + + + private function runWorker( + InputInterface $input, + OutputInterface $output, + ?array $jobClasses, + string $prefix = ''): void { + while (true) { // Stop if we exceeded stop_after value - if ($stopAfterSeconds !== null && ($startTime + $stopAfterSeconds) < time()) { - $output->writeln('stop_after time has been exceeded, exiting...', OutputInterface::VERBOSITY_VERBOSE); + if ($this->stopAfterSeconds !== null && ($this->startTime + $this->stopAfterSeconds) < time()) { + $output->writeln($prefix . 'stop_after time has been exceeded, exiting...', OutputInterface::VERBOSITY_VERBOSE); break; } // Handle canceling of the process try { $this->abortIfInterrupted(); - } catch (InterruptedException $e) { - $output->writeln('Background job worker stopped'); - break; + } catch (InterruptedException) { + $output->writeln($prefix . 'Background job worker stopped'); + return; } $this->printSummary($input, $output); @@ -112,15 +184,15 @@ protected function execute(InputInterface $input, OutputInterface $output): int if (!$job) { if ($input->getOption('once') === true) { if ($jobClasses === null) { - $output->writeln('No job is currently queued', OutputInterface::VERBOSITY_VERBOSE); + $output->writeln($prefix . 'No job is currently queued', OutputInterface::VERBOSITY_VERBOSE); } else { - $output->writeln('No job of classes [' . implode(', ', $jobClasses) . '] is currently queued', OutputInterface::VERBOSITY_VERBOSE); + $output->writeln($prefix . 'No job of classes [' . implode(', ', $jobClasses) . '] is currently queued', OutputInterface::VERBOSITY_VERBOSE); } - $output->writeln('Exiting...', OutputInterface::VERBOSITY_VERBOSE); + $output->writeln($prefix . 'Exiting...', OutputInterface::VERBOSITY_VERBOSE); break; } - $output->writeln('Waiting for new jobs to be queued', OutputInterface::VERBOSITY_VERBOSE); + $output->writeln($prefix . 'Waiting for new jobs to be queued', OutputInterface::VERBOSITY_VERBOSE); if ((int)$input->getOption('interval') === 0) { break; } @@ -130,7 +202,13 @@ protected function execute(InputInterface $input, OutputInterface $output): int } $jobClassName = get_class($job); - $output->writeln('Running job ' . $jobClassName . ' with ID ' . $job->getId()); + $now = new DateTimeImmutable(); + + if ($input->getOption('output') === 'row') { + $output->writeln($prefix . ' ' . $now->format('Y-m-d H:i:s.v') . ' | ' . str_pad($job->getId(), 20) . ' | ' . str_pad((string)$job->getLastRun(), 14) . ' | ' . $jobClassName); + } else { + $output->writeln($prefix . 'Running job ' . $jobClassName . ' with ID ' . $job->getId() . ' ' . $job->getLastRun()); + } if ($output->isVerbose()) { $this->printJobInfo($job->getId(), $job, $output); @@ -148,7 +226,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int // It should be a temporary state until a proper job runner is implemented. $this->jobRuns->finished($jobRunId, (int)($timeSpent * 1000), (int)($jobMemoryPeak / 1024)); - $output->writeln('Job ' . $job->getId() . ' has finished', OutputInterface::VERBOSITY_VERBOSE); + $output->writeln($prefix . 'Job ' . $job->getId() . ' has finished', OutputInterface::VERBOSITY_VERBOSE); // clean up after unclean jobs $this->setupManager->tearDown(); @@ -161,8 +239,6 @@ protected function execute(InputInterface $input, OutputInterface $output): int break; } } - - return 0; } private function printSummary(InputInterface $input, OutputInterface $output): void { @@ -193,4 +269,17 @@ private function parseStopAfter(string $value): ?int { } return null; } + + public function waitForChild(): void { + if (!extension_loaded('posix')) { + return; + } + + while (pcntl_waitpid(0, $status) !== -1) { + } + } + + public function createRandomColor(): string { + return '#' . str_pad(dechex(mt_rand(0, 0xFFFFFF)), 6, '0', STR_PAD_LEFT); + } }