add lock

Author: jochem-brouwer

src/baseTrie.ts

@@ -255,7 +255,7 @@ export class Trie {
               const childRef = child[1] as Buffer
               const childKey = key.concat(keyExtension)
               const priority = childKey.length
-              taskExecutor.execute(priority, async (taskCallback: Function) => {
+              await taskExecutor.execute(priority, async (taskCallback: Function) => {
                 const childNode = await self._lookupNode(childRef)
                 taskCallback()
                 if (childNode) {
@@ -275,7 +275,7 @@ export class Trie {
             const childKey = key.slice()
             childKey.push(childIndex)
             const priority = childKey.length
-            taskExecutor.execute(priority, async (taskCallback: Function) => {
+            await taskExecutor.execute(priority, async (taskCallback: Function) => {
               const childNode = await self._lookupNode(childRef)
               taskCallback()
               if (childNode) {

src/prioritizedTaskExecutor.ts

@@ -1,3 +1,5 @@
+import Semaphore from 'semaphore-async-await'
+
 interface Task {
   priority: number
   fn: Function
@@ -10,6 +12,8 @@ export class PrioritizedTaskExecutor {
   private currentPoolSize: number
   /** The task queue */
   private queue: Task[]
+  /** The Lock */
+  private lock: Semaphore
 
   /**
    * Executes tasks up to maxPoolSize at a time, other items are put in a priority queue.
@@ -21,6 +25,7 @@ export class PrioritizedTaskExecutor {
     this.maxPoolSize = maxPoolSize
     this.currentPoolSize = 0
     this.queue = []
+    this.lock = new Semaphore(1)
   }
 
   /**
@@ -29,15 +34,18 @@ export class PrioritizedTaskExecutor {
    * @param priority The priority of the task
    * @param fn The function that accepts the callback, which must be called upon the task completion.
    */
-  execute(priority: number, fn: Function) {
+  async execute(priority: number, fn: Function) {
+    await this.lock.acquire()
     if (this.currentPoolSize < this.maxPoolSize) {
       this.currentPoolSize++
-      fn(() => {
+      fn(async () => {
+        await this.lock.acquire()
         this.currentPoolSize--
         if (this.queue.length > 0) {
           const item = this.queue.shift()
           this.execute(item!.priority, item!.fn)
         }
+        this.lock.signal()
       })
     } else {
       if (this.queue.length == 0) {
@@ -70,6 +78,7 @@ export class PrioritizedTaskExecutor {
         }
       }
     }
+    this.lock.signal()
   }
 
   /**