Skip to content

[poc] feat(jobs-worker): multi-threading #64199

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ArtificialOwl wants to merge 3 commits into
base: master
Choose a base branch
from
Draft

[poc] feat(jobs-worker): multi-threading #64199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d929eef
fix(jobs-worker): delay next check by one second
ArtificialOwl Sep 10, 2026
bfc45c8
fix(jobs-worker): validate freshly available job
ArtificialOwl Sep 10, 2026
57b6544
feat(jobs-worker): multi-threading
ArtificialOwl Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 106 additions & 17 deletions core/Command/Background/JobWorker.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,26 +9,34 @@

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,
private ITempManager $tempManager,
private ISetupManager $setupManager,
private readonly JobRuns $jobRuns,
private readonly JobClassesRegistry $jobClassesRegistry,
private readonly IDBConnection $connection,
) {
parent::__construct($jobList, $logger);
}
Expand Down Expand Up @@ -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',
Expand All @@ -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('<info>Background job worker will stop after ' . $stopAfterSeconds . ' seconds</info>');
if ($this->stopAfterSeconds !== null) {
$output->writeln('<info>Background job worker will stop after ' . $this->stopAfterSeconds . ' seconds</info>');
}

$jobClasses = $input->getArgument('job-classes');
Expand All @@ -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, "<bg={$color}> </> ");
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('<info>Background job worker stopped</info>');
break;
} catch (InterruptedException) {
$output->writeln($prefix . '<info>Background job worker stopped</info>');
return;
}

$this->printSummary($input, $output);
Expand All @@ -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;
}
Expand All @@ -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);
Expand All @@ -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();
Expand All @@ -161,8 +239,6 @@ protected function execute(InputInterface $input, OutputInterface $output): int
break;
}
}

return 0;
}

private function printSummary(InputInterface $input, OutputInterface $output): void {
Expand Down Expand Up @@ -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);
}
}
2 changes: 1 addition & 1 deletion lib/private/BackgroundJob/JobList.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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')));
Expand Down
2 changes: 1 addition & 1 deletion lib/public/BackgroundJob/TimedJob.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.');
}
Expand Down
Loading