Transactional saving of volume annotations

frontend/javascripts/libs/async/async_fifo_resolver.ts

@@ -0,0 +1,34 @@
+/*
+ * This class can be used to await promises
+ * in the order they were passed to
+ * orderedWaitFor.
+ *
+ * This enables scheduling of asynchronous work
+ * concurrently while ensuring that the results
+ * are processed in the order they were requested
+ * (instead of the order in which they finished).
+ *
+ * Example:
+ * const resolver = new AsyncFifoResolver();
+ * const promise1Done = resolver.orderedWaitFor(promise1);
+ * const promise2Done = resolver.orderedWaitFor(promise2);
+ *
+ * Even if promise2 resolves before promise1, promise2Done
+ * will resolve *after* promise1Done.
+ */
+
+export class AsyncFifoResolver<T> {
+  queue: Promise<T>[];
+  constructor() {
+    this.queue = [];
+  }
+
+  async orderedWaitFor(promise: Promise<T>): Promise<T> {
+    this.queue.push(promise);
+    const promiseCount = this.queue.length;
+    const retVals = await Promise.all(this.queue);
+    // Trim awaited promises
+    this.queue = this.queue.slice(promiseCount);
+    return retVals[promiseCount - 1];
+  }
+}

frontend/javascripts/libs/async/debounced_abortable_saga.ts

@@ -0,0 +1,95 @@
+import { call, type Saga } from "oxalis/model/sagas/effect-generators";
+import { buffers, Channel, channel, runSaga } from "redux-saga";
+import { delay, race, take } from "redux-saga/effects";
+
+/*
+ * This function takes a saga and a debounce threshold
+ * and returns a function F that will trigger the given saga
+ * in a debounced manner.
+ * In contrast to a normal debouncing mechanism, the saga
+ * will be cancelled if F is called while the saga is running.
+ * Note that this means that concurrent executions of the saga
+ * are impossible that way (by design).
+ *
+ * Also note that the performance of this debouncing mechanism
+ * is slower than a standard _.debounce. Also see
+ * debounced_abortable_saga.spec.ts for a small benchmark.
+ */
+export function createDebouncedAbortableCallable<T, C>(
+  fn: (param1: T) => Saga<void>,
+  debounceThreshold: number,
+  context: C,
+) {
+  // The communication with the saga is done via a channel.
+  // That way, we can expose a normal function that
+  // does the triggering by filling the channel.
+
+  // Only the most recent invocation should survive.
+  // Therefore, create a sliding buffer with size 1.
+  const buffer = buffers.sliding<T>(1);
+  const triggerChannel: Channel<T> = channel<T>(buffer);
+
+  const _task = runSaga(
+    {},
+    debouncedAbortableSagaRunner,
+    debounceThreshold,
+    triggerChannel,
+    // @ts-expect-error TS thinks fn doesnt match, but it does.
+    fn,
+    context,
+  );
+
+  return (msg: T) => {
+    triggerChannel.put(msg);
+  };
+}
+
+export function createDebouncedAbortableParameterlessCallable<C>(
+  fn: () => Saga<void>,
+  debounceThreshold: number,
+  context: C,
+) {
+  const wrappedFn = createDebouncedAbortableCallable(fn, debounceThreshold, context);
+  const dummyParameter = {};
+  return () => {
+    wrappedFn(dummyParameter);
+  };
+}
+
+function* debouncedAbortableSagaRunner<T, C>(
+  debounceThreshold: number,
+  triggerChannel: Channel<T>,
+  abortableFn: (param: T) => Saga<void>,
+  context: C,
+): Saga<void> {
+  while (true) {
+    // Wait for a trigger-call by consuming
+    // the channel.
+    let msg = yield take(triggerChannel);
+
+    // Repeatedly try to execute abortableFn (each try
+    // might be cancelled due to new initiation-requests)
+    while (true) {
+      const { debounced, latestMessage } = yield race({
+        debounced: delay(debounceThreshold),
+        latestMessage: take(triggerChannel),
+      });
+
+      if (latestMessage) {
+        msg = latestMessage;
+      }
+
+      if (debounced) {
+        const { abortingMessage } = yield race({
+          finished: call([context, abortableFn], msg),
+          abortingMessage: take(triggerChannel),
+        });
+        if (abortingMessage) {
+          msg = abortingMessage;
+        } else {
+          break;
+        }
+      }
+    }
+  }
+}

frontend/javascripts/libs/latest_task_executor.ts → frontend/javascripts/libs/async/latest_task_executor.ts

@@ -1,4 +1,4 @@
-import Deferred from "libs/deferred";
+import Deferred from "libs/async/deferred";
 type Task<T> = () => Promise<T>;
 export const SKIPPED_TASK_REASON = "Skipped task";
 /*
@@ -11,6 +11,10 @@ export const SKIPPED_TASK_REASON = "Skipped task";
  * LatestTaskExecutor instance.
  *
  * See the corresponding spec for examples.
+ *
+ * If you need the same behavior plus cancellation of running
+ * tasks, take a look at the saga-based `createDebouncedAbortableCallable`
+ * utility.
  */
 
 export default class LatestTaskExecutor<T> {

frontend/javascripts/libs/task_pool.ts → frontend/javascripts/libs/async/task_pool.ts

@@ -1,5 +1,5 @@
-import type { Saga, Task } from "oxalis/model/sagas/effect-generators";
-import { join, call, fork } from "typed-redux-saga";
+import type { Saga } from "oxalis/model/sagas/effect-generators";
+import { join, call, fork, FixedTask } from "typed-redux-saga";
 
 /*
   Given an array of async tasks, processTaskWithPool
@@ -10,12 +10,11 @@ export default function* processTaskWithPool(
   tasks: Array<() => Saga<void>>,
   poolSize: number,
 ): Saga<void> {
-  const startedTasks: Array<Task<void>> = [];
+  const startedTasks: Array<FixedTask<void>> = [];
   let isFinalResolveScheduled = false;
   let error: Error | null = null;
 
-  // @ts-expect-error ts-migrate(7006) FIXME: Parameter 'fn' implicitly has an 'any' type.
-  function* forkSafely(fn): Saga<void> {
+  function* forkSafely(fn: () => Saga<void>): Saga<void> {
     // Errors from forked tasks cannot be caught, see https://redux-saga.js.org/docs/advanced/ForkModel/#error-propagation
     // However, the task pool should not abort if a single task fails.
     // Therefore, use this wrapper to safely execute all tasks and possibly rethrow the last error in the end.
@@ -32,17 +31,15 @@ export default function* processTaskWithPool(
         isFinalResolveScheduled = true;
         // All tasks were kicked off, which is why all tasks can be
         // awaited now together.
-        // @ts-expect-error ts-migrate(2769) FIXME: No overload matches this call.
         yield* join(startedTasks);
         if (error != null) throw error;
       }
 
       return;
     }
 
-    const task = tasks.shift();
+    const task = tasks.shift() as () => Saga<void>;
     const newTask = yield* fork(forkSafely, task);
-    // @ts-expect-error ts-migrate(2345) FIXME: Argument of type 'FixedTask<void>' is not assignab... Remove this comment to see the full error message
     startedTasks.push(newTask);
     // If that task is done, process a new one (that way,
     // the pool size stays constant until the queue is almost empty.)

frontend/javascripts/libs/worker_pool.ts → frontend/javascripts/libs/webworker_pool.ts

@@ -1,11 +1,11 @@
 import _ from "lodash";
-export default class WorkerPool<P, R> {
+export default class WebWorkerPool<P, R> {
   // This class can be used to instantiate multiple web workers
   // which are then used for computation in a simple round-robin manner.
   //
   // Example:
-  // const compressionPool = new WorkerPool(
-  //   () => createWorker(ByteArrayToLz4Base64Worker),
+  // const compressionPool = new WebWorkerPool(
+  //   () => createWorker(ByteArraysToLz4Base64Worker),
   //   COMPRESSION_WORKER_COUNT,
   // );
   // const promise1 = compressionPool.submit(data1);

frontend/javascripts/oxalis/model.ts

@@ -1,5 +1,4 @@
 import _ from "lodash";
-import { COMPRESSING_BATCH_SIZE } from "oxalis/model/bucket_data_handling/pushqueue";
 import type { Vector3 } from "oxalis/constants";
 import type { Versions } from "oxalis/view/version_view";
 import { getActiveSegmentationTracingLayer } from "oxalis/model/accessors/volumetracing_accessor";
@@ -286,16 +285,23 @@ export class OxalisModel {
     return storeStateSaved && pushQueuesSaved;
   }
 
+  getLongestPushQueueWaitTime() {
+    return (
+      _.max(
+        Utils.values(this.dataLayers).map((layer) => layer.pushQueue.getTransactionWaitTime()),
+      ) || 0
+    );
+  }
+
   getPushQueueStats() {
     const compressingBucketCount = _.sum(
-      Utils.values(this.dataLayers).map(
-        (dataLayer) =>
-          dataLayer.pushQueue.compressionTaskQueue.tasks.length * COMPRESSING_BATCH_SIZE,
+      Utils.values(this.dataLayers).map((dataLayer) =>
+        dataLayer.pushQueue.getCompressingBucketCount(),
       ),
     );
 
     const waitingForCompressionBucketCount = _.sum(
-      Utils.values(this.dataLayers).map((dataLayer) => dataLayer.pushQueue.pendingQueue.size),
+      Utils.values(this.dataLayers).map((dataLayer) => dataLayer.pushQueue.getPendingQueueSize()),
     );
 
     return {