Enabled timeouts for lock acquisition

Author: CMCDragonkai

package-lock.json

@@ -20,7 +20,8 @@
   },
   "dependencies": {
     "@matrixai/resources": "^1.0.0",
-    "async-mutex": "^0.3.2"
+    "async-mutex": "^0.3.2",
+    "ts-custom-error": "^3.2.0"
   },
   "devDependencies": {
     "@types/jest": "^27.0.2",

package.json

@@ -1,45 +1,64 @@
 import type { MutexInterface } from 'async-mutex';
 import type { ResourceAcquire } from '@matrixai/resources';
-import { Mutex } from 'async-mutex';
+import { Mutex, withTimeout } from 'async-mutex';
 import { withF, withG } from '@matrixai/resources';
+import { yieldMicro } from './utils';
+import { ErrorAsyncLocksTimeout } from './errors';
 
 class Lock {
-  protected lock: Mutex = new Mutex();
-  protected release: MutexInterface.Releaser;
+  protected _lock: Mutex = new Mutex();
   protected _count: number = 0;
 
-  public acquire: ResourceAcquire<Lock> = async () => {
-    ++this._count;
-    this.release = await this.lock.acquire();
-    return [
-      async () => {
+  public lock(timeout?: number): ResourceAcquire<Lock> {
+    return async () => {
+      ++this._count;
+      let lock: MutexInterface = this._lock;
+      if (timeout != null) {
+        lock = withTimeout(this._lock, timeout, new ErrorAsyncLocksTimeout());
+      }
+      let release: MutexInterface.Releaser;
+      try {
+        release = await lock.acquire();
+      } catch (e) {
         --this._count;
-        this.release();
-      },
-      this,
-    ];
-  };
+        throw e;
+      }
+      return [
+        async () => {
+          --this._count;
+          release();
+          // Allow semaphore to settle https://github.com/DirtyHairy/async-mutex/issues/54
+          await yieldMicro();
+        },
+        this,
+      ];
+    };
+  }
 
   public get count(): number {
     return this._count;
   }
 
   public isLocked(): boolean {
-    return this.lock.isLocked();
+    return this._lock.isLocked();
   }
 
   public async waitForUnlock(): Promise<void> {
-    return this.lock.waitForUnlock();
+    return this._lock.waitForUnlock();
   }
 
-  public async withF<T>(f: (resources: [Lock]) => Promise<T>): Promise<T> {
-    return withF([this.acquire], f);
+  public async withF<T>(
+    f: (resources: [Lock]) => Promise<T>,
+    timeout?: number,
+  ): Promise<T> {
+    return withF([this.lock(timeout)], f);
   }
 
   public withG<T, TReturn, TNext>(
     g: (resources: [Lock]) => AsyncGenerator<T, TReturn, TNext>,
+    timeout?: number,
   ): AsyncGenerator<T, TReturn, TNext> {
-    return withG([this.acquire], g);
+    return withG([this.lock(timeout)], g);
   }
 }
 

src/Lock.ts

@@ -1,7 +1,9 @@
 import type { MutexInterface } from 'async-mutex';
 import type { ResourceAcquire } from '@matrixai/resources';
-import { Mutex } from 'async-mutex';
+import { Mutex, withTimeout } from 'async-mutex';
 import { withF, withG } from '@matrixai/resources';
+import { yieldMicro } from './utils';
+import { ErrorAsyncLocksTimeout } from './errors';
 
 /**
  * Read-preferring read write lock
@@ -12,35 +14,62 @@ class RWLockReader {
   protected lock: Mutex = new Mutex();
   protected release: MutexInterface.Releaser;
 
-  public acquireRead: ResourceAcquire<RWLockReader> = async () => {
-    const readerCount = ++this._readerCount;
-    // The first reader locks
-    if (readerCount === 1) {
-      this.release = await this.lock.acquire();
-    }
-    return [
-      async () => {
-        const readerCount = --this._readerCount;
-        // The last reader unlocks
-        if (readerCount === 0) {
-          this.release();
+  public read(timeout?: number): ResourceAcquire<RWLockReader> {
+    return async () => {
+      const readerCount = ++this._readerCount;
+      // The first reader locks
+      if (readerCount === 1) {
+        let lock: MutexInterface = this.lock;
+        if (timeout != null) {
+          lock = withTimeout(this.lock, timeout, new ErrorAsyncLocksTimeout());
         }
-      },
-      this,
-    ];
-  };
+        try {
+          this.release = await lock.acquire();
+        } catch (e) {
+          --this._readerCount;
+          throw e;
+        }
+      }
+      return [
+        async () => {
+          const readerCount = --this._readerCount;
+          // The last reader unlocks
+          if (readerCount === 0) {
+            this.release();
+          }
+          // Allow semaphore to settle https://github.com/DirtyHairy/async-mutex/issues/54
+          await yieldMicro();
+        },
+        this,
+      ];
+    };
+  }
 
-  public acquireWrite: ResourceAcquire<RWLockReader> = async () => {
-    ++this._writerCount;
-    this.release = await this.lock.acquire();
-    return [
-      async () => {
+  public write(timeout?: number): ResourceAcquire<RWLockReader> {
+    return async () => {
+      ++this._writerCount;
+      let lock: MutexInterface = this.lock;
+      if (timeout != null) {
+        lock = withTimeout(this.lock, timeout, new ErrorAsyncLocksTimeout());
+      }
+      let release: MutexInterface.Releaser;
+      try {
+        release = await lock.acquire();
+      } catch (e) {
         --this._writerCount;
-        this.release();
-      },
-      this,
-    ];
-  };
+        throw e;
+      }
+      return [
+        async () => {
+          release();
+          --this._writerCount;
+          // Allow semaphore to settle https://github.com/DirtyHairy/async-mutex/issues/54
+          await yieldMicro();
+        },
+        this,
+      ];
+    };
+  }
 
   public get readerCount(): number {
     return this._readerCount;
@@ -60,26 +89,30 @@ class RWLockReader {
 
   public async withReadF<T>(
     f: (resources: [RWLockReader]) => Promise<T>,
+    timeout?: number,
   ): Promise<T> {
-    return withF([this.acquireRead], f);
+    return withF([this.read(timeout)], f);
   }
 
   public async withWriteF<T>(
     f: (resources: [RWLockReader]) => Promise<T>,
+    timeout?: number,
   ): Promise<T> {
-    return withF([this.acquireWrite], f);
+    return withF([this.write(timeout)], f);
   }
 
   public withReadG<T, TReturn, TNext>(
     g: (resources: [RWLockReader]) => AsyncGenerator<T, TReturn, TNext>,
+    timeout?: number,
   ): AsyncGenerator<T, TReturn, TNext> {
-    return withG([this.acquireRead], g);
+    return withG([this.read(timeout)], g);
   }
 
   public withWriteG<T, TReturn, TNext>(
     g: (resources: [RWLockReader]) => AsyncGenerator<T, TReturn, TNext>,
+    timeout?: number,
   ): AsyncGenerator<T, TReturn, TNext> {
-    return withG([this.acquireWrite], g);
+    return withG([this.write(timeout)], g);
   }
 }