Merge pull request #393 from scratchfoundation/task-queue-manager

Add QueueManager & improve TaskQueue

Author: cwillisf

package-lock.json

@@ -91,6 +91,11 @@ const baseConfig = new ScratchWebpackConfigBuilder(
                 from: 'chunks/fetch-worker.*.{js,js.map}',
                 noErrorOnMissing: true
             },
+            {
+                context: '../../node_modules/scratch-storage/dist/web',
+                from: 'chunks/vendors-*.{js,js.map}',
+                noErrorOnMissing: true
+            },
             {
                 from: '../../node_modules/@mediapipe/face_detection',
                 to: 'chunks/mediapipe/face_detection'

packages/scratch-gui/webpack.config.js

@@ -69,9 +69,10 @@ class BitmapSkin extends Skin {
         // memory.
         let textureData = bitmapData;
         if (bitmapData instanceof HTMLCanvasElement) {
-            // Given a HTMLCanvasElement get the image data to pass to webgl and
-            // Silhouette.
-            const context = bitmapData.getContext('2d');
+            // Given a HTMLCanvasElement, get the image data to pass to WebGL and the Silhouette. The 2D context was
+            // likely already created, so `willReadFrequently` is likely ignored here, but it's good documentation and
+            // may help if the code path changes someday.
+            const context = bitmapData.getContext('2d', {willReadFrequently: true});
             textureData = context.getImageData(0, 0, bitmapData.width, bitmapData.height);
         }
 

packages/scratch-render/src/BitmapSkin.js

@@ -145,7 +145,7 @@ const fetchBitmapCanvas_ = function (costume, runtime, rotationCenter) {
             mergeCanvas.width = baseImageElement.width;
             mergeCanvas.height = baseImageElement.height;
 
-            const ctx = mergeCanvas.getContext('2d');
+            const ctx = mergeCanvas.getContext('2d', {willReadFrequently: true});
             ctx.drawImage(baseImageElement, 0, 0);
             if (textImageElement) {
                 ctx.drawImage(textImageElement, 0, 0);

packages/scratch-vm/src/import/load-costume.js

@@ -48,8 +48,5 @@
   },
   "overrides": {
     "vite": "npm:[email protected]"
-  },
-  "dependencies": {
-    "p-limit": "7.2.0"
   }
 }

packages/task-herder/package.json

@@ -0,0 +1,63 @@
+import { TaskQueue, type QueueOptions } from './TaskQueue'
+
+/**
+ * Manages multiple named task queues with shared or individual configurations. Each queue can be accessed by a
+ * unique name or other identifier. For example, this can manage a separate queue for each of several different
+ * servers, using the server's DNS name as the key.
+ */
+export class QueueManager<T = string> {
+  private readonly queues: Map<T, TaskQueue>
+  private readonly defaultOptions: QueueOptions
+
+  /**
+   * @param defaultOptions The options that will be used to initialize each queue.
+   * These can be overridden later on a per-queue basis.
+   * @param iterable An optional iterable of key-value pairs to initialize the manager with existing queues.
+   */
+  constructor(defaultOptions: QueueOptions, iterable?: Iterable<readonly [T, TaskQueue]> | null) {
+    this.queues = new Map(iterable)
+    this.defaultOptions = defaultOptions
+  }
+
+  /**
+   * Create a new task queue with the given identifier. If a queue with that identifier already exists, it will be
+   * replaced. If you need to cancel tasks in that queue before replacing it, do so manually first.
+   * @param id The identifier for the queue.
+   * @param overrides Optional overrides for the default QueueOptions for this specific queue.
+   * @returns The newly created TaskQueue.
+   */
+  public create(id: T, overrides: Partial<QueueOptions> = {}): TaskQueue {
+    const queue = new TaskQueue({ ...this.defaultOptions, ...overrides })
+    this.queues.set(id, queue)
+    return queue
+  }
+
+  /**
+   * Get the task queue for the given identifier.
+   * @param id The identifier for the queue.
+   * @returns The TaskQueue associated with the given identifier, or undefined if none exists.
+   */
+  public get(id: T): TaskQueue | undefined {
+    return this.queues.get(id)
+  }
+
+  /**
+   * Get the task queue for the given identifier, creating it if it does not already exist.
+   * @param id The identifier for the queue.
+   * @param overrides Optional overrides for the default QueueOptions for this specific queue. Only used if the queue
+   * did not already exist.
+   * @returns The TaskQueue associated with the given identifier.
+   */
+  public getOrCreate(id: T, overrides: Partial<QueueOptions> = {}): TaskQueue {
+    return this.get(id) ?? this.create(id, overrides)
+  }
+
+  /**
+   * @returns A copy of the default queue options. Used primarily for testing and inspection.
+   */
+  public options(): Readonly<QueueOptions> {
+    return {
+      ...this.defaultOptions,
+    }
+  }
+}

packages/task-herder/src/QueueManager.ts

@@ -1,5 +1,5 @@
-import pLimit, { type LimitFunction } from 'p-limit'
 import { CancelReason } from './CancelReason'
+import { PromiseWithResolvers } from './PromiseWithResolvers'
 import { TaskRecord, type TaskOptions } from './TaskRecord'
 
 export interface QueueOptions {
@@ -29,27 +29,44 @@ export class TaskQueue {
   private readonly burstLimit: number
   private readonly sustainRate: number
   private readonly queueCostLimit: number
-  private readonly concurrencyLimiter: LimitFunction
-  private readonly boundRunTasks = this.runTasks.bind(this)
+  private readonly concurrencyLimit: number
   private tokenCount: number
 
+  private runningTasks = 0
   private pendingTaskRecords: TaskRecord<unknown>[] = []
-  private timeout: number | null = null
   private lastRefillTime: number = Date.now()
+  private onTaskAdded = PromiseWithResolvers<void>().resolve // start with a no-op of correct type
+  private onTaskFinished = PromiseWithResolvers<void>().resolve // start with a no-op of correct type
 
   constructor(options: QueueOptions) {
     this.burstLimit = options.burstLimit
     this.sustainRate = options.sustainRate
     this.tokenCount = options.startingTokens ?? options.burstLimit
     this.queueCostLimit = options.queueCostLimit ?? Infinity
-    this.concurrencyLimiter = pLimit(options.concurrency ?? 1)
+    this.concurrencyLimit = options.concurrency ?? 1
+    void this.runTasks()
   }
 
   /** @returns The number of tasks currently in the queue */
   get length(): number {
     return this.pendingTaskRecords.length
   }
 
+  /**
+   * @returns The current configuration options of the queue. Used primarily for testing and inspection.
+   * Note that the `startingTokens` value returned here reflects the current token count, which is only guaranteed to
+   * match the originally configured starting tokens value if no time has passed and no tasks have been processed.
+   */
+  get options(): Readonly<QueueOptions> {
+    return {
+      burstLimit: this.burstLimit,
+      sustainRate: this.sustainRate,
+      startingTokens: this.tokenCount,
+      queueCostLimit: this.queueCostLimit,
+      concurrency: this.concurrencyLimit,
+    }
+  }
+
   /**
    * Adds a task to the queue. The task will first wait until enough tokens are available, then will wait its turn in
    * the concurrency queue.
@@ -77,10 +94,7 @@ export class TaskQueue {
       this.cancel(taskRecord.promise, new Error(CancelReason.Aborted))
     })
 
-    // If the queue was empty, we need to prime the pump
-    if (this.pendingTaskRecords.length === 1) {
-      void this.runTasks()
-    }
+    this.onTaskAdded()
 
     return taskRecord.promise
   }
@@ -96,9 +110,6 @@ export class TaskQueue {
     if (taskIndex !== -1) {
       const [taskRecord] = this.pendingTaskRecords.splice(taskIndex, 1)
       taskRecord.cancel(reason ?? new Error(CancelReason.Cancel))
-      if (taskIndex === 0 && this.pendingTaskRecords.length > 0) {
-        void this.runTasks()
-      }
       return true
     }
     return false
@@ -110,10 +121,6 @@ export class TaskQueue {
    * @returns The number of tasks that were cancelled.
    */
   cancelAll(reason?: Error): number {
-    if (this.timeout !== null) {
-      clearTimeout(this.timeout)
-      this.timeout = null
-    }
     const oldTasks = this.pendingTaskRecords
     this.pendingTaskRecords = []
     reason = reason ?? new Error(CancelReason.Cancel)
@@ -164,17 +171,15 @@ export class TaskQueue {
   /**
    * Run tasks from the queue as tokens become available.
    */
-  private runTasks(): void {
-    if (this.timeout !== null) {
-      clearTimeout(this.timeout)
-      this.timeout = null
-    }
-
+  private async runTasks(): Promise<void> {
     for (;;) {
       const nextRecord = this.pendingTaskRecords.shift()
       if (!nextRecord) {
         // No more tasks to run
-        return
+        const { promise, resolve } = PromiseWithResolvers<void>()
+        this.onTaskAdded = resolve
+        await promise // wait until a task is added
+        continue // then try again
       }
 
       if (nextRecord.cost > this.burstLimit) {
@@ -185,16 +190,34 @@ export class TaskQueue {
 
       // Refill before each task in case the time it took for the last task to run was enough to afford the next.
       if (this.refillAndSpend(nextRecord.cost)) {
-        // Run the task within the concurrency limiter
-        void this.concurrencyLimiter(nextRecord.run)
+        if (this.runningTasks >= this.concurrencyLimit) {
+          const { promise, resolve } = PromiseWithResolvers<void>()
+          this.onTaskFinished = resolve
+          await promise // wait until a task finishes
+          // then we know there's room for at least one more task
+        }
+        void this.runTask(nextRecord)
       } else {
         // We can't currently afford this task. Put it back and wait until we can, then try again.
         this.pendingTaskRecords.unshift(nextRecord)
         const tokensNeeded = Math.max(nextRecord.cost - this.tokenCount, 0)
         const estimatedWait = Math.ceil((1000 * tokensNeeded) / this.sustainRate)
-        this.timeout = setTimeout(this.boundRunTasks, estimatedWait)
-        return
+        await new Promise(resolve => setTimeout(resolve, estimatedWait))
       }
     }
   }
+
+  /**
+   * Run a task record right now, managing the running tasks count.
+   * @param taskRecord The task that should run.
+   */
+  private async runTask(taskRecord: TaskRecord<unknown>): Promise<void> {
+    this.runningTasks++
+    try {
+      await taskRecord.run()
+    } finally {
+      this.runningTasks--
+      this.onTaskFinished()
+    }
+  }
 }

packages/task-herder/src/TaskQueue.ts

@@ -1,2 +1,3 @@
 export { CancelReason } from './CancelReason'
 export { type QueueOptions, TaskQueue } from './TaskQueue'
+export { QueueManager } from './QueueManager'