feat: make threadsafe with mutex (#19)

Author: billiegoose

README.md

@@ -38,6 +38,12 @@ The in-memory portion of the filesystem is persisted to IndexedDB with a debounc
 The files themselves are not currently cached in memory, because I don't want to waste a lot of memory.
 Applications can always *add* an LRU cache on top of `lightning-fs` - if I add one internally and it isn't tuned well for your application, it might be much harder to work around.
 
+### Multi-threaded filesystem access
+
+Multiple tabs (and web workers) can share a filesystem. However, because SharedArrayBuffer is still not available in most browsers, the in-memory cache that makes LightningFS fast cannot be shared. If each thread was allowed to update its cache independently, then you'd have a complex distributed system and would need a fancy algorithm to resolve conflicts. Instead, I'm counting on the fact that your multi-threaded applications will NOT be IO bound, and thus a simpler strategy for sharing the filesystem will work. Filesystem access is bottlenecked by a mutex (implemented via polling and an atomic compare-and-replace operation in IndexedDB) to ensure that only one thread has access to the filesystem at a time. If the active thread is constantly using the filesystem, no other threads will get a chance. However if the active thread's filesystem goes idle - no operations are pending and no new operations are started - then after 500ms its in-memory cache is serialized and saved to IndexedDB and the mutex is released. (500ms was chosen experimentally such that an [isomorphic-git](https://github.com/isomorphic-git/isomorphic-git) `clone` operation didn't thrash the mutex.)
+
+While the mutex is being held by another thread, any fs operations will be stuck waiting until the mutex becomes available. If the mutex is not available even after ten minutes then the filesystem operations will fail with an error. This could happen if say, you are trying to write to a log file every 100ms. You can overcome this by making sure that the filesystem is allowed to go idle for >500ms every now and then.
+
 ## Usage
 
 ### `new FS(name, opts?)`
@@ -49,7 +55,7 @@ import FS from '@isomorphic-git/lightning-fs';
 const fs = new FS("testfs")
 ```
 
-**Note: do not create multiple `fs` instances using the same name.** If you do, you'll have two distinct FileSystems both fighting over the same IndexedDb store.
+**Note: It is better not to create multiple `FS` instances using the same name in a single thread.** Memory usage will be higher as each instance maintains its own cache, and throughput may be lower as each instance will have to compete over the mutex for access to the IndexedDb store.
 
 Options object:
 

karma.conf.js

@@ -34,6 +34,18 @@ module.exports = function (config) {
         watched: false,
         included: false
       },
+      {
+        pattern: 'src/**/*.worker.js',
+        served: true,
+        watched: true,
+        included: false
+      },
+      {
+        pattern: 'dist/**',
+        served: true,
+        watched: true,
+        included: false
+      },
     ],
     // list of files to exclude
     // exclude: [

package-lock.json

@@ -16,7 +16,7 @@
     "semantic-release": "semantic-release"
   },
   "dependencies": {
-    "idb-keyval": "3.1.0",
+    "@wmhilton/idb-keyval": "^3.3.0",
     "isomorphic-textencoder": "1.0.1",
     "just-debounce-it": "1.1.0",
     "just-once": "1.1.0"

package.json

@@ -5,19 +5,26 @@ const STAT = 0;
 
 module.exports = class CacheFS {
   constructor() {
-    this._root = new Map([["/", this._makeRoot()]]);
   }
   _makeRoot(root = new Map()) {
     root.set(STAT, { mode: 0o777, type: "dir", size: 0, ino: 0, mtimeMs: Date.now() });
     return root
   }
-  loadSuperBlock(superblock) {
-    if (typeof superblock === 'string') {
+  activate(superblock = null) {
+    if (superblock === null) {
+      this._root = new Map([["/", this._makeRoot()]]);
+    } else if (typeof superblock === 'string') {
       this._root = new Map([["/", this._makeRoot(this.parse(superblock))]]);
     } else {
       this._root = superblock
     }
   }
+  get activated () {
+    return !!this._root
+  }
+  deactivate () {
+    this._root = void 0
+  }
   size () {
     // subtract 1 to ignore the root directory itself from the count.
     return this._countInodes(this._root.get("/")) - 1;

src/CacheFS.js

@@ -1,4 +1,4 @@
-const idb = require("idb-keyval");
+const idb = require("@wmhilton/idb-keyval");
 
 module.exports = class IdbBackend {
   constructor(name) {
@@ -23,4 +23,7 @@ module.exports = class IdbBackend {
   wipe() {
     return idb.clear(this._store)
   }
+  close() {
+    return idb.close(this._store)
+  }
 }

src/IdbBackend.js

@@ -0,0 +1,94 @@
+const idb = require("@wmhilton/idb-keyval");
+
+const sleep = ms => new Promise(r => setTimeout(r, ms))
+
+module.exports = class Mutex {
+  constructor(name) {
+    this._id = Math.random()
+    this._database = name
+    this._store = new idb.Store(this._database + "_lock", this._database + "_lock")
+    this._has = false
+    this._keepAliveTimeout = null
+  }
+  has () {
+    return this._has
+  }
+  // Returns true if successful
+  async acquire ({ ttl = 5000, refreshPeriod } = {}) {
+    let success
+    let expired
+    let doubleLock
+    await idb.update("lock", (current) => {
+      const now = Date.now()
+      expired = current && current.expires < now
+      success = current === undefined || expired
+      doubleLock = current && current.holder === this._id
+      this._has = success || doubleLock
+      return success ? { holder: this._id, expires: now + ttl } : current
+    }, this._store)
+    if (doubleLock) {
+      throw new Error('Mutex double-locked')
+    }
+    if (success) {
+      this._keepAlive({ ttl, refreshPeriod })
+    }
+    return success
+  }
+  // check at 10Hz, give up after 10 minutes
+  async wait ({ interval = 100, limit = 6000, ttl, refreshPeriod } = {}) {
+    while (limit--) {
+      if (await this.acquire({ ttl, refreshPeriod })) return true
+      await sleep(interval)
+    }
+    throw new Error('Mutex timeout')
+  }
+  // Returns true if successful
+  async release ({ force = false } = {}) {
+    let success
+    let doubleFree
+    let someoneElseHasIt
+    this._stopKeepAlive()
+    await idb.update("lock", (current) => {
+      success = force || (current && current.holder === this._id)
+      doubleFree = current === void 0
+      someoneElseHasIt = current && current.holder !== this._id
+      this._has = !success
+      return success ? void 0 : current
+    }, this._store)
+    if (!this._has) {
+      await idb.close(this._store)
+    }
+    if (!success && !force) {
+      if (doubleFree) throw new Error('Mutex double-freed')
+      if (someoneElseHasIt) throw new Error('Mutex lost ownership')
+    }
+    return success
+  }
+  // Note: Chrome throttles & batches timers in background tabs to 1Hz,
+  // so there's not much point in having a refreshPeriod shorter than 1000.
+  // And TTL obviously needs to be greater than refreshPeriod.
+  async _keepAlive ({ ttl = 5000, refreshPeriod = 3000 } = {}) {
+    const keepAliveFn = async () => {
+      let success
+      let someoneDeletedIt
+      let someoneElseHasIt
+      await idb.update("lock", (current) => {
+        const now = Date.now()
+        someoneDeletedIt = current === void 0
+        someoneElseHasIt = current && current.holder !== this._id
+        success = !someoneDeletedIt && !someoneElseHasIt
+        this._has = success
+        return success ? { holder: this._id, expires: now + ttl } : current
+      }, this._store)
+      if (!success) this._stopKeepAlive()
+      if (someoneDeletedIt) throw new Error('Mutex was deleted')
+      if (someoneElseHasIt) throw new Error('Mutex lost ownership')
+    }
+    this._keepAliveTimeout = setInterval(keepAliveFn, refreshPeriod)
+  }
+  _stopKeepAlive () {
+    if (this._keepAliveTimeout) {
+      clearInterval(this._keepAliveTimeout)
+    }
+  }
+}