win: serialize read/write/appendFile and rename

README.md

@@ -23,7 +23,9 @@ graceful-fs:
 
 On Windows, it retries renaming a file for up to one second if `EACCESS`
 or `EPERM` error occurs, likely because antivirus software has locked
-the directory.
+the directory. It also queues up `rename` while `readFile`, `writeFile`
+or `appendFile` are running on the target file and vice-versa, to avoid
+conflicts replacing a file being accessed by these calls.
 
 ## USAGE
 

graceful-fs.js

@@ -24,6 +24,48 @@ if (/\bgfs4\b/i.test(process.env.NODE_DEBUG || '')) {
   })
 }
 
+var fileEnqueue = function (file, elem) {
+  return elem[1].apply(null, elem[2])
+}
+var fileNext = function (file) {}
+
+if (process.platform === 'win32') {
+  var fileQueues = {}
+
+  fileEnqueue = function (file, elem) {
+    var queue = fileQueues[file]
+    if (!queue) {
+      fileQueues[file] = [elem[0]]
+      elem[1].apply(null, elem[2])
+    } else if ((queue[queue.length - 1] === 'share')
+               && (elem[0] === 'share')) {
+      queue.push('share')
+      elem[1].apply(null, elem[2])
+    } else {
+      queue.push(elem)
+    }
+  }
+
+  fileNext = function (file) {
+    var queue = fileQueues[file]
+    var prev = queue.shift()
+    require('assert').equal(typeof prev, 'string')
+    if (queue.length === 0) {
+      delete fileQueues[file]
+    } else if (queue[0][0] === 'lock') {
+      var elem = queue[0]
+      queue[0] = 'lock'
+      elem[1].apply(null, elem[2])
+    } else {
+      for (var i = 0; (i < queue.length) && (queue[i][0] === 'share'); ++i) {
+        var elem = queue[i]
+        queue[i] = 'share'
+        elem[1].apply(null, elem[2])
+      }
+    }
+  }
+}
+
 module.exports = patch(require('./fs.js'))
 if (process.env.TEST_GRACEFUL_FS_GLOBAL_PATCH) {
   module.exports = patch(fs)
@@ -53,7 +95,7 @@ fs.closeSync = (function (fs$closeSync) { return function (fd) {
 
 function patch (fs) {
   // Everything that references the open() function needs to be in here
-  polyfills(fs)
+  polyfills(fs, fileEnqueue, fileNext)
   fs.gracefulify = patch
   fs.FileReadStream = ReadStream;  // Legacy name.
   fs.FileWriteStream = WriteStream;  // Legacy name.
@@ -68,15 +110,16 @@ function patch (fs) {
     return go$readFile(path, options, cb)
 
     function go$readFile (path, options, cb) {
-      return fs$readFile(path, options, function (err) {
+      fileEnqueue(path, [ 'share', fs$readFile, [path, options, function (err) {
+        fileNext(path)
         if (err && (err.code === 'EMFILE' || err.code === 'ENFILE'))
           enqueue([go$readFile, [path, options, cb]])
         else {
           if (typeof cb === 'function')
             cb.apply(this, arguments)
           retry()
         }
-      })
+      }]])
     }
   }
 
@@ -89,15 +132,16 @@ function patch (fs) {
     return go$writeFile(path, data, options, cb)
 
     function go$writeFile (path, data, options, cb) {
-      return fs$writeFile(path, data, options, function (err) {
+      fileEnqueue(path, [ 'share', fs$writeFile, [path, data, options, function (err) {
+        fileNext(path)
         if (err && (err.code === 'EMFILE' || err.code === 'ENFILE'))
           enqueue([go$writeFile, [path, data, options, cb]])
         else {
           if (typeof cb === 'function')
             cb.apply(this, arguments)
           retry()
         }
-      })
+      }]])
     }
   }
 
@@ -111,15 +155,16 @@ function patch (fs) {
     return go$appendFile(path, data, options, cb)
 
     function go$appendFile (path, data, options, cb) {
-      return fs$appendFile(path, data, options, function (err) {
+      fileEnqueue(path, [ 'share', fs$appendFile, [path, data, options, function (err) {
+        fileNext(path)
         if (err && (err.code === 'EMFILE' || err.code === 'ENFILE'))
           enqueue([go$appendFile, [path, data, options, cb]])
         else {
           if (typeof cb === 'function')
             cb.apply(this, arguments)
           retry()
         }
-      })
+      }]])
     }
   }
 

polyfills.js

@@ -20,7 +20,7 @@ process.chdir = function(d) {
 
 module.exports = patch
 
-function patch (fs) {
+function patch (fs, fileEnqueue, fileNext) {
   // (re-)implement some things that are known busted or missing.
 
   // lchmod, broken prior to 0.6.2
@@ -75,15 +75,18 @@ function patch (fs) {
   // created files.  Try again on failure, for up to 1 second.
   if (process.platform === "win32") {
     fs.rename = (function (fs$rename) { return function (from, to, cb) {
-      var start = Date.now()
-      fs$rename(from, to, function CB (er) {
-        if (er
-            && (er.code === "EACCES" || er.code === "EPERM")
-            && Date.now() - start < 1000) {
-          return fs$rename(from, to, CB)
-        }
-        if (cb) cb(er)
-      })
+      fileEnqueue(to, [ 'lock', function (from, to, cb) {
+        var start = Date.now()
+        fs$rename(from, to, function CB (er) {
+          if (er
+              && (er.code === "EACCES" || er.code === "EPERM")
+              && Date.now() - start < 1000) {
+            return fs$rename(from, to, CB)
+          }
+          fileNext(to)
+          if (cb) cb(er)
+        })
+      }, [from, to, cb]])
     }})(fs.rename)
   }
 

test/concurrentrename.js

@@ -0,0 +1,84 @@
+'use strict'
+
+var fs = require('../')
+var rimraf = require('rimraf')
+var mkdirp = require('mkdirp')
+var test = require('tap').test
+var p = require('path').resolve(__dirname, 'files')
+
+process.chdir(__dirname)
+
+// Make sure to reserve the stderr fd
+process.stderr.write('')
+
+var num = 1025
+var paths = new Array(num)
+
+test('prepare files', function (t) {
+  rimraf.sync(p)
+  mkdirp.sync(p)
+
+  t.plan(num + 1)
+  for (var i = 0; i < num; ++i) {
+    paths[i] = 'files/file-' + i
+    fs.writeFile(paths[i], 'content-rename-' + i, 'ascii', function (er) {
+      if (er)
+        throw er
+      t.pass('written')
+    })
+  }
+  fs.writeFile('files/file', 'initial', 'ascii', function (er) {
+    if (er)
+      throw er
+    t.pass('written')
+  })
+})
+
+test('read and replace files', function (t) {
+  t.plan(num * 4)
+  var queue = []
+  function CB (er) {
+    if (er)
+      throw er
+    t.pass('renamed')
+  }
+  for (var i = 0; i < num; ++i) {
+    (function (i) {
+      queue.push(function () { fs.readFile('files/file', 'ascii', CB) })
+      queue.push(function () { fs.writeFile('files/file', 'write-' + i, 'ascii', CB) })
+      queue.push(function () { fs.appendFile('files/file', 'append-' + i, 'ascii', CB) })
+      queue.push(function () { fs.rename(paths[i], 'files/file', CB) })
+    })(i)
+  }
+  function swap (arr, a, b) {
+    var tmp = arr[a]
+    arr[a] = arr[b]
+    arr[b] = tmp
+  }
+  for (var i = queue.length; i >= 2; --i) {
+    swap(queue, i - 1, Math.floor(Math.random() * i))
+  }
+  for (var i = 0; i < queue.length; ++i) {
+    queue[i]()
+  }
+})
+
+test('confirm renames', function (t) {
+  t.plan(num)
+  for (var i = 0; i < num; ++i) {
+    if (fs.access) {
+      fs.access(paths[i], function (er) {
+        t.equal(er.code, 'ENOENT', 'was renamed')
+      })
+    } else {
+      fs.exists(paths[i], function (exists) {
+        t.notOk(exists, 'was renamed')
+      })
+    }
+  }
+})
+
+test('cleanup', function (t) {
+  rimraf.sync(p)
+  t.end()
+})