diff --git a/lib/Command/Index.php b/lib/Command/Index.php index 5153e45ce..d15d73c3a 100644 --- a/lib/Command/Index.php +++ b/lib/Command/Index.php @@ -44,6 +44,7 @@ class IndexOpts public ?string $group = null; public bool $retry = false; public bool $skipCleanup = false; + public int $jobs = 1; public function __construct(InputInterface $input) { @@ -54,6 +55,7 @@ public function __construct(InputInterface $input) $this->retry = (bool) $input->getOption('retry'); $this->skipCleanup = (bool) $input->getOption('skip-cleanup'); $this->group = $input->getOption('group'); + $this->jobs = max(1, (int) ($input->getOption('jobs') ?? 1)); } } @@ -86,6 +88,7 @@ protected function configure(): void ->addOption('clear', null, InputOption::VALUE_NONE, 'Clear all existing index entries') ->addOption('retry', null, InputOption::VALUE_NONE, 'Retry indexing of failed files') ->addOption('skip-cleanup', null, InputOption::VALUE_NONE, 'Skip cleanup step (removing index entries with missing files)') + ->addOption('jobs', 'j', InputOption::VALUE_REQUIRED, 'Number of parallel indexing jobs (requires pcntl)', '1') ; } @@ -99,6 +102,13 @@ protected function execute(InputInterface $input, OutputInterface $output): int $this->output = $output; $this->opts = new IndexOpts($input); + // Check if parallel processing is requested but not available + if ($this->opts->jobs > 1 && !\extension_loaded('pcntl')) { + $this->output->writeln('Parallel processing requires the pcntl extension'); + $this->output->writeln('Falling back to single-threaded mode'.PHP_EOL); + $this->opts->jobs = 1; + } + // Assign to indexer $this->indexer->output = $output; $this->indexer->section = $output->section(); @@ -114,8 +124,12 @@ protected function execute(InputInterface $input, OutputInterface $output): int $this->checkForce(); $this->checkRetry(); - // Run the indexer - $this->runIndex(); + // Run the indexer (parallel or single-threaded) + if ($this->opts->jobs > 1) { + $this->runIndexParallel(); + } else { + $this->runIndex(); + } // Clean up the index if (!$this->opts->skipCleanup) { @@ -208,6 +222,279 @@ protected function runIndex(): void }); } + /** + * Run the indexer in parallel using multiple worker processes. + */ + protected function runIndexParallel(): void + { + $numJobs = $this->opts->jobs; + + // If user/path specified, use filtered approach; otherwise use fast DB query + if ($this->opts->user || $this->opts->path || $this->opts->group) { + $this->runIndexParallelFiltered(); + + return; + } + + $this->output->writeln('Querying database for files needing indexing...'); + + // Single database query to find all files needing indexing (fastest) + $allFileIds = $this->indexer->getFilesNeedingIndex(); + + $numFiles = \count($allFileIds); + if (0 === $numFiles) { + $this->output->writeln('No files need indexing'); + + return; + } + + // Partition file IDs among workers + $partitions = $this->partitionArray($allFileIds, $numJobs); + $actualJobs = \count(array_filter($partitions, static fn ($p) => !empty($p))); + + $this->output->writeln("Found {$numFiles} file(s) to index, using {$actualJobs} parallel job(s)"); + $this->output->writeln(''); + + // Reserve lines for worker status display + for ($i = 0; $i < $actualJobs; ++$i) { + $this->output->writeln("[Worker {$i}] Starting..."); + } + + // Close exiftool before forking - each child will create its own + \OCA\Memories\Exif::closeStaticExiftoolProc(); + + $pids = []; + $workerNum = 0; + + foreach ($partitions as $workerIndex => $fileIdPartition) { + if (empty($fileIdPartition)) { + continue; + } + + $pid = pcntl_fork(); + + if (-1 === $pid) { + $this->output->writeln('Failed to fork worker process'); + + continue; + } + + if (0 === $pid) { + // Child process - directly index assigned file IDs + // Pass the line offset for display (count from bottom of reserved area) + $lineOffset = $actualJobs - $workerNum; + $this->runWorker($workerNum, $fileIdPartition, $lineOffset); + exit(0); + } + + // Parent process + $pids[] = $pid; + ++$workerNum; + } + + // Wait for all children to complete + $exitCodes = []; + foreach ($pids as $pid) { + pcntl_waitpid($pid, $status); + $exitCodes[] = pcntl_wexitstatus($status); + } + + // Move cursor below the status area + $this->output->writeln(''); + + // Re-initialize exiftool for cleanup phase + \OCA\Memories\Exif::ensureStaticExiftoolProc(); + + // Report results + $failed = \count(array_filter($exitCodes, static fn ($code) => 0 !== $code)); + if ($failed > 0) { + $this->output->writeln("{$failed} worker(s) exited with errors"); + } + + $this->output->writeln('All workers finished'.PHP_EOL); + } + + /** + * Run parallel indexing with user/path filters (uses folder traversal). + */ + protected function runIndexParallelFiltered(): void + { + $users = $this->collectUsers(); + + if (empty($users)) { + $this->output->writeln('No users to index'); + + return; + } + + $numJobs = $this->opts->jobs; + $this->output->writeln('Scanning for files to index (filtered mode)...'); + + // Collect files by traversing folders (respects user/path filters) + $allFileIds = []; + foreach ($users as $user) { + try { + $userFiles = $this->indexer->getFilesForUser($user, $this->opts->path); + foreach ($userFiles as $fileId) { + $allFileIds[$fileId] = true; + } + } catch (\Exception $e) { + $this->output->writeln("Error scanning user {$user->getUID()}: {$e->getMessage()}"); + } + } + + $numFiles = \count($allFileIds); + if (0 === $numFiles) { + $this->output->writeln('No files need indexing'); + + return; + } + + $fileIdList = array_keys($allFileIds); + $partitions = $this->partitionArray($fileIdList, $numJobs); + $actualJobs = \count(array_filter($partitions, static fn ($p) => !empty($p))); + + $this->output->writeln("Found {$numFiles} file(s) to index, using {$actualJobs} parallel job(s)"); + $this->output->writeln(''); + + // Reserve lines for worker status display + for ($i = 0; $i < $actualJobs; ++$i) { + $this->output->writeln("[Worker {$i}] Starting..."); + } + + \OCA\Memories\Exif::closeStaticExiftoolProc(); + + $pids = []; + $workerNum = 0; + + foreach ($partitions as $workerIndex => $fileIdPartition) { + if (empty($fileIdPartition)) { + continue; + } + + $pid = pcntl_fork(); + if (-1 === $pid) { + $this->output->writeln('Failed to fork worker process'); + + continue; + } + + if (0 === $pid) { + $lineOffset = $actualJobs - $workerNum; + $this->runWorker($workerNum, $fileIdPartition, $lineOffset); + exit(0); + } + + $pids[] = $pid; + ++$workerNum; + } + + $exitCodes = []; + foreach ($pids as $pid) { + pcntl_waitpid($pid, $status); + $exitCodes[] = pcntl_wexitstatus($status); + } + + $this->output->writeln(''); + \OCA\Memories\Exif::ensureStaticExiftoolProc(); + + $failed = \count(array_filter($exitCodes, static fn ($code) => 0 !== $code)); + if ($failed > 0) { + $this->output->writeln("{$failed} worker(s) exited with errors"); + } + + $this->output->writeln('All workers finished'.PHP_EOL); + } + + /** + * Partition an array into n roughly equal chunks. + * + * @param array $array Array to partition + * @param int $numParts Number of partitions + * + * @return array> Partitioned arrays + */ + private function partitionArray(array $array, int $numParts): array + { + $count = \count($array); + if (0 === $count) { + return array_fill(0, $numParts, []); + } + + $partitions = []; + $chunkSize = (int) ceil($count / $numParts); + + for ($i = 0; $i < $numParts; ++$i) { + $partitions[$i] = \array_slice($array, $i * $chunkSize, $chunkSize); + } + + return $partitions; + } + + /** + * Run a worker process that indexes assigned files by ID. + * + * @param int $workerIndex Worker identifier + * @param array $fileIds File IDs to process + * @param int $lineOffset Line offset from cursor for status updates + */ + private function runWorker(int $workerIndex, array $fileIds, int $lineOffset): void + { + // Each worker needs its own exiftool process + \OCA\Memories\Exif::ensureStaticExiftoolProc(); + + try { + // Process files directly by ID - real-time line updates + $this->indexer->indexByIds($fileIds, $workerIndex, $lineOffset); + } catch (\Exception $e) { + $this->updateWorkerLine($lineOffset, "[Worker {$workerIndex}] Error: {$e->getMessage()}"); + } finally { + \OCA\Memories\Exif::closeStaticExiftoolProc(); + } + } + + /** + * Update a specific line in the terminal using ANSI escape codes. + * + * @param int $lineOffset Lines up from current cursor position + * @param string $content Content to display + */ + private function updateWorkerLine(int $lineOffset, string $content): void + { + // Move up, clear line, write content, move back down + fwrite(STDERR, "\033[{$lineOffset}A\r\033[K{$content}\033[{$lineOffset}B\r"); + } + + /** + * Collect all users that need to be indexed. + * + * @return IUser[] + */ + private function collectUsers(): array + { + $users = []; + + if ($uid = $this->opts->user) { + if ($user = $this->userManager->get($uid)) { + $users[] = $user; + } else { + $this->output->writeln("User {$uid} not found".PHP_EOL); + } + } elseif ($gid = $this->opts->group) { + if ($group = $this->groupManager->get($gid)) { + $users = array_values($group->getUsers()); + } else { + $this->output->writeln("Group {$gid} not found".PHP_EOL); + } + } else { + $this->userManager->callForSeenUsers(static function (IUser $user) use (&$users): void { + $users[] = $user; + }); + } + + return $users; + } + /** * Run function for all users (or selected user if set). * diff --git a/lib/Service/Index.php b/lib/Service/Index.php index e26398b1b..a4228ca1b 100644 --- a/lib/Service/Index.php +++ b/lib/Service/Index.php @@ -57,6 +57,7 @@ class Index */ public ?\Closure $continueCheck = null; + /** @var string[] */ private static ?array $mimeList = null; @@ -124,6 +125,329 @@ public function indexUser(IUser $user, ?string $path = null): void } } + /** + * Get all file IDs that need indexing from the database directly. + * This is much faster than folder traversal for parallel processing. + * + * @return array List of file IDs + */ + public function getFilesNeedingIndex(): array + { + $mimes = self::getMimeList(); + + // Get mime type IDs + $mimeQuery = $this->db->getQueryBuilder(); + $mimeQuery->select('id') + ->from('mimetypes') + ->where($mimeQuery->expr()->in('mimetype', $mimeQuery->createNamedParameter($mimes, IQueryBuilder::PARAM_STR_ARRAY))) + ; + $mimeIds = $mimeQuery->executeQuery()->fetchAll(\PDO::FETCH_COLUMN); + + if (empty($mimeIds)) { + return []; + } + + // Build main query for files needing indexing + $query = $this->db->getQueryBuilder(); + $query->select('f.fileid') + ->from('filecache', 'f') + ->where($query->expr()->in('f.mimetype', $query->createNamedParameter($mimeIds, IQueryBuilder::PARAM_INT_ARRAY))) + ->andWhere($query->expr()->gt('f.size', $query->expr()->literal(0))) + ; + + // Apply path blacklist pattern if configured + $blacklist = trim(SystemConfig::get('memories.index.path.blacklist') ?: ''); + if (!empty($blacklist)) { + // Note: This is a basic filter; complex regex patterns may need post-filtering + $query->andWhere($query->expr()->notLike('f.path', $query->createNamedParameter('%/.trashed-%'))); + } else { + // Always exclude trashed files + $query->andWhere($query->expr()->notLike('f.path', $query->createNamedParameter('%/.trashed-%'))); + } + + // Exclude files in .nomedia/.nomemories folders (approximate - check path contains) + // Full check happens during indexing + + // Filter out already indexed (non-orphaned, same mtime) + $getFilter = function (string $table, bool $notOrphaned) use (&$query): IQueryFunction { + $clause = $this->db->getQueryBuilder(); + $clause->select($clause->expr()->literal(1)) + ->from($table, 'a') + ->andWhere($clause->expr()->eq('f.fileid', 'a.fileid')) + ->andWhere($clause->expr()->eq('f.mtime', 'a.mtime')) + ; + if ($notOrphaned) { + $clause->andWhere($clause->expr()->eq('a.orphan', $clause->expr()->literal(0))); + } + + return SQL::notExists($query, $clause); + }; + + $query->andWhere($getFilter('memories', true)); + $query->andWhere($getFilter('memories_livephoto', true)); + $query->andWhere($getFilter('memories_failures', false)); + + return Util::transaction(static fn (): array => $query->executeQuery()->fetchAll(\PDO::FETCH_COLUMN)); + } + + /** + * Get file IDs needing indexing for a specific user via folder traversal. + * Used when user/path filters are specified. + * + * @return array List of file IDs + */ + public function getFilesForUser(IUser $user, ?string $path = null): array + { + if (!$this->appManager->isEnabledForUser('memories', $user)) { + return []; + } + + $uid = $user->getUID(); + + \OC_Util::tearDownFS(); + \OC_Util::setupFS($uid); + + $root = $this->rootFolder->getUserFolder($uid); + + // Get paths to scan + $mode = SystemConfig::get('memories.index.mode'); + if (null !== $path) { + $paths = [$path]; + } elseif ('1' === $mode || '0' === $mode) { + $paths = ['/']; + } elseif ('2' === $mode) { + $paths = Util::getTimelinePaths($uid); + } elseif ('3' === $mode) { + $paths = [SystemConfig::get('memories.index.path')]; + } else { + throw new \Exception('Invalid index mode'); + } + + $fileIds = []; + foreach ($paths as $scanPath) { + try { + $node = $root->get($scanPath); + } catch (\Exception $e) { + continue; + } + + if ($node instanceof Folder) { + $this->collectFolderFiles($node, $fileIds); + } elseif ($node instanceof File && self::isSupported($node)) { + $fileIds[] = $node->getId(); + } + } + + // Filter to only files needing indexing + if (empty($fileIds)) { + return []; + } + + return $this->filterFilesNeedingIndex($fileIds); + } + + /** + * Collect file IDs from a folder recursively. + * + * @param Folder $folder Folder to scan + * @param array $fileIds Array to populate + */ + private function collectFolderFiles(Folder $folder, array &$fileIds): void + { + $path = $folder->getPath(); + + if (!$this->isPathAllowed($path.'/')) { + return; + } + + if ($folder->nodeExists('.nomedia') || $folder->nodeExists('.nomemories')) { + return; + } + + $nodes = $folder->getDirectoryListing(); + $mimes = self::getMimeList(); + + foreach ($nodes as $node) { + if ($node instanceof File + && \in_array($node->getMimeType(), $mimes, true) + && self::isPathAllowed($node->getPath())) { + $fileIds[] = $node->getId(); + } elseif ($node instanceof Folder) { + $this->collectFolderFiles($node, $fileIds); + } + } + } + + /** + * Filter file IDs to only those needing indexing. + * + * @param array $fileIds File IDs to check + * + * @return array File IDs that need indexing + */ + private function filterFilesNeedingIndex(array $fileIds): array + { + $result = []; + $chunks = array_chunk($fileIds, 250); + + foreach ($chunks as $chunk) { + $query = $this->db->getQueryBuilder(); + $query->select('f.fileid') + ->from('filecache', 'f') + ->where($query->expr()->in('f.fileid', $query->createNamedParameter($chunk, IQueryBuilder::PARAM_INT_ARRAY))) + ->andWhere($query->expr()->gt('f.size', $query->expr()->literal(0))) + ; + + $getFilter = function (string $table, bool $notOrphaned) use (&$query): IQueryFunction { + $clause = $this->db->getQueryBuilder(); + $clause->select($clause->expr()->literal(1)) + ->from($table, 'a') + ->andWhere($clause->expr()->eq('f.fileid', 'a.fileid')) + ->andWhere($clause->expr()->eq('f.mtime', 'a.mtime')) + ; + if ($notOrphaned) { + $clause->andWhere($clause->expr()->eq('a.orphan', $clause->expr()->literal(0))); + } + + return SQL::notExists($query, $clause); + }; + + $query->andWhere($getFilter('memories', true)); + $query->andWhere($getFilter('memories_livephoto', true)); + $query->andWhere($getFilter('memories_failures', false)); + + $ids = Util::transaction(static fn (): array => $query->executeQuery()->fetchAll(\PDO::FETCH_COLUMN)); + foreach ($ids as $id) { + $result[] = (int) $id; + } + } + + return $result; + } + + /** + * Index files by their IDs directly (no folder traversal). + * + * @param array $fileIds File IDs to index + * @param int $workerId Worker ID for display (-1 = single mode) + * @param int $lineOffset Line offset for ANSI cursor positioning (0 = no positioning) + */ + public function indexByIds(array $fileIds, int $workerId = -1, int $lineOffset = 0): void + { + $total = \count($fileIds); + $processed = 0; + $indexed = 0; + $errors = 0; + $startTime = microtime(true); + $lastUpdate = $startTime; + + $parallelMode = $workerId >= 0; + + // Update display helper + $updateStatus = function (bool $final = false) use ($workerId, $lineOffset, $parallelMode, &$processed, &$indexed, &$errors, $total, $startTime): void { + $elapsed = max(0.1, microtime(true) - $startTime); + $rate = round($processed / $elapsed, 1); + $pct = $total > 0 ? round($processed / $total * 100, 1) : 0; + + if ($parallelMode && $lineOffset > 0) { + // Update specific line using ANSI codes + $status = $final ? 'Done' : 'Working'; + $errStr = $errors > 0 ? " ({$errors} errors)" : ''; + $content = "[Worker {$workerId}] {$status}: {$processed}/{$total} ({$pct}%) | {$indexed} indexed | {$rate}/s{$errStr}"; + $this->updateLine($lineOffset, $content); + } elseif (!$parallelMode) { + $this->log("Indexing file {$processed}/{$total}", true); + } + }; + + foreach ($fileIds as $fileId) { + $this->ensureContinueOk(); + ++$processed; + + // Update display every 100ms or every 100 files in parallel mode + $now = microtime(true); + if ($parallelMode) { + if ($now - $lastUpdate >= 0.1 || $processed % 100 === 0) { + $updateStatus(); + $lastUpdate = $now; + } + } else { + $updateStatus(); + } + + try { + // Look up file by ID + $nodes = $this->rootFolder->getById($fileId); + if (empty($nodes)) { + continue; // File no longer exists + } + + $file = $nodes[0]; + if (!($file instanceof File)) { + continue; + } + + // Check path exclusions (.nomedia, .nomemories, blacklist) + if (!$this->isFileAllowed($file)) { + continue; + } + + $this->indexFile($file); + ++$indexed; + } catch (\OCP\Lock\LockedException $e) { + // Skip silently in parallel mode + if (!$parallelMode) { + $this->log("Skipping file {$fileId} due to lock", true); + } + } catch (\Exception $e) { + ++$errors; + if (!$parallelMode) { + $this->error("Failed to index file {$fileId}: {$e->getMessage()}"); + } + } + } + + // Final status update + $updateStatus(true); + } + + /** + * Update a specific terminal line using ANSI escape codes. + */ + private function updateLine(int $lineOffset, string $content): void + { + // Save position, move up N lines, clear line, write, restore position + fwrite(STDERR, "\033[s\033[{$lineOffset}A\r\033[K{$content}\033[u"); + } + + /** + * Check if a file is allowed to be indexed (path checks). + */ + private function isFileAllowed(File $file): bool + { + $path = $file->getPath(); + + // Check path blacklist + if (!self::isPathAllowed($path)) { + return false; + } + + // Check for .nomedia/.nomemories in parent folders + $parent = $file->getParent(); + while ($parent instanceof Folder) { + try { + if ($parent->nodeExists('.nomedia') || $parent->nodeExists('.nomemories')) { + return false; + } + $parent = $parent->getParent(); + } catch (\Exception $e) { + break; + } + } + + return true; + } + /** * Index all files in a folder. *