chore: handle adding data in same way as go

License: MIT
Signed-off-by: achingbrain <[email protected]>

Author: achingbrain

README.md

@@ -58,6 +58,18 @@ Loading this module through a script tag will make the `mfs` obj available in th
 <script src="https://npmcdn.com/ipfs-mfs/dist/index.js"></script>
 ```
 
+### A note on concurrency
+
+The mfs works by storing a reference to the root node's CID in LevelDB. LevelDB does not support concurrent access so there are read/write locks around bits of the code that modify the the root node's CID.
+
+A lock is kept on the main thread and any requests to read/write from workers or the main thread itself are queued pending release of the lock by the existing holder.
+
+Reads are executed together, writes are executed sequentially and prevent any reads from starting.
+
+If you are using IPFS in a single process or with the node [cluster](https://nodejs.org/api/cluster.html) module this should be completely transparent.
+
+If you are using [Web Workers](https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API) there is no way to globally listen to messages sent between workers and the main thread so you will need to also use the [observable-webworkers](https://www.npmjs.com/package/observable-webworkers) module to ensure the right message transports are set up to allow requesting/releasing the locks.
+
 ## Contribute
 
 All are welcome, please join in!

package.json

@@ -9,7 +9,8 @@
   "scripts": {
     "test": "cross-env aegir test -f test/browser.js",
     "test:node": "aegir test -t node",
-    "test:browser": "aegir test -t browser -t webworker -f test/browser.js",
+    "test:browser": "aegir test -t browser -f test/browser.js",
+    "test:webworker": "aegir test -t webworker -f test/browser.js",
     "build": "aegir build",
     "lint": "aegir lint",
     "release": "aegir release",
@@ -41,8 +42,9 @@
     "aegir": "^13.1.0",
     "chai": "^4.1.2",
     "cross-env": "^5.1.4",
+    "detect-webworker": "^1.0.0",
     "dirty-chai": "^2.0.1",
-    "ipfs": "~0.28.2",
+    "ipfs": "~0.29.0",
     "pre-commit": "^1.2.2",
     "pre-push": "~0.1.1",
     "safe-buffer": "^5.1.1",
@@ -63,7 +65,7 @@
     "ipfs-unixfs-engine": "~0.29.0",
     "is-pull-stream": "~0.0.0",
     "is-stream": "^1.1.0",
-    "key_mutex": "achingbrain/key_mutex",
+    "mortice": "^1.0.0",
     "promisify-es6": "^1.0.3",
     "pull-cat": "^1.1.11",
     "pull-paramap": "^1.2.2",

src/cli/cp.js

@@ -41,14 +41,10 @@ module.exports = {
       hash
     } = argv
 
-    ipfs.files.cp(source, dest, {
+    return ipfs.files.cp(source, dest, {
       parents,
       format,
       hashAlg: hash
-    }, (error) => {
-      if (error) {
-        throw error
-      }
     })
   }
 }

src/cli/flush.js

@@ -17,10 +17,6 @@ module.exports = {
       ipfs
     } = argv
 
-    ipfs.files.flush(path || FILE_SEPARATOR, {}, (error) => {
-      if (error) {
-        throw error
-      }
-    })
+    return ipfs.files.flush(path || FILE_SEPARATOR, {})
   }
 }

src/cli/ls.js

@@ -30,50 +30,45 @@ module.exports = {
       long
     } = argv
 
-    ipfs.files.ls(path || FILE_SEPARATOR, {
-      long
-    }, (error, files) => {
-      if (error) {
-        throw error
-      }
-
-      if (long) {
-        const table = []
-        const lengths = {}
-
-        files.forEach(link => {
-          const row = {
-            name: `${link.name}`,
-            hash: `${link.hash}`,
-            size: `${link.size}`
-          }
-
-          Object.keys(row).forEach(key => {
-            const value = row[key]
-
-            lengths[key] = lengths[key] > value.length ? lengths[key] : value.length
+    return ipfs.files.ls(path || FILE_SEPARATOR)
+      .then(files => {
+        if (long) {
+          const table = []
+          const lengths = {}
+
+          files.forEach(link => {
+            const row = {
+              name: `${link.name}`,
+              hash: `${link.hash}`,
+              size: `${link.size}`
+            }
+
+            Object.keys(row).forEach(key => {
+              const value = row[key]
+
+              lengths[key] = lengths[key] > value.length ? lengths[key] : value.length
+            })
+
+            table.push(row)
           })
 
-          table.push(row)
-        })
+          table.forEach(row => {
+            let line = ''
 
-        table.forEach(row => {
-          let line = ''
+            Object.keys(row).forEach(key => {
+              const value = row[key]
 
-          Object.keys(row).forEach(key => {
-            const value = row[key]
+              line += value.padEnd(lengths[key])
+              line += '\t'
+            })
 
-            line += value.padEnd(lengths[key] - value.length)
-            line += '\t'
+            print(line)
           })
 
-          print(line)
-        })
-
-        return
-      }
+          return
+        }
 
-      files.forEach(link => print(link.name))
-    })
+        files.forEach(link => print(link.name))
+      })
   }
 }

src/cli/mkdir.js

@@ -1,7 +1,6 @@
 'use strict'
 
 const {
-  print,
   asBoolean
 } = require('./utils')
 
@@ -45,17 +44,11 @@ module.exports = {
       flush
     } = argv
 
-    ipfs.files.mkdir(path, {
+    return ipfs.files.mkdir(path, {
       parents,
       cidVersion,
       hash,
       flush
-    }, (error, result) => {
-      if (error) {
-        throw error
-      }
-
-      print(result)
     })
   }
 }

src/cli/mv.js

@@ -35,13 +35,9 @@ module.exports = {
       recursive
     } = argv
 
-    ipfs.files.mv(source, dest, {
+    return ipfs.files.mv(source, dest, {
       parents,
       recursive
-    }, (error) => {
-      if (error) {
-        throw error
-      }
     })
   }
 }

src/cli/read.js

@@ -34,24 +34,28 @@ module.exports = {
       length
     } = argv
 
-    waterfall([
-      (cb) => ipfs.files.readPullStream(path, {
-        offset,
-        length
-      }, cb),
-      (stream, cb) => {
-        pull(
-          stream,
-          through(buffer => {
-            print(buffer, false)
-          }),
-          collect(cb)
-        )
-      }
-    ], (error) => {
-      if (error) {
-        throw error
-      }
+    return new Promise((resolve, reject) => {
+      waterfall([
+        (cb) => ipfs.files.readPullStream(path, {
+          offset,
+          length
+        }, cb),
+        (stream, cb) => {
+          pull(
+            stream,
+            through(buffer => {
+              print(buffer, false)
+            }),
+            collect(cb)
+          )
+        }
+      ], (error) => {
+        if (error) {
+          return reject(error)
+        }
+
+        resolve()
+      })
     })
   }
 }

src/cli/rm.js

@@ -26,12 +26,8 @@ module.exports = {
       recursive
     } = argv
 
-    ipfs.files.rm(path, {
+    return ipfs.files.rm(path, {
       recursive
-    }, (error) => {
-      if (error) {
-        throw error
-      }
     })
   }
 }

src/cli/stat.js

@@ -1,8 +1,8 @@
 'use strict'
 
 const {
-  print,
-  asBoolean
+  asBoolean,
+  print
 } = require('./utils')
 
 module.exports = {
@@ -54,28 +54,25 @@ Type: <type>`,
       withLocal
     } = argv
 
-    ipfs.files.stat(path, {
+    return ipfs.files.stat(path, {
       withLocal
-    }, (error, stats) => {
-      if (error) {
-        throw error
-      }
-
-      if (hash) {
-        return print(stats.hash)
-      }
+    })
+      .then((stats) => {
+        if (hash) {
+          return print(stats.hash)
+        }
 
-      if (size) {
-        return print(stats.size)
-      }
+        if (size) {
+          return print(stats.size)
+        }
 
-      print(format
-        .replace('<hash>', stats.hash)
-        .replace('<size>', stats.size)
-        .replace('<cumulsize>', stats.cumulativeSize)
-        .replace('<childs>', stats.blocks)
-        .replace('<type>', stats.type)
-      )
-    })
+        print(format
+          .replace('<hash>', stats.hash)
+          .replace('<size>', stats.size)
+          .replace('<cumulsize>', stats.cumulativeSize)
+          .replace('<childs>', stats.blocks)
+          .replace('<type>', stats.type)
+        )
+      })
   }
 }