diff --git a/.github/workflows/cluster-faces-test.yml b/.github/workflows/cluster-faces-test.yml
index 66c0b618..771dee20 100644
--- a/.github/workflows/cluster-faces-test.yml
+++ b/.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
diff --git a/.github/workflows/full-run-test.yml b/.github/workflows/full-run-test.yml
index e01d9268..0cef06a5 100644
--- a/.github/workflows/full-run-test.yml
+++ b/.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
diff --git a/lib/BackgroundJobs/ClusterFacesJob.php b/lib/BackgroundJobs/ClusterFacesJob.php
index 81908657..110dff7c 100644
--- a/lib/BackgroundJobs/ClusterFacesJob.php
+++ b/lib/BackgroundJobs/ClusterFacesJob.php
@@ -13,20 +13,37 @@
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;
}
/**
@@ -34,22 +51,84 @@ public function __construct(ITimeFactory $time, Logger $logger, IJobList $jobLis
*/
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));
}
}
diff --git a/lib/Command/ClusterFaces.php b/lib/Command/ClusterFaces.php
index 27faf7f9..feef64ab 100644
--- a/lib/Command/ClusterFaces.php
+++ b/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]);
diff --git a/lib/Service/FaceClusterAnalyzer.php b/lib/Service/FaceClusterAnalyzer.php
index 1376f0e6..28151503 100644
--- a/lib/Service/FaceClusterAnalyzer.php
+++ b/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,6 +95,7 @@ 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
@@ -101,6 +103,7 @@ public function calculateClusters(string $userId, int $batchSize = 0): void {
$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;
}
/**
diff --git a/lib/Service/SettingsService.php b/lib/Service/SettingsService.php
index 8ea9d217..adfeb1fd 100644
--- a/lib/Service/SettingsService.php
+++ b/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;
diff --git a/src/components/ViewAdmin.vue b/src/components/ViewAdmin.vue
index 21e1d894..aea7418d 100644
--- a/src/components/ViewAdmin.vue
+++ b/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" />
+
@@ -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']