support an fd pool to limit how many active fds are in use (#33)

* support an fd pool to limit how many active fds are in use

* pool test

* increase test timeout

Author: mafintosh

index.js

@@ -20,6 +20,31 @@ const RDONLY = constants.O_RDONLY
 const WRONLY = constants.O_WRONLY
 const CREAT = constants.O_CREAT
 
+class Pool {
+  constructor (maxSize) {
+    this.maxSize = maxSize
+    this.active = []
+  }
+
+  _onactive (file) {
+    // suspend a random one when the pool
+    if (this.active.length >= this.maxSize) {
+      const r = Math.floor(Math.random() * this.active.length)
+      this.active[r].suspend()
+    }
+
+    file._pi = this.active.push(file) - 1
+  }
+
+  _oninactive (file) {
+    const head = this.active.pop()
+    if (head !== file) {
+      head._pi = file._pi
+      this.active[head._pi] = head
+    }
+  }
+}
+
 module.exports = class RandomAccessFile extends RandomAccessStorage {
   constructor (filename, opts = {}) {
     const size = opts.size || (opts.truncate ? 0 : -1)
@@ -39,6 +64,8 @@ module.exports = class RandomAccessFile extends RandomAccessStorage {
 
     this.mode = readable && writable ? RDWR : (readable ? RDONLY : WRONLY)
 
+    this._pi = 0 // pool index
+    this._pool = opts.pool || null
     this._size = size
     this._rmdir = !!opts.rmdir
     this._lock = opts.lock === true
@@ -47,6 +74,10 @@ module.exports = class RandomAccessFile extends RandomAccessStorage {
     this._alwaysCreate = size >= 0
   }
 
+  static createPool (maxSize) {
+    return new Pool(maxSize)
+  }
+
   _open (req) {
     const create = this._alwaysCreate || this.writing // .writing comes from RAS
     const self = this
@@ -92,7 +123,7 @@ module.exports = class RandomAccessFile extends RandomAccessStorage {
 
     function ontruncate (err) {
       if (err) return onerrorafteropen(err)
-
+      if (self._pool !== null) self._pool._onactive(self)
       req.callback(null)
     }
 
@@ -180,6 +211,7 @@ module.exports = class RandomAccessFile extends RandomAccessStorage {
 
     function onclose (err) {
       if (err) return req.callback(err)
+      if (self._pool !== null) self._pool._oninactive(self)
       self.fd = 0
       req.callback(null)
     }

test/basic.js

@@ -361,6 +361,7 @@ test('truncate', function (t) {
 })
 
 test('open and close many times', function (t) {
+  t.timeout(120000) // on ci sometimes this takes a while
   t.plan(3)
 
   const name = gen()
@@ -438,6 +439,37 @@ test('unlink', async function (t) {
   }
 })
 
+test('pool', function (t) {
+  t.plan(8)
+
+  const pool = RAF.createPool(2)
+
+  const a = new RAF(gen(), { pool })
+  const b = new RAF(gen(), { pool })
+  const c = new RAF(gen(), { pool })
+
+  a.write(0, Buffer.from('hello'), function (err) {
+    t.absent(err, 'no error')
+    b.write(0, Buffer.from('hello'), function (err) {
+      t.absent(err, 'no error')
+      c.write(0, Buffer.from('hello'), function (err) {
+        t.absent(err, 'no error')
+        setTimeout(function () {
+          t.is(pool.active.length, 2)
+          const all = [a, b, c]
+          t.is(all.filter(f => f.suspended).length, 1)
+
+          for (const f of all) {
+            f.read(0, 5, function (_, buf) {
+              t.alike(buf, Buffer.from('hello'))
+            })
+          }
+        }, 100)
+      })
+    })
+  })
+})
+
 function gen () {
   return path.join(tmp, ++i + '.txt')
 }