feat: implement spawning multiple workers for generate-all

lib/Command/Generate.php

@@ -10,6 +10,7 @@
 namespace OCA\PreviewGenerator\Command;
 
 use OCA\Files_External\Service\GlobalStoragesService;
+use OCA\PreviewGenerator\Model\WorkerConfig;
 use OCA\PreviewGenerator\SizeHelper;
 use OCP\Encryption\IManager;
 use OCP\Files\File;
@@ -32,6 +33,9 @@
 use Symfony\Component\Console\Output\OutputInterface;
 
 class Generate extends Command {
+	private const ENV_WORKER_CONF = 'PREVIEWGENERATOR_WORKER_CONF';
+	private const OPT_WORKERS = 'workers';
+
 	/* @return array{width: int, height: int, crop: bool} */
 	protected array $specifications;
 
@@ -44,6 +48,8 @@ class Generate extends Command {
 	protected IManager $encryptionManager;
 	protected SizeHelper $sizeHelper;
 
+	private ?WorkerConfig $workerConfig = null;
+
 	public function __construct(IRootFolder $rootFolder,
 		IUserManager $userManager,
 		IPreview $previewGenerator,
@@ -80,6 +86,11 @@ protected function configure(): void {
 				'p',
 				InputOption::VALUE_OPTIONAL | InputOption::VALUE_IS_ARRAY,
 				'limit scan to this path, eg. --path="/alice/files/Photos", the user_id is determined by the path and all user_id arguments are ignored, multiple usages allowed'
+			)->addOption(
+				self::OPT_WORKERS,
+				'w',
+				InputOption::VALUE_OPTIONAL,
+				'Spawn multiple parallel workers to increase speed of preview generation',
 			);
 	}
 
@@ -95,10 +106,76 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 		$this->output = $output;
 
 		$this->specifications = $this->sizeHelper->generateSpecifications();
-		if ($this->output->getVerbosity() > OutputInterface::VERBOSITY_VERY_VERBOSE) {
+		if ($this->output->getVerbosity() > OutputInterface::VERBOSITY_VERY_VERBOSE
+			&& !getenv(self::ENV_WORKER_CONF)
+		) {
 			$output->writeln('Specifications: ' . json_encode($this->specifications));
 		}
 
+		if (getenv(self::ENV_WORKER_CONF)) {
+			return $this->executeWorker($input);
+		}
+
+		if ($input->getOption(self::OPT_WORKERS)) {
+			return $this->executeCoordinator($input);
+		}
+
+		return $this->executeDefault($input);
+	}
+
+	private function executeCoordinator(InputInterface $input) {
+
+		$workerCount = (int)$input->getOption(self::OPT_WORKERS);
+		if ($workerCount <= 0) {
+			$this->output->writeln("<error>Invalid worker count: $workerCount</error>");
+			return 1;
+		}
+
+		$workerPids = [];
+		for ($i = 0; $i < $workerCount; $i++) {
+			$this->output->writeln("Spawning worker $i");
+
+			$workerconfig = new WorkerConfig($i, $workerCount);
+			$pid = pcntl_fork();
+			if ($pid == -1) {
+				$this->output->writeln('<error>Failed to fork worker</error>');
+				return 1;
+			} elseif ($pid) {
+				// Parent
+				$workerPids[] = $pid;
+			} else {
+				// Child
+				$argv = $_SERVER['argv'];
+				$env = getenv();
+				$env[self::ENV_WORKER_CONF] = json_encode($workerconfig, JSON_THROW_ON_ERROR);
+				pcntl_exec($argv[0], array_slice($argv, 1), $env);
+			}
+		}
+
+		$workerFailed = false;
+		foreach ($workerPids as $index => $pid) {
+			$status = 0;
+			pcntl_waitpid($pid, $status);
+			$exitCode = pcntl_wexitstatus($status);
+
+			if ($exitCode !== 0) {
+				$workerFailed = true;
+			}
+
+			$this->output->writeln("Worker $index exited with code $exitCode");
+		}
+
+		return $workerFailed ? 1 : 0;
+	}
+
+	private function executeWorker(InputInterface $input): int {
+		$workerConfigEnv = getenv(self::ENV_WORKER_CONF);
+		$data = json_decode($workerConfigEnv, true);
+		$this->workerConfig = WorkerConfig::fromJson($data);
+		return $this->executeDefault($input);
+	}
+
+	private function executeDefault(InputInterface $input): int {
 		$inputPaths = $input->getOption('path');
 		if ($inputPaths) {
 			foreach ($inputPaths as $inputPath) {
@@ -177,11 +254,15 @@ private function parseFolder(Folder $folder, array $noPreviewMountPaths): void {
 			// Respect the '.nomedia' file. If present don't traverse the folder
 			// Same for external mounts with previews disabled
 			if ($folder->nodeExists('.nomedia') || in_array($folderPath, $noPreviewMountPaths)) {
-				$this->output->writeln('Skipping folder ' . $folderPath);
+				if ($this->workerConfig === null) {
+					$this->output->writeln('Skipping folder ' . $folderPath);
+				}
 				return;
 			}
 
-			$this->output->writeln('Scanning folder ' . $folderPath);
+			if ($this->workerConfig === null) {
+				$this->output->writeln('Scanning folder ' . $folderPath);
+			}
 
 			$nodes = $folder->getDirectoryListing();
 
@@ -201,20 +282,48 @@ private function parseFolder(Folder $folder, array $noPreviewMountPaths): void {
 	}
 
 	private function parseFile(File $file): void {
-		if ($this->previewGenerator->isMimeSupported($file->getMimeType())) {
-			if ($this->output->getVerbosity() > OutputInterface::VERBOSITY_VERBOSE) {
-				$this->output->writeln('Generating previews for ' . $file->getPath());
+		if (!$this->previewGenerator->isMimeSupported($file->getMimeType())) {
+			return;
+		}
+
+		if ($this->workerConfig !== null) {
+			$hash = $this->hashFileId($file->getId());
+			if (($hash % $this->workerConfig->getWorkerCount()) !== $this->workerConfig->getWorkerIndex()) {
+				return;
 			}
+		}
 
-			try {
-				$this->previewGenerator->generatePreviews($file, $this->specifications);
-			} catch (NotFoundException $e) {
-				// Maybe log that previews could not be generated?
-			} catch (\InvalidArgumentException|GenericFileException $e) {
-				$class = $e::class;
-				$error = $e->getMessage();
-				$this->output->writeln("<error>{$class}: {$error}</error>");
+		if ($this->output->getVerbosity() > OutputInterface::VERBOSITY_VERBOSE) {
+			$prefix = '';
+			if ($this->workerConfig !== null) {
+				$workerIndex = $this->workerConfig->getWorkerIndex();
+				$prefix = "[WORKER $workerIndex] ";
 			}
+			$this->output->writeln("{$prefix}Generating previews for " . $file->getPath());
+		}
+
+		try {
+			$this->previewGenerator->generatePreviews($file, $this->specifications);
+		} catch (NotFoundException $e) {
+			// Maybe log that previews could not be generated?
+		} catch (\InvalidArgumentException|GenericFileException $e) {
+			$class = $e::class;
+			$error = $e->getMessage();
+			$this->output->writeln("<error>{$class}: {$error}</error>");
 		}
 	}
+
+	/**
+	 * Hash the given file id into an integer to ensure even distribution of work between workers.
+	 */
+	private function hashFileId(int $fileId): int {
+		// Fall back to 32 bit hash on 32 bit systems
+		if (PHP_INT_SIZE === 4) {
+			$digest = hash('xxh32', (string)$fileId, true);
+			return unpack('l', $digest)[1];
+		}
+
+		$digest = hash('xxh3', (string)$fileId, true);
+		return unpack('q', $digest)[1];
+	}
 }

lib/Model/WorkerConfig.php

@@ -0,0 +1,54 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2025 Richard Steinmetz
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace OCA\PreviewGenerator\Model;
+
+final class WorkerConfig implements \JsonSerializable {
+	private const WORKER_INDEX_KEY = 'workerIndex';
+	private const WORKER_COUNT_KEY = 'workerCount';
+
+	public function __construct(
+		private readonly int $workerIndex,
+		private readonly int $workerCount,
+	) {
+	}
+
+	/**
+	 * @throws \InvalidArgumentException If the given JSON data is not valid
+	 */
+	public static function fromJson(array $data): self {
+		$workerIndex = $data[self::WORKER_INDEX_KEY] ?? null;
+		if (!is_int($workerIndex)) {
+			throw new \InvalidArgumentException('Invalid worker data: Missing worker index');
+		}
+
+		$workerCount = $data[self::WORKER_COUNT_KEY] ?? null;
+		if (!is_int($workerCount)) {
+			throw new \InvalidArgumentException('Invalid worker data: Missing worker count');
+		}
+
+		return new self($workerIndex, $workerCount);
+	}
+
+	public function getWorkerIndex(): int {
+		return $this->workerIndex;
+	}
+
+	public function getWorkerCount(): int {
+		return $this->workerCount;
+	}
+
+	#[\ReturnTypeWillChange]
+	public function jsonSerialize() {
+		return [
+			self::WORKER_INDEX_KEY => $this->workerIndex,
+			self::WORKER_COUNT_KEY => $this->workerCount,
+		];
+	}
+}