Updated async-mutex and use write-preferring RWLock for Schema

package.json

@@ -77,7 +77,7 @@
     "@matrixai/logger": "^2.1.0",
     "@matrixai/workers": "^1.2.5",
     "ajv": "^7.0.4",
-    "async-mutex": "^0.2.4",
+    "async-mutex": "^0.3.2",
     "bip39": "^3.0.3",
     "canonicalize": "^1.0.5",
     "cheerio": "^1.0.0-rc.5",

src/schema/Schema.ts

@@ -142,7 +142,7 @@ class Schema {
   }
 
   public async readVersion(): Promise<StateVersion | undefined> {
-    return await this.lock.read(async () => {
+    return await this.lock.withRead(async () => {
       let stateVersionData: string;
       try {
         stateVersionData = await this.fs.promises.readFile(
@@ -169,7 +169,7 @@ class Schema {
   }
 
   protected async writeVersion(stateVersion: StateVersion): Promise<void> {
-    return await this.lock.write(async () => {
+    return await this.lock.withWrite(async () => {
       try {
         await this.fs.promises.writeFile(
           this.stateVersionPath,
@@ -191,7 +191,7 @@ class Schema {
    * This is only called when the version is older.
    */
   protected async upgradeVersion(_stateVersion: StateVersion): Promise<void> {
-    return await this.lock.write(async () => {
+    return await this.lock.withWrite(async () => {
       // TODO: to be implemented
       throw new schemaErrors.ErrorSchemaVersionTooOld();
     });

src/utils/locks.ts

@@ -1,35 +1,84 @@
 import type { MutexInterface } from 'async-mutex';
 import { Mutex } from 'async-mutex';
 
+/**
+ * Single threaded write-preferring read write lock
+ */
 class RWLock {
-  protected readerCount: number = 0;
-  protected lock: Mutex = new Mutex();
-  protected release: MutexInterface.Releaser;
+  protected readersLock: Mutex = new Mutex();
+  protected writersLock: Mutex = new Mutex();
+  protected readersRelease: MutexInterface.Releaser;
+  protected readerCountBlocked: number = 0;
+  protected _readerCount: number = 0;
+  protected _writerCount: number = 0;
 
-  public async read<T>(f: () => Promise<T>): Promise<T> {
-    let readerCount = ++this.readerCount;
-    // The first reader locks
-    if (readerCount === 1) {
-      this.release = await this.lock.acquire();
-    }
+  public get readerCount(): number {
+    return this._readerCount + this.readerCountBlocked;
+  }
+
+  public get writerCount(): number {
+    return this._writerCount;
+  }
+
+  public async withRead<T>(f: () => Promise<T>): Promise<T> {
+    const release = await this.acquireRead();
     try {
       return await f();
     } finally {
-      readerCount = --this.readerCount;
-      // The last reader unlocks
-      if (readerCount === 0) {
-        this.release();
-      }
+      release();
     }
   }
 
-  public async write<T>(f: () => Promise<T>): Promise<T> {
-    this.release = await this.lock.acquire();
+  public async withWrite<T>(f: () => Promise<T>): Promise<T> {
+    const release = await this.acquireWrite();
     try {
       return await f();
     } finally {
-      this.release();
+      release();
+    }
+  }
+
+  public async acquireRead(): Promise<() => void> {
+    if (this._writerCount > 0) {
+      ++this.readerCountBlocked;
+      await this.writersLock.waitForUnlock();
+      --this.readerCountBlocked;
     }
+    const readerCount = ++this._readerCount;
+    // The first reader locks
+    if (readerCount === 1) {
+      this.readersRelease = await this.readersLock.acquire();
+    }
+    return () => {
+      const readerCount = --this._readerCount;
+      // The last reader unlocks
+      if (readerCount === 0) {
+        this.readersRelease();
+      }
+    };
+  }
+
+  public async acquireWrite(): Promise<() => void> {
+    ++this._writerCount;
+    const writersRelease = await this.writersLock.acquire();
+    this.readersRelease = await this.readersLock.acquire();
+    return () => {
+      this.readersRelease();
+      writersRelease();
+      --this._writerCount;
+    };
+  }
+
+  public isLocked(): boolean {
+    return this.readersLock.isLocked() || this.writersLock.isLocked();
+  }
+
+  public async waitForUnlock(): Promise<void> {
+    await Promise.all([
+      this.readersLock.waitForUnlock(),
+      this.writersLock.waitForUnlock(),
+    ]);
+    return;
   }
 }