Merge pull request #8005 from cvat-ai/release-2.14.2

Release v2.14.2

CHANGELOG.md

@@ -16,6 +16,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- scriv-insert-here -->
 
+<a id='changelog-2.14.2'></a>
+## \[2.14.2\] - 2024-06-07
+
+### Fixed
+
+- Queued jobs are not considered in deferring logic
+  (<https://github.com/cvat-ai/cvat/pull/7907>)
+
+- Significant memory leak related to the frames, which did not memory after become unused
+  (<https://github.com/cvat-ai/cvat/pull/7995>)
+
 <a id='changelog-2.14.1'></a>
 ## \[2.14.1\] - 2024-06-05
 

cvat-canvas/package.json

@@ -1,6 +1,6 @@
 {
   "name": "cvat-canvas",
-  "version": "2.20.2",
+  "version": "2.20.3",
   "type": "module",
   "description": "Part of Computer Vision Annotation Tool which presents its canvas library",
   "main": "src/canvas.ts",

cvat-canvas/src/typescript/canvasView.ts

@@ -451,6 +451,7 @@ export class CanvasViewImpl implements CanvasView, Listener {
                     results.forEach((bitmap, idx) => {
                         const [curLeft, curTop] = objects[idx].points.slice(-4, -2);
                         canvas.getContext('2d').drawImage(bitmap, curLeft - left, curTop - top);
+                        bitmap.close();
                     });
 
                     const imageData = canvas.getContext('2d')

cvat-cli/requirements/base.txt

@@ -1,3 +1,3 @@
-cvat-sdk~=2.14.1
+cvat-sdk~=2.14.2
 Pillow>=10.3.0
 setuptools>=70.0.0 # not directly required, pinned by Snyk to avoid a vulnerability

cvat-cli/src/cvat_cli/version.py

@@ -1 +1 @@
-VERSION = "2.14.1"
+VERSION = "2.14.2"

cvat-core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "cvat-core",
-  "version": "15.0.5",
+  "version": "15.0.6",
   "type": "module",
   "description": "Part of Computer Vision Tool which presents an interface for client-side integration",
   "main": "src/api.ts",

cvat-core/src/frames.ts

@@ -295,8 +295,8 @@ Object.defineProperty(FrameData.prototype.data, 'implementation', {
                     const nextChunkNumber = findTheNextNotDecodedChunk(this.number);
                     const predecodeChunksMax = Math.floor(decodedBlocksCacheSize / 2);
                     if (nextChunkNumber * chunkSize <= stopFrame &&
-                        nextChunkNumber <= chunkNumber + predecodeChunksMax) {
-                        provider.cleanup(1);
+                        nextChunkNumber <= chunkNumber + predecodeChunksMax
+                    ) {
                         frameDataCache[this.jobID].activeChunkRequest = new Promise((resolveForward) => {
                             const releasePromise = (): void => {
                                 resolveForward();
@@ -306,6 +306,14 @@ Object.defineProperty(FrameData.prototype.data, 'implementation', {
                             frameDataCache[this.jobID].getChunk(
                                 nextChunkNumber, ChunkQuality.COMPRESSED,
                             ).then((chunk: ArrayBuffer) => {
+                                if (!(this.jobID in frameDataCache)) {
+                                    // check if frameDataCache still exist
+                                    // as it may be released during chunk request
+                                    resolveForward();
+                                    return;
+                                }
+
+                                provider.cleanup(1);
                                 provider.requestDecodeBlock(
                                     chunk,
                                     nextChunkNumber * chunkSize,
@@ -333,13 +341,13 @@ Object.defineProperty(FrameData.prototype.data, 'implementation', {
             onServerRequest();
             frameDataCache[this.jobID].latestFrameDecodeRequest = requestId;
             (frameDataCache[this.jobID].activeChunkRequest || Promise.resolve()).finally(() => {
-                if (frameDataCache[this.jobID].latestFrameDecodeRequest !== requestId) {
+                if (frameDataCache[this.jobID]?.latestFrameDecodeRequest !== requestId) {
                     // not relevant request anymore
                     reject(this.number);
                     return;
                 }
 
-                // it might appear during decoding, so, check again
+                // it might appear during previous decoding, so, check again
                 const currentFrame = provider.frame(this.number);
                 if (currentFrame) {
                     resolve({
@@ -359,6 +367,14 @@ Object.defineProperty(FrameData.prototype.data, 'implementation', {
                         chunkNumber, ChunkQuality.COMPRESSED,
                     ).then((chunk: ArrayBuffer) => {
                         try {
+                            if (!(this.jobID in frameDataCache)) {
+                                // check if frameDataCache still exist
+                                // as it may be released during chunk request
+                                resolveLoadAndDecode();
+                                reject(this.number);
+                                return;
+                            }
+
                             provider
                                 .requestDecodeBlock(
                                     chunk,
@@ -703,6 +719,13 @@ export function getCachedChunks(jobID): number[] {
 
 export function clear(jobID: number): void {
     if (jobID in frameDataCache) {
+        frameDataCache[jobID].provider.close();
+        for (const contextImagesByFrame of Object.values(frameDataCache[jobID].contextCache)) {
+            for (const image of Object.values(contextImagesByFrame.data)) {
+                image.close();
+            }
+        }
+
         delete frameDataCache[jobID];
         delete frameMetaCache[jobID];
     }

cvat-data/package.json

@@ -1,7 +1,7 @@
 {
   "name": "cvat-data",
   "type": "module",
-  "version": "2.0.0",
+  "version": "2.1.0",
   "description": "",
   "main": "src/ts/cvat-data.ts",
   "scripts": {

cvat-data/src/ts/cvat-data.ts

@@ -98,7 +98,8 @@ export class FrameDecoder {
     // used for video chunks to get correct side after decoding
     private renderWidth: number;
     private renderHeight: number;
-    private zipWorker: Worker;
+    private zipWorker: Worker | null;
+    private videoWorker: Worker | null;
 
     constructor(
         blockType: BlockType,
@@ -108,6 +109,8 @@ export class FrameDecoder {
     ) {
         this.mutex = new Mutex();
         this.orderedStack = [];
+        this.zipWorker = null;
+        this.videoWorker = null;
 
         this.cachedChunksLimit = Math.max(1, cachedBlockCount);
         this.dimension = dimension;
@@ -139,6 +142,14 @@ export class FrameDecoder {
             if (typeof lastChunk === 'undefined') {
                 return;
             }
+
+            for (const frame of Object.keys(this.decodedChunks[lastChunk])) {
+                const data = this.decodedChunks[lastChunk][frame];
+                if (data instanceof ImageBitmap) {
+                    data.close();
+                }
+            }
+
             delete this.decodedChunks[lastChunk];
             length--;
         }
@@ -232,7 +243,15 @@ export class FrameDecoder {
 
     async startDecode(): Promise<void> {
         const blockToDecode = { ...this.requestedChunkToDecode };
-        const release = await this.mutex.acquire();
+        const releaseMutex = await this.mutex.acquire();
+        const release = (): void => {
+            if (this.videoWorker) {
+                this.videoWorker.terminate();
+                this.videoWorker = null;
+            }
+
+            releaseMutex();
+        };
         try {
             const { start, end, block } = this.requestedChunkToDecode;
             if (start !== blockToDecode.start) {
@@ -251,12 +270,12 @@ export class FrameDecoder {
             this.requestedChunkToDecode = null;
 
             if (this.blockType === BlockType.MP4VIDEO) {
-                const worker = new Worker(
+                this.videoWorker = new Worker(
                     new URL('./3rdparty/Decoder.worker', import.meta.url),
                 );
                 let index = start;
 
-                worker.onmessage = (e) => {
+                this.videoWorker.onmessage = (e) => {
                     if (e.data.consoleLog) {
                         // ignore initialization message
                         return;
@@ -283,22 +302,20 @@ export class FrameDecoder {
                             this.decodedChunks[chunkNumber] = decodedFrames;
                             this.chunkIsBeingDecoded.onDecodeAll();
                             this.chunkIsBeingDecoded = null;
-                            worker.terminate();
                             release();
                         }
                     });
 
                     index++;
                 };
 
-                worker.onerror = (event: ErrorEvent) => {
+                this.videoWorker.onerror = (event: ErrorEvent) => {
                     release();
-                    worker.terminate();
                     this.chunkIsBeingDecoded.onReject(event.error);
                     this.chunkIsBeingDecoded = null;
                 };
 
-                worker.postMessage({
+                this.videoWorker.postMessage({
                     type: 'Broadway.js - Worker init',
                     options: {
                         rgb: true,
@@ -314,12 +331,12 @@ export class FrameDecoder {
                 const sps = avc.sps[0];
                 const pps = avc.pps[0];
 
-                worker.postMessage({ buf: sps, offset: 0, length: sps.length });
-                worker.postMessage({ buf: pps, offset: 0, length: pps.length });
+                this.videoWorker.postMessage({ buf: sps, offset: 0, length: sps.length });
+                this.videoWorker.postMessage({ buf: pps, offset: 0, length: pps.length });
 
                 for (let sample = 0; sample < video.getSampleCount(); sample++) {
                     video.getSampleNALUnits(sample).forEach((nal) => {
-                        worker.postMessage({ buf: nal, offset: 0, length: nal.length });
+                        this.videoWorker.postMessage({ buf: nal, offset: 0, length: nal.length });
                     });
                 }
             } else {
@@ -368,6 +385,20 @@ export class FrameDecoder {
         }
     }
 
+    public close(): void {
+        if (this.zipWorker) {
+            this.zipWorker.terminate();
+            this.zipWorker = null;
+        }
+
+        if (this.videoWorker) {
+            this.videoWorker.terminate();
+            this.videoWorker = null;
+        }
+
+        this.cleanup(Number.MAX_SAFE_INTEGER);
+    }
+
     public cachedChunks(includeInProgress = false): number[] {
         const chunkIsBeingDecoded = includeInProgress && this.chunkIsBeingDecoded ?
             Math.floor(this.chunkIsBeingDecoded.start / this.chunkSize) : null;

cvat-sdk/gen/generate.sh

@@ -8,7 +8,7 @@ set -e
 
 GENERATOR_VERSION="v6.0.1"
 
-VERSION="2.14.1"
+VERSION="2.14.2"
 LIB_NAME="cvat_sdk"
 LAYER1_LIB_NAME="${LIB_NAME}/api_client"
 DST_DIR="$(cd "$(dirname -- "$0")/.." && pwd)"

cvat-ui/src/actions/annotation-actions.ts

@@ -872,7 +872,11 @@ export function closeJob(): ThunkAction {
     return async (dispatch: ActionCreator<Dispatch>, getState): Promise<void> => {
         const state = getState();
         const { instance: canvasInstance } = state.annotation.canvas;
-        const { jobInstance } = receiveAnnotationsParameters();
+        const { jobInstance, groundTruthInstance } = receiveAnnotationsParameters();
+
+        if (groundTruthInstance) {
+            await groundTruthInstance.close();
+        }
 
         if (jobInstance) {
             await jobInstance.close();

cvat/__init__.py

@@ -4,6 +4,6 @@
 
 from cvat.utils.version import get_version
 
-VERSION = (2, 14, 1, 'final', 0)
+VERSION = (2, 14, 2, 'final', 0)
 
 __version__ = get_version(VERSION)

cvat/apps/engine/utils.py

@@ -165,25 +165,18 @@ def define_dependent_job(
     if not should_be_dependent:
         return None
 
-    started_user_jobs = [
-        job
-        for job in queue.job_class.fetch_many(
-            queue.started_job_registry.get_job_ids(), queue.connection
-        )
-        if job and job.meta.get("user", {}).get("id") == user_id
-    ]
-    deferred_user_jobs = [
-        job
-        for job in queue.job_class.fetch_many(
-            queue.deferred_job_registry.get_job_ids(), queue.connection
-        )
-        # Since there is no cleanup implementation in DeferredJobRegistry,
-        # this registry can contain "outdated" jobs that weren't deleted from it
-        # but were added to another registry. Probably such situations can occur
-        # if there are active or deferred jobs when restarting the worker container.
-        if job and job.meta.get("user", {}).get("id") == user_id and job.is_deferred
-    ]
-    all_user_jobs = started_user_jobs + deferred_user_jobs
+    queues = [queue.deferred_job_registry, queue, queue.started_job_registry]
+    # Since there is no cleanup implementation in DeferredJobRegistry,
+    # this registry can contain "outdated" jobs that weren't deleted from it
+    # but were added to another registry. Probably such situations can occur
+    # if there are active or deferred jobs when restarting the worker container.
+    filters = [lambda job: job.is_deferred, lambda _: True, lambda _: True]
+    all_user_jobs = []
+    for q, f in zip(queues, filters):
+        job_ids = q.get_job_ids()
+        jobs = q.job_class.fetch_many(job_ids, q.connection)
+        jobs = filter(lambda job: job and job.meta.get("user", {}).get("id") == user_id and f(job), jobs)
+        all_user_jobs.extend(jobs)
 
     # prevent possible cyclic dependencies
     if rq_id:

cvat/schema.yml

@@ -1,7 +1,7 @@
 openapi: 3.0.3
 info:
   title: CVAT REST API
-  version: 2.14.1
+  version: 2.14.2
   description: REST API for Computer Vision Annotation Tool (CVAT)
   termsOfService: https://www.google.com/policies/terms/
   contact: