feat: promise queue implementation

Author: r4mmer

babel.config.js

@@ -15,5 +15,6 @@ module.exports = {
   plugins: [
     "@babel/plugin-transform-async-generator-functions",
     "@babel/plugin-transform-class-properties",
+    "@babel/plugin-transform-private-methods",
   ],
 };

package-lock.json

@@ -24,7 +24,6 @@
     "level": "8.0.1",
     "lodash": "4.17.21",
     "long": "5.2.3",
-    "queue-promise": "^2.2.1",
     "ws": "8.17.1"
   },
   "scripts": {

package.json

@@ -0,0 +1,179 @@
+/* eslint class-methods-use-this: ["error", { "exceptMethods": ["parent", "left", "right", "comparator"] }] */
+
+interface PriorityQueueNode<T> {
+  value: T;
+  priority: number;
+}
+
+/**
+ * Priority queue implementation using an underlying heap for performance.
+ * @example
+ * const prioQ = new PriorityQueue<string>();
+ * prioQ.push(prioQ.makeNode(0, 'lower prio'));
+ * prioQ.push(prioQ.makeNode(10, 'higher prio'));
+ * prioQ.push(prioQ.makeNode(5, 'medium prio'));
+ * // Returns 'higher prio'
+ * prioQ.pop()
+ * // Returns 'medium prio'
+ * prioQ.pop()
+ * // Returns 'lower prio'
+ * prioQ.pop()
+ *
+ * @template [T=unknown]
+ */
+export default class PriorityQueue<T = unknown> {
+  #heap: PriorityQueueNode<T>[];
+
+  readonly #top: number;
+
+  constructor() {
+    this.#heap = [];
+    this.#top = 0;
+  }
+
+  /**
+   * Utility to create a node from a value and priority.
+   * @example
+   * // returns { value: 'foobar', 'priority': 10 }
+   * prioQueue.makeNode(10, 'foobar');
+   */
+  static makeNode<N>(priority: number, value: N): PriorityQueueNode<N> {
+    return { value, priority };
+  }
+
+  /**
+   * Get the number of elements on the priority queue.
+   */
+  public get size() {
+    return this.#heap.length;
+  }
+
+  /**
+   * Check if the priority queue is empty
+   */
+  public isEmpty() {
+    return this.size === 0;
+  }
+
+  /**
+   * Get the top value of the queue without removing it from the queue.
+   */
+  public peek() {
+    if (this.size === 0) {
+      // Heap is empty
+      return undefined;
+    }
+    return this.#heap[this.#top].value;
+  }
+
+  /**
+   * Add a node to the priority queue and maintain the priority order.
+   */
+  public push(value: PriorityQueueNode<T>) {
+    this.#heap.push(value);
+    this._siftUp();
+  }
+
+  /**
+   * Add multiple values to the priority queue while maintaining the priority order.
+   */
+  public add(...values: PriorityQueueNode<T>[]) {
+    values.forEach(value => {
+      this.push(value);
+    });
+    return this.size;
+  }
+
+  /**
+   * Get the node with highest priority and remove it from the priority queue.
+   */
+  public pop() {
+    const poppedValue = this.peek();
+    // Check if poppedValue is null or undefined, meaning the queue is empty
+    if (!(poppedValue ?? false)) {
+      return undefined;
+    }
+    const bottom = this.size - 1;
+    if (bottom > this.#top) {
+      this._swap(this.#top, bottom);
+    }
+    this.#heap.pop();
+    this._siftDown();
+    return poppedValue;
+  }
+
+  /** Compare 2 nodes and return true if the left one has higher priority. */
+  private comparator(a: PriorityQueueNode<T>, b: PriorityQueueNode<T>): boolean {
+    return a.priority > b.priority;
+  }
+
+  /** Given a node index on the heap get the parent index */
+  private parent(i: number) {
+    return ((i + 1) >>> 1) - 1;
+  }
+
+  /** Given a node index on the heap get the left child index */
+  private left(i: number) {
+    return (i << 1) + 1;
+  }
+
+  /** Given a node index on the heap get the right child index */
+  private right(i: number) {
+    return (i + 1) << 1;
+  }
+
+  /**
+   * Compare the nodes at index `i`, `j` on the heap
+   * Return true if the node at `i` is higher priority than the node at `j`
+   * Return false otherwise.
+   */
+  private _greater(i: number, j: number) {
+    return this.comparator(this.#heap[i], this.#heap[j]);
+  }
+
+  /** swap the nodes at index `i` and `j` on the heap */
+  private _swap(i: number, j: number) {
+    [this.#heap[i], this.#heap[j]] = [this.#heap[j], this.#heap[i]];
+  }
+
+  /**
+   * The last node of the heap will work its way up the heap until it meets a node
+   * of higher priority or reaches the top of the heap.
+   */
+  private _siftUp() {
+    // Start from the last index and work our way up the heap
+    let node = this.size - 1;
+    // While the current node is not at the top and the priority is greater
+    // than the parent we continue sifting up
+    while (node > this.#top && this._greater(node, this.parent(node))) {
+      // parent is lower priority, swap with child (current node)
+      this._swap(node, this.parent(node));
+      // Start from the parent
+      node = this.parent(node);
+    }
+  }
+
+  /**
+   * The top node of the heap will work its way down until no child is of higher
+   * priority or it reaches the bottom of the heap.
+   */
+  private _siftDown() {
+    // Start from the top index and work our way down the heap
+    let node = this.#top;
+    // If a child is in the heap and has higher prioroty, swap with parent and go down
+    while (
+      (this.left(node) < this.size && this._greater(this.left(node), node)) ||
+      (this.right(node) < this.size && this._greater(this.right(node), node))
+    ) {
+      // Get the child with higher priority
+      const maxChild =
+        this.right(node) < this.size && this._greater(this.right(node), this.left(node))
+          ? this.right(node)
+          : this.left(node);
+      // Swap with parent and continue sifting down from the child
+      this._swap(node, maxChild);
+      // Start from the child
+      node = maxChild;
+    }
+  }
+}

src/models/priority_queue.ts

@@ -0,0 +1,151 @@
+import EventEmitter from 'events';
+import PriorityQueue from './priority_queue';
+
+type TaskOptions = { signal?: AbortSignal; };
+
+type Task<TaskResultType> =
+	| ((options?: TaskOptions) => PromiseLike<TaskResultType>)
+	| ((options?: TaskOptions) => TaskResultType);
+
+type AddTaskOptions = {
+  priority?: number;
+  signal?: AbortSignal;
+}
+	
+export default class PromiseQueue extends EventEmitter {
+  #queue: PriorityQueue<() => Promise<void>>;
+
+  // Number of active jobs currently running.
+  #jobsRunning: number;
+
+  #allowedConcurrentJobs: number = 1;
+
+  #isPaused = false;
+
+  #intervalId: ReturnType<typeof setInterval>|null = null;
+
+  constructor() {
+    super();
+    this.#queue = new PriorityQueue<() => Promise<void>>();
+    this.#jobsRunning = 0;
+    this.#startInterval();
+  }
+
+  #startInterval() {
+    this.#clearInterval();
+    this.#intervalId = setInterval(() => {
+      this.processQueue();
+    }, 1000);
+  }
+
+  #clearInterval() {
+    if (this.#intervalId) {
+      clearInterval(this.#intervalId);
+      this.#intervalId = null;
+    }
+  }
+
+  get isPaused() {
+    return this.#isPaused;
+  }
+
+  stop() {
+    this.#isPaused = true;
+    this.#clearInterval();
+  }
+
+  continue() {
+    this.#isPaused = false;
+    this.processQueue();
+    this.#startInterval();
+  }
+
+  #next() {
+    this.emit('next')
+    this.#jobsRunning--;
+    this.#tryToStartNextJob();
+  }
+
+  get #canStartJob() {
+    return this.#jobsRunning < this.concurrent;
+  }
+
+  get concurrent() {
+    return this.#allowedConcurrentJobs;
+  }
+
+  set concurrent(value) {
+    if (value < 1) {
+      throw new Error('Cannot have less than 1 job running.');
+    }
+    this.#allowedConcurrentJobs = value;
+  }
+
+  #tryToStartNextJob(): boolean {
+    if (this.#queue.isEmpty()) {
+      // No more tasks to run
+      this.emit('queue_empty');
+      if (this.#jobsRunning === 0) {
+        this.emit('idle');
+      }
+      return false;
+    }
+
+    if (this.#canStartJob) {
+      const job = this.#queue.pop();
+      if (!job) {
+        // Should never happen, but treating for typing
+        return false;
+      }
+      this.emit('job_start');
+      job();
+      return true;
+    }
+
+    return false;
+  }
+
+	static async throwOnAbort(signal: AbortSignal): Promise<never> {
+		return new Promise((_resolve, reject) => {
+			signal.addEventListener('abort', () => {
+				reject(signal.reason);
+			}, {once: true});
+		});
+	}
+
+	/**
+	 * Try to start jobs until the concurrency limit is reached.
+	 */
+	processQueue(): void {
+		while (this.#tryToStartNextJob()) {}
+	}
+
+  async add<TaskResultType>(task: Task<TaskResultType>, options?: AddTaskOptions) {
+    this.emit('new_job');
+    return new Promise((resolve, reject) => {
+      this.#queue.push(PriorityQueue.makeNode(options?.priority || 0, async () => {
+        this.#jobsRunning++;
+        try {
+          // Throw if the operation was aborted and don't run.
+          options?.signal?.throwIfAborted();
+
+          // Run the task until completion or until the task aborts.
+          let operation = task({signal: options?.signal});
+          // If a signal was passed, we may abort the operation from the outside.
+          // This does not abort internally the task, it has to also manage the abort signal.
+          if (options?.signal) {
+            operation = Promise.race([operation, PromiseQueue.throwOnAbort(options.signal)]);
+          }
+          const result = await operation;
+          resolve(result);
+          // Completed task
+        } catch (error: unknown) {
+          reject(error);
+        } finally {
+          this.emit('finished_job');
+          this.#next();
+        }
+      }));
+    });
+  }
+}

src/models/promise_queue.ts

@@ -49,7 +49,7 @@ import { MemoryStore, Storage } from '../storage';
 import { deriveAddressP2PKH, deriveAddressP2SH, getAddressFromPubkey } from '../utils/address';
 import NanoContractTransactionBuilder from '../nano_contracts/builder';
 import { prepareNanoSendTransaction } from '../nano_contracts/utils';
-import { addTask } from '../sync/gll';
+import GLL from '../sync/gll';
 
 const ERROR_MESSAGE_PIN_REQUIRED = 'Pin is required.';
 
@@ -2859,9 +2859,11 @@ class HathorWallet extends EventEmitter {
       throw new Error('Trying to use an unsupported sync method for this wallet.');
     }
     const syncMethod = getHistorySyncMethod(this.historySyncMode);
-    await addTask(async () => {
+    // This will add the task to the GLL queue and return a promise that
+    // resolves when the task finishes executing
+    await GLL.add(async () => {
       await syncMethod(startIndex, count, this.storage, this.conn, shouldProcessHistory);
-    }, this.logger);
+    });
   }
 
   /**