Transactional saving of volume annotations

frontend/javascripts/libs/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
+ * 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/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

@@ -5,7 +5,7 @@ export default class WorkerPool<P, R> {
   //
   // Example:
   // const compressionPool = new WorkerPool(
-  //   () => createWorker(ByteArrayToLz4Base64Worker),
+  //   () => createWorker(ByteArraysToLz4Base64Worker),
   //   COMPRESSION_WORKER_COUNT,
   // );
   // const promise1 = compressionPool.submit(data1);

frontend/javascripts/oxalis/model/bucket_data_handling/data_cube.ts

@@ -679,6 +679,7 @@ class DataCube {
   }
 
   triggerPushQueue() {
+    console.log("triggerPushQueue");
     this.pushQueue.push();
   }
 

frontend/javascripts/oxalis/model/bucket_data_handling/pushqueue.ts

@@ -5,17 +5,18 @@ import { sendToStore } from "oxalis/model/bucket_data_handling/wkstore_adapter";
 import AsyncTaskQueue from "libs/async_task_queue";
 import type DataCube from "oxalis/model/bucket_data_handling/data_cube";
 import Toast from "libs/toast";
+import { createDebouncedAbortableParameterlessCallable } from "libs/debounced_abortable_saga";
+import { call } from "redux-saga/effects";
 export const COMPRESSING_BATCH_SIZE = 32;
-// Only process the PushQueue after there was no user interaction
-// for PUSH_DEBOUNCE_TIME milliseconds...
+// Only process the PushQueue after there was no user interaction (or bucket modification due to
+// downsampling) for PUSH_DEBOUNCE_TIME milliseconds...
 const PUSH_DEBOUNCE_TIME = 1000;
 // ...unless a timeout of PUSH_DEBOUNCE_MAX_WAIT_TIME milliseconds
 // is exceeded. Then, initiate a push.
-const PUSH_DEBOUNCE_MAX_WAIT_TIME = 30000;
+// todo: reactivate?
+const _PUSH_DEBOUNCE_MAX_WAIT_TIME = 30000;
 
 class PushQueue {
-  // @ts-expect-error ts-migrate(2564) FIXME: Property 'dataSetName' has no initializer and is n... Remove this comment to see the full error message
-  dataSetName: string;
   cube: DataCube;
   compressionTaskQueue: AsyncTaskQueue;
   sendData: boolean;
@@ -65,7 +66,6 @@ class PushQueue {
       this.pendingQueue.add(bucket);
       bucket.dirtyCount++;
     }
-
     this.push();
   }
 
@@ -77,41 +77,41 @@ class PushQueue {
     this.pendingQueue.forEach((e) => console.log(e));
   }
 
-  pushImpl = async () => {
-    await this.cube.temporalBucketManager.getAllLoadedPromise();
+  pushImpl = function* (this: PushQueue) {
+    try {
+      console.log("pushImpl start");
+      yield call(this.cube.temporalBucketManager.getAllLoadedPromise);
 
-    if (!this.sendData) {
-      return;
-    }
+      if (!this.sendData) {
+        return;
+      }
 
-    while (this.pendingQueue.size) {
-      let batchSize = Math.min(COMPRESSING_BATCH_SIZE, this.pendingQueue.size);
-      const batch: DataBucket[] = [];
+      console.log("this.pendingQueue.size", this.pendingQueue.size);
 
-      for (const bucket of this.pendingQueue) {
-        if (batchSize <= 0) break;
-        this.pendingQueue.delete(bucket);
-        batch.push(bucket);
-        batchSize--;
-      }
+      // Flush pendingQueue. Note that it's important to do this synchronously.
+      // If other actors could add to queue concurrently, the front-end could
+      // send an inconsistent state for a transaction.
+      const batch: DataBucket[] = Array.from(this.pendingQueue);
+      this.pendingQueue = new Set();
 
       // fire and forget
-      this.compressionTaskQueue.scheduleTask(() => this.pushBatch(batch));
-    }
-
-    try {
-      // wait here
-      await this.compressionTaskQueue.join();
+      this.pushBatch(batch);
     } catch (_error) {
+      // todo: somewhere else?
       alert("We've encountered a permanent issue while saving. Please try to reload the page.");
     }
+    console.log("pushImpl end");
   };
 
-  push = _.debounce(this.pushImpl, PUSH_DEBOUNCE_TIME, {
-    maxWait: PUSH_DEBOUNCE_MAX_WAIT_TIME,
-  });
+  // push = _.debounce(this.pushImpl, PUSH_DEBOUNCE_TIME, {
+  //   maxWait: PUSH_DEBOUNCE_MAX_WAIT_TIME,
+  // });
+
+  // todo: prevent user from brushing for eternity?
+  push = createDebouncedAbortableParameterlessCallable(this.pushImpl, PUSH_DEBOUNCE_TIME, this);
 
   pushBatch(batch: Array<DataBucket>): Promise<void> {
+    // The batch will be put into one transaction.
     return sendToStore(batch, this.cube.layerName);
   }
 }

frontend/javascripts/oxalis/model/bucket_data_handling/temporal_bucket_manager.ts

@@ -51,9 +51,9 @@ class TemporalBucketManager {
     return loadedPromise;
   }
 
-  async getAllLoadedPromise(): Promise<void> {
+  getAllLoadedPromise = async () => {
     await Promise.all(this.loadedPromises);
-  }
+  };
 }
 
 export default TemporalBucketManager;

frontend/javascripts/oxalis/model/bucket_data_handling/wkstore_adapter.ts

@@ -13,7 +13,7 @@ import { parseAsMaybe } from "libs/utils";
 import { pushSaveQueueTransaction } from "oxalis/model/actions/save_actions";
 import type { UpdateAction } from "oxalis/model/sagas/update_actions";
 import { updateBucket } from "oxalis/model/sagas/update_actions";
-import ByteArrayToLz4Base64Worker from "oxalis/workers/byte_array_to_lz4_base64.worker";
+import ByteArraysToLz4Base64Worker from "oxalis/workers/byte_arrays_to_lz4_base64.worker";
 import DecodeFourBitWorker from "oxalis/workers/decode_four_bit.worker";
 import ErrorHandling from "libs/error_handling";
 import Request from "libs/request";
@@ -25,13 +25,20 @@ import constants, { MappingStatusEnum } from "oxalis/constants";
 import window from "libs/window";
 import { getGlobalDataConnectionInfo } from "../data_connection_info";
 import { ResolutionInfo } from "../helpers/resolution_info";
+import _ from "lodash";
 
 const decodeFourBit = createWorker(DecodeFourBitWorker);
+
+// For 64-bit buckets with 32^3 voxel, a COMPRESSION_BATCH_SIZE of
+// 128 corresponds to 16.8 MB that are sent to a webworker in one
+// go.
+const COMPRESSION_BATCH_SIZE = 128;
 const COMPRESSION_WORKER_COUNT = 2;
 const compressionPool = new WorkerPool(
-  () => createWorker(ByteArrayToLz4Base64Worker),
+  () => createWorker(ByteArraysToLz4Base64Worker),
   COMPRESSION_WORKER_COUNT,
 );
+
 export const REQUEST_TIMEOUT = 60000;
 export type SendBucketInfo = {
   position: Vector3;
@@ -239,14 +246,25 @@ function sliceBufferIntoPieces(
 }
 
 export async function sendToStore(batch: Array<DataBucket>, tracingId: string): Promise<void> {
-  const items: Array<UpdateAction> = await Promise.all(
-    batch.map(async (bucket): Promise<UpdateAction> => {
-      const data = bucket.getCopyOfData();
-      const bucketInfo = createSendBucketInfo(bucket.zoomedAddress, bucket.cube.resolutionInfo);
-      const byteArray = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
-      const compressedBase64 = await compressionPool.submit(byteArray);
-      return updateBucket(bucketInfo, compressedBase64);
-    }),
+  const items: Array<UpdateAction> = _.flatten(
+    await Promise.all(
+      _.chunk(batch, COMPRESSION_BATCH_SIZE).map(async (batchSubset): Promise<UpdateAction[]> => {
+        const byteArrays = [];
+        for (const bucket of batchSubset) {
+          const data = bucket.getCopyOfData();
+          const byteArray = new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
+          byteArrays.push(byteArray);
+        }
+
+        const compressedBase64Strings = await compressionPool.submit(byteArrays);
+        return compressedBase64Strings.map((compressedBase64, index) => {
+          const bucket = batchSubset[index];
+          const bucketInfo = createSendBucketInfo(bucket.zoomedAddress, bucket.cube.resolutionInfo);
+          return updateBucket(bucketInfo, compressedBase64);
+        });
+      }),
+    ),
   );
+
   Store.dispatch(pushSaveQueueTransaction(items, "volume", tracingId));
 }

frontend/javascripts/oxalis/model/sagas/save_saga.ts

@@ -154,6 +154,7 @@ let didShowFailedSimultaneousTracingError = false;
 export function* sendRequestToServer(saveQueueType: SaveQueueType, tracingId: string): Saga<void> {
   const fullSaveQueue = yield* select((state) => selectQueue(state, saveQueueType, tracingId));
   const saveQueue = sliceAppropriateBatchCount(fullSaveQueue);
+  console.log("saving", saveQueue.length, "items out of", fullSaveQueue.length);
   let compactedSaveQueue = compactSaveQueue(saveQueue);
   const { version, type } = yield* select((state) =>
     selectTracing(state, saveQueueType, tracingId),
@@ -162,6 +163,7 @@ export function* sendRequestToServer(saveQueueType: SaveQueueType, tracingId: st
   compactedSaveQueue = addVersionNumbers(compactedSaveQueue, version);
   let retryCount = 0;
 
+  // This while-loop only exists for the purpose of a retry-mechanism
   while (true) {
     let exceptionDuringMarkBucketsAsNotDirty = false;
 

frontend/javascripts/oxalis/model/sagas/save_saga_constants.ts

@@ -11,5 +11,5 @@ export const UNDO_HISTORY_SIZE = 20;
 export const SETTINGS_RETRY_DELAY = 15 * 1000;
 export const SETTINGS_MAX_RETRY_COUNT = 20; // 20 * 15s == 5m
 
-export const maximumActionCountPerBatch = 5000;
-export const maximumActionCountPerSave = 15000;
+export const maximumActionCountPerBatch = 20;
+export const maximumActionCountPerSave = 60;

frontend/javascripts/oxalis/model/volumetracing/volume_annotation_sampling.ts

@@ -80,7 +80,7 @@ function upsampleVoxelMap(
         ]);
 
         if (currentGoalBucket.type === "null") {
-          console.warn(warnAboutCouldNotCreate([...currentGoalBucketAddress, targetZoomStep]));
+          warnAboutCouldNotCreate([...currentGoalBucketAddress, targetZoomStep]);
           continue;
         }