fix(ClusterFaces): Fix memory usage and add concurrency limits

.github/workflows/cluster-faces-test.yml

@@ -497,6 +497,13 @@ jobs:
           sudo docker image prune --all --force || true
           echo "disk after cleanup: $(df -h / | awk 'NR==2 {print $4" free ("$5" used)"}')"
 
+      - name: Checkout app_api
+        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+        with:
+          path: apps/app_api
+          repository: nextcloud/app_api
+          ref: ${{ matrix.server-versions == 'master' && 'main' || matrix.server-versions }}
+
       - name: Enable app_api
         run: ./occ app:enable -vvv -f app_api
 
@@ -619,10 +626,10 @@ jobs:
           # up the runner disk; swap in the CPU build like the full-run test does.
           sed -i 's/^onnxruntime-gpu.*/onnxruntime/' requirements.txt
           python3 -m pip install --upgrade pip
-          # The default PyPI torch wheels bundle ~5GB of NVIDIA CUDA libraries and
-          # blow up the runner disk. Install the CPU-only build first so the torch
-          # constraints in requirements.txt are already satisfied.
-          python3 -m pip install --index-url https://download.pytorch.org/whl/cpu torch torchvision torchaudio
+          # requirements.txt pins the CUDA torch builds (+cu128 from the cu128 index),
+          # which pull in ~5GB of NVIDIA libraries and blow up the runner disk.
+          # Rewrite the pins to the matching CPU-only builds.
+          sed -i -e 's#/whl/cu128#/whl/cpu#' -e 's/+cu128$/+cpu/' requirements.txt
           python3 -m pip install -r requirements.txt
 
       - name: Run ExApp
@@ -635,6 +642,9 @@ jobs:
 
       - name: Register ExApp with app_api
         if: steps.db-cache.outputs.cache-hit != 'true'
+        # --wait-finish blocks forever if the ExApp never comes up; fail fast so
+        # the "Show ExApp logs" step can tell us why.
+        timeout-minutes: 15
         run: |
           ./occ app_api:daemon:register --net host manual_install "Manual Install" manual-install http localhost http://localhost:8080
           ./occ app_api:app:register recognize_backend manual_install --json-info "{\"appid\":\"recognize_backend\",\"name\":\"Recognize Backend\",\"daemon_config_name\":\"manual_install\",\"version\":\"${{ steps.backendinfo.outputs.result }}\",\"secret\":\"12345\",\"port\":9031,\"scopes\":[\"TASK_PROCESSING\",\"FILES\"]}" --force-scopes --wait-finish

.github/workflows/full-run-test.yml

@@ -361,10 +361,10 @@ jobs:
         run: |
           # onnxruntime-gpu needs CUDA libs at import time; swap in the CPU build for the runner
           sed -i 's/^onnxruntime-gpu.*/onnxruntime/' requirements.txt
-          # The default PyPI torch wheels bundle ~5GB of NVIDIA CUDA libraries and
-          # blow up the runner disk. Install the CPU-only build first so the torch
-          # constraints in requirements.txt are already satisfied.
-          pip install --index-url https://download.pytorch.org/whl/cpu torch torchvision torchaudio
+          # requirements.txt pins the CUDA torch builds (+cu128 from the cu128 index),
+          # which pull in ~5GB of NVIDIA libraries and blow up the runner disk.
+          # Rewrite the pins to the matching CPU-only builds.
+          sed -i -e 's#/whl/cu128#/whl/cpu#' -e 's/+cu128$/+cpu/' requirements.txt
           pip install -r requirements.txt
 
       - name: Cache recognize_backend models

lib/BackgroundJobs/ClusterFacesJob.php

@@ -13,43 +13,122 @@
 use OCP\AppFramework\Utility\ITimeFactory;
 use OCP\BackgroundJob\IJobList;
 use OCP\BackgroundJob\QueuedJob;
+use OCP\Lock\ILockingProvider;
+use OCP\Lock\LockedException;
 use Psr\Log\LoggerInterface;
 
 final class ClusterFacesJob extends QueuedJob {
+	public const DEFAULT_BATCH_SIZE = 10_000;
+	// Peak memory of FaceClusterAnalyzer::calculateClusters is roughly
+	// QUADRATIC_BYTES * n² + LINEAR_BYTES * n for n detections. The quadratic part is
+	// the interpoint distance cache in MrdBallTree (measured: ~3.4GB for n=10,000 on real data)
+	public const QUADRATIC_BYTES = 34;
+	public const LINEAR_BYTES = 8_000;
+	// Batch size per byte of memory_limit that is known to work for low memory limits
+	public const LINEAR_BATCH_SIZE_PER_BYTE = 5_000 / 1_200_000_000;
+	// Share of the free memory we allow clustering to use
+	public const MEMORY_HEADROOM = 0.7;
+	// Seconds to wait before retrying when another ClusterFacesJob for the same user is running or all slots are taken
+	public const RETRY_DELAY = 10 * 60;
+
 	private FaceClusterAnalyzer $clusterAnalyzer;
 	private IJobList $jobList;
 	private LoggerInterface $logger;
 	private SettingsService $settingsService;
+	private ILockingProvider $lockingProvider;
 
-	public function __construct(ITimeFactory $time, Logger $logger, IJobList $jobList, FaceClusterAnalyzer $clusterAnalyzer, SettingsService $settingsService) {
+	public function __construct(ITimeFactory $time, Logger $logger, IJobList $jobList, FaceClusterAnalyzer $clusterAnalyzer, SettingsService $settingsService, ILockingProvider $lockingProvider) {
 		parent::__construct($time);
 		$this->logger = $logger;
 		$this->jobList = $jobList;
 		$this->clusterAnalyzer = $clusterAnalyzer;
 		$this->settingsService = $settingsService;
+		$this->lockingProvider = $lockingProvider;
 	}
 
 	/**
 	 * @param array{storageId: int, rootId: int, userId: string} $argument
 	 */
 	protected function run($argument): void {
 		$userId = (string)$argument['userId'];
+
+		// QueuedJob removes itself from the job list before running, so new face detections
+		// can schedule another ClusterFacesJob for this user while this one is still running.
+		// Make sure only one runs per user at a time; the lock is released on shutdown even after a fatal error.
+		$userLockKey = 'recognize/cluster-faces/' . $userId;
 		try {
-			$iniValue = ini_get('memory_limit');
-			if ($iniValue === false || $iniValue === '') {
-				$batchSize = 10_000;
-			} else {
-				$memoryBytes = ini_parse_quantity($iniValue);
-				if ($memoryBytes === -1) {
-					$batchSize = 10_000;
-				} else {
-					$batchSize = (int)($memoryBytes * 5_000 / 120_000_0000);
-				}
-			}
-			$this->clusterAnalyzer->calculateClusters($userId, $batchSize);
+			$this->lockingProvider->acquireLock($userLockKey, ILockingProvider::LOCK_EXCLUSIVE);
+		} catch (LockedException $e) {
+			$this->logger->debug('ClusterFacesJob for user ' . $userId . ' is already running, rescheduling');
+			// Re-add the job so that detections added after the running job has fetched its batch are not left behind,
+			// but delay it to avoid cron workers picking it up again and again while the other job is still running
+			$this->jobList->scheduleAfter(self::class, $this->time->getTime() + self::RETRY_DELAY, $argument);
+			return;
+		}
+
+		// Clustering is memory hungry, so also limit how many users are clustered at the same time
+		$slotLockKey = $this->acquireSlot();
+		if ($slotLockKey === null) {
+			$this->lockingProvider->releaseLock($userLockKey, ILockingProvider::LOCK_EXCLUSIVE);
+			$this->logger->debug('Maximum number of concurrent ClusterFacesJobs reached, rescheduling job for user ' . $userId);
+			$this->jobList->scheduleAfter(self::class, $this->time->getTime() + self::RETRY_DELAY, $argument);
+			return;
+		}
+
+		$moreDetectionsLeft = false;
+		try {
+			$moreDetectionsLeft = $this->clusterAnalyzer->calculateClusters($userId, self::getBatchSize());
 		} catch (\Throwable $e) {
 			$this->settingsService->setSetting('clusterFaces.status', 'false');
 			$this->logger->error('Failed to calculate face clusters', ['exception' => $e]);
+		} finally {
+			$this->lockingProvider->releaseLock($slotLockKey, ILockingProvider::LOCK_EXCLUSIVE);
+			$this->lockingProvider->releaseLock($userLockKey, ILockingProvider::LOCK_EXCLUSIVE);
+		}
+
+		if ($moreDetectionsLeft) {
+			// Process the next batch right away instead of waiting for MaintenanceJob
+			$this->jobList->add(self::class, $argument);
+		}
+	}
+
+	/**
+	 * Try to acquire one of the clusterFaces.concurrency global slots
+	 *
+	 * @return string|null The key of the acquired slot lock or null if all slots are taken
+	 */
+	private function acquireSlot(): ?string {
+		$concurrency = max(1, (int)$this->settingsService->getSetting('clusterFaces.concurrency'));
+		for ($i = 0; $i < $concurrency; $i++) {
+			$slotLockKey = 'recognize/cluster-faces-slot/' . $i;
+			try {
+				$this->lockingProvider->acquireLock($slotLockKey, ILockingProvider::LOCK_EXCLUSIVE);
+				return $slotLockKey;
+			} catch (LockedException $e) {
+				continue;
+			}
+		}
+		return null;
+	}
+
+	/**
+	 * Determine how many face detections fit into the memory available to this process
+	 */
+	public static function getBatchSize(): int {
+		$iniValue = ini_get('memory_limit');
+		if ($iniValue === false || $iniValue === '') {
+			return self::DEFAULT_BATCH_SIZE;
+		}
+		$memoryBytes = ini_parse_quantity($iniValue);
+		if ($memoryBytes <= 0) {
+			return self::DEFAULT_BATCH_SIZE;
 		}
+		$budget = max(0, $memoryBytes - memory_get_usage(true)) * self::MEMORY_HEADROOM;
+		// Solve QUADRATIC_BYTES * n² + LINEAR_BYTES * n = budget for n
+		$a = self::QUADRATIC_BYTES;
+		$b = self::LINEAR_BYTES;
+		$quadraticBatchSize = (int)((-$b + sqrt($b * $b + 4 * $a * $budget)) / (2 * $a));
+		$linearBatchSize = (int)($memoryBytes * self::LINEAR_BATCH_SIZE_PER_BYTE);
+		return max(FaceClusterAnalyzer::MIN_DATASET_SIZE, min($linearBatchSize, $quadraticBatchSize));
 	}
 }

lib/Command/ClusterFaces.php

@@ -65,7 +65,9 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 		foreach ($userIds as $userId) {
 			$this->logger->info('Clustering face detections for user ' . $userId);
 			try {
-				$this->clusterAnalyzer->calculateClusters($userId, (int)$input->getOption('batch-size'));
+				do {
+					$moreDetectionsLeft = $this->clusterAnalyzer->calculateClusters($userId, (int)$input->getOption('batch-size'));
+				} while ($moreDetectionsLeft);
 			} catch (\JsonException|Exception $e) {
 				$this->settingsService->setSetting('clusterFaces.status', 'false');
 				$this->logger->error($e->getMessage(), ['exception' => $e]);

lib/Service/FaceClusterAnalyzer.php

@@ -66,10 +66,11 @@ public function setMinDatasetSize(int $minSize) : void {
 	}
 
 	/**
+	 * @return bool Whether there may be more unclustered detections left for this user that did not fit into this batch
 	 * @throws \OCP\DB\Exception
 	 * @throws \JsonException
 	 */
-	public function calculateClusters(string $userId, int $batchSize = 0): void {
+	public function calculateClusters(string $userId, int $batchSize = 0): bool {
 		$this->logger->debug('ClusterDebug: Retrieving face detections for user ' . $userId);
 
 		if ($batchSize === 0) {
@@ -94,13 +95,15 @@ public function calculateClusters(string $userId, int $batchSize = 0): void {
 			$this->logger->warning('ClusterDebug: The batch size of ' . $batchSize . ' detections is too small for the ' . count($existingClusters) . ' existing clusters of user ' . $userId . '; loading ' . count($sampledDetections) . ' reference detections instead. Consider raising the PHP memory limit.');
 		}
 
+		$batchFull = false;
 		if ($batchSize > 0) {
 			$rejectedDetections = $this->faceDetections->sampleRejectedDetectionsByUserId($userId, $this->getRejectSampleSize($batchSize), $this->minDetectionSize, $this->minDetectionSize);
 			// Guarantee forward progress even when samples and rejects have eaten the whole
 			// budget, but keep the floor relative so a small batch size stays a small batch.
 			$freshDetectionFloor = min(500, (int)round($batchSize * (1.0 - self::REFERENCE_SAMPLE_BUDGET_SHARE)));
 			$requestedFreshDetectionCount = max($batchSize - count($rejectedDetections) - count($sampledDetections), $freshDetectionFloor);
 			$freshDetections = $this->faceDetections->findUnclusteredByUserId($userId, $requestedFreshDetectionCount, $this->minDetectionSize, $this->minDetectionSize);
+			$batchFull = count($freshDetections) >= $requestedFreshDetectionCount;
 		} else {
 			$freshDetections = $this->faceDetections->findUnclusteredByUserId($userId, 0, $this->minDetectionSize, $this->minDetectionSize);
 			$rejectedDetections = $this->faceDetections->sampleRejectedDetectionsByUserId($userId, $this->getRejectSampleSize(count($freshDetections)), $this->minDetectionSize, $this->minDetectionSize);
@@ -112,7 +115,7 @@ public function calculateClusters(string $userId, int $batchSize = 0): void {
 
 		if (count($detections) < $this->minDatasetSize || count($freshDetections) === 0) {
 			$this->logger->debug('ClusterDebug: Not enough face detections found');
-			return;
+			return false;
 		}
 
 		$this->logger->debug('ClusterDebug: Found ' . count($freshDetections) . " fresh detections. Adding " . count($rejectedDetections). " old detections and " . count($sampledDetections). " sampled detections from already existing clusters. Calculating clusters on " . count($detections) . " detections.");
@@ -214,6 +217,7 @@ public function calculateClusters(string $userId, int $batchSize = 0): void {
 
 		$this->settingsService->setSetting('clusterFaces.status', 'true');
 		$this->settingsService->setSetting('clusterFaces.lastRun', (string)time());
+		return $batchFull;
 	}
 
 	/**

lib/Service/SettingsService.php

@@ -60,6 +60,7 @@ final class SettingsService {
 		'nice_binary' => '',
 		'nice_value' => '0',
 		'concurrency.enabled' => 'false',
+		'clusterFaces.concurrency' => '1',
 		'ffmpeg_binary' => '',
 	];
 
@@ -94,7 +95,8 @@ final class SettingsService {
 		'landmarks.batchSize',
 		'movinet.batchSize',
 		'musicnn.batchSize',
-		'concurrency.enabled'
+		'concurrency.enabled',
+		'clusterFaces.concurrency',
 	];
 
 	private IAppConfig $config;

src/components/ViewAdmin.vue

@@ -87,6 +87,15 @@
 					:label="t('recognize', 'The number of files to process per job run (A job will be scheduled every 5 minutes; For normal operation ~500 or more, in WASM mode ~50 is recommended)')"
 					:title="t('recognize', 'The number of files to process per job run (A job will be scheduled every 5 minutes; For normal operation ~500 or more, in WASM mode ~50 is recommended)')"
 					@update:model-value="onChange" />
+				<NcTextField :disabled="!settings['faces.enabled']"
+					v-model="settings['clusterFaces.concurrency']"
+					type="number"
+					:min="1"
+					:step="1"
+					:label-visible="true"
+					:label="t('recognize', 'The maximum number of face clustering processes to run at the same time (Each one can take up to the PHP memory limit)')"
+					:title="t('recognize', 'The maximum number of face clustering processes to run at the same time (Each one can take up to the PHP memory limit)')"
+					@update:model-value="onChange" />
 			</p>
 		</NcSettingsSection>
 		<NcSettingsSection :name="t('recognize', 'Object detection & landmark recognition')">
@@ -445,7 +454,7 @@ const TASK_PROCESSING_TASK_TYPES = {
 	musicnn: 'recognize:audio:classification',
 }
 
-const SETTINGS = ['tensorflow.cores', 'tensorflow.gpu', 'tensorflow.purejs', 'imagenet.enabled', 'landmarks.enabled', 'faces.enabled', 'musicnn.enabled', 'movinet.enabled', 'node_binary', 'ffmpeg_binary', 'faces.status', 'imagenet.status', 'landmarks.status', 'movinet.status', 'musicnn.status', 'faces.lastFile', 'imagenet.lastFile', 'landmarks.lastFile', 'movinet.lastFile', 'musicnn.lastFile', 'faces.batchSize', 'imagenet.batchSize', 'landmarks.batchSize', 'movinet.batchSize', 'musicnn.batchSize', 'clusterFaces.status', 'clusterFaces.lastRun', 'nice_binary', 'nice_value', 'concurrency.enabled', 'taskprocessing.enabled']
+const SETTINGS = ['tensorflow.cores', 'tensorflow.gpu', 'tensorflow.purejs', 'imagenet.enabled', 'landmarks.enabled', 'faces.enabled', 'musicnn.enabled', 'movinet.enabled', 'node_binary', 'ffmpeg_binary', 'faces.status', 'imagenet.status', 'landmarks.status', 'movinet.status', 'musicnn.status', 'faces.lastFile', 'imagenet.lastFile', 'landmarks.lastFile', 'movinet.lastFile', 'musicnn.lastFile', 'faces.batchSize', 'imagenet.batchSize', 'landmarks.batchSize', 'movinet.batchSize', 'musicnn.batchSize', 'clusterFaces.status', 'clusterFaces.lastRun', 'nice_binary', 'nice_value', 'concurrency.enabled', 'clusterFaces.concurrency', 'taskprocessing.enabled']
 
 const BOOLEAN_SETTINGS = ['tensorflow.gpu', 'tensorflow.purejs', 'imagenet.enabled', 'landmarks.enabled', 'faces.enabled', 'musicnn.enabled', 'movinet.enabled', 'faces.status', 'imagenet.status', 'landmarks.status', 'movinet.status', 'musicnn.status', 'faces.lastFile', 'imagenet.lastFile', 'landmarks.lastFile', 'movinet.lastFile', 'musicnn.lastFile', 'clusterFaces.status', 'concurrency.enabled', 'taskprocessing.enabled']