[wip] buffer write on memory

.gitignore

@@ -63,3 +63,5 @@ typings/
 
 
 # End of https://www.gitignore.io/api/node
+
+bench/db

bench/different_takes.js

@@ -0,0 +1,82 @@
+const bench = require('bench');
+const async = require('async');
+const _ = require('lodash');
+const rimraf = require('rimraf');
+const path = require('path');
+const fs = require('fs');
+const randomstring = require('randomstring');
+
+const LimitDB = require('../lib/db');
+
+const types = {
+  ip: {
+    size: 10,
+    per_second: 5
+  }
+};
+
+const dbBase = path.join(__dirname, 'db');
+
+if (fs.existsSync(dbBase)) {
+  rimraf.sync(dbBase);
+}
+
+fs.mkdirSync(dbBase);
+
+const db1 = new LimitDB({
+  path: path.join(dbBase, 'db1'),
+  types
+});
+
+
+const db2 = new LimitDB({
+  path: path.join(dbBase, 'db2'),
+  types,
+  flushInterval: '5m'
+});
+
+
+exports.compareCount = 10;
+exports.countPerLap = 10000;
+
+const fixture = _.range(exports.compareCount).map(() => {
+  return _.range(exports.countPerLap).map(() => {
+    return randomstring.generate(4);
+  });
+});
+
+var lapOnTest1 = 0, lapOnTest2 = 0;
+
+exports.compare = {
+  'random take on disk db' : function (done) {
+    const keys = fixture[lapOnTest1++];
+    async.forEach(keys, (key, done) => {
+      db1.take({
+        type: 'ip',
+        key
+      }, (err) => {
+        if (err) { console.error(err.message); process.exit(1); }
+        done();
+      });
+    }, done);
+  },
+  'random take with flushing' : function (done) {
+    const keys = fixture[lapOnTest2++];
+    async.forEach(keys, (key, done) => {
+      db2.take({
+        type: 'ip',
+        key
+      }, (err) => {
+        if (err) { console.error(err.message); process.exit(1); }
+        done();
+      });
+    }, done);
+  }
+};
+
+exports.done = (results) => {
+  bench.show(results);
+  process.exit(0);
+};
+
+require("bench").runMain();

bench/same_takes.js

@@ -0,0 +1,82 @@
+const bench = require('bench');
+const async = require('async');
+const _ = require('lodash');
+const rimraf = require('rimraf');
+const path = require('path');
+const fs = require('fs');
+const randomstring = require('randomstring');
+
+const LimitDB = require('../lib/db');
+
+const types = {
+  ip: {
+    size: 10,
+    per_second: 5
+  }
+};
+
+const dbBase = path.join(__dirname, 'db');
+
+if (fs.existsSync(dbBase)) {
+  rimraf.sync(dbBase);
+}
+
+fs.mkdirSync(dbBase);
+
+const db1 = new LimitDB({
+  path: path.join(dbBase, 'db1'),
+  types
+});
+
+
+const db2 = new LimitDB({
+  path: path.join(dbBase, 'db2'),
+  types,
+  flushInterval: '5m'
+});
+
+
+exports.compareCount = 20;
+exports.countPerLap = 1000;
+
+const fixture = _.range(exports.compareCount).map(() => {
+  return _.range(exports.countPerLap).map(() => {
+    return 'x';
+  });
+});
+
+var lapOnTest1 = 0, lapOnTest2 = 0;
+
+exports.compare = {
+  'same take on disk db' : function (done) {
+    const keys = fixture[lapOnTest1++];
+    async.forEach(keys, (key, done) => {
+      db1.take({
+        type: 'ip',
+        key
+      }, (err) => {
+        if (err) { console.error(err.message); process.exit(1); }
+        done();
+      });
+    }, done);
+  },
+  'same take on memory' : function (done) {
+    const keys = fixture[lapOnTest2++];
+    async.forEach(keys, (key, done) => {
+      db2.take({
+        type: 'ip',
+        key
+      }, (err) => {
+        if (err) { console.error(err.message); process.exit(1); }
+        done();
+      });
+    }, done);
+  }
+};
+
+exports.done = (results) => {
+  bench.show(results);
+  process.exit(0);
+};
+
+require("bench").runMain();

bench/same_takes_diff_ticks.js

@@ -0,0 +1,79 @@
+const bench = require('bench');
+const async = require('async');
+const _ = require('lodash');
+const rimraf = require('rimraf');
+const path = require('path');
+const fs = require('fs');
+
+const LimitDB = require('../lib/db');
+
+const types = {
+  ip: {
+    size: 10,
+    per_second: 5
+  }
+};
+
+const dbBase = path.join(__dirname, 'db');
+
+if (fs.existsSync(dbBase)) {
+  rimraf.sync(dbBase);
+}
+
+fs.mkdirSync(dbBase);
+
+const db1 = new LimitDB({
+  path: path.join(dbBase, 'db1'),
+  types
+});
+
+
+const db2 = new LimitDB({
+  path: path.join(dbBase, 'db2'),
+  types,
+  flushInterval: '5m'
+});
+
+
+exports.compareCount = 20;
+exports.countPerLap = 5;
+
+const fixture = _.range(exports.compareCount).map(() => {
+  return ['a', 'b', 'c', 'd', 'e'];
+});
+
+var lapOnTest1 = 0, lapOnTest2 = 0;
+
+exports.compare = {
+  'same take on disk db' : function (done) {
+    const keys = fixture[lapOnTest1++];
+    async.forEach(keys, (key, done) => {
+      db1.take({
+        type: 'ip',
+        key
+      }, (err) => {
+        if (err) { console.error(err.message); process.exit(1); }
+        done();
+      });
+    }, done);
+  },
+  'same take on memory' : function (done) {
+    const keys = fixture[lapOnTest2++];
+    async.forEach(keys, (key, done) => {
+      db2.take({
+        type: 'ip',
+        key
+      }, (err) => {
+        if (err) { console.error(err.message); process.exit(1); }
+        done();
+      });
+    }, done);
+  }
+};
+
+exports.done = (results) => {
+  bench.show(results);
+  process.exit(0);
+};
+
+require("bench").runMain();

lib/DbBuffer.js

@@ -0,0 +1,137 @@
+'use strict';
+
+const ms = require('ms');
+const async = require('async');
+const _ = require('lodash');
+
+//inspired by https://github.com/dominictarr/hashlru
+
+class DbBuffer {
+  constructor(params) {
+    this.max = 5000;
+    this.size = 0;
+    this.current = Object.create(null);
+    this.previous = Object.create(null);
+
+    this._db = params.db;
+
+    const flushInterval = typeof params.flushInterval === 'string' ?
+                            ms(params.flushInterval):
+                            params.flushInterval;
+
+    this._setFlushInterval(flushInterval || ms('5m'));
+
+    this.flushed = false;
+    this.locks = {};
+  }
+
+  _setFlushInterval(flushInterval) {
+    setTimeout(() => {
+      this.flush(() => {
+        this._setFlushInterval(flushInterval);
+      });
+    }, flushInterval);
+  }
+
+  _flushStage(stage, done) {
+    async.forEach(Object.keys(stage), (key, saved) => {
+      this._db.put(key, stage[key], saved);
+    }, done);
+  }
+
+  flush(callback) {
+    this.flushed = true;
+
+    async.parallel([
+      done => this._flushStage(this.current, done),
+      done => this._flushStage(this.previous, done)
+    ], callback || _.noop);
+  }
+
+  _update(key, value) {
+    this.current[key] = value;
+    this.size++;
+
+    if(this.size >= this.max) {
+      this.flush();
+      this.size = 0;
+      this.previous = this.current;
+      this.current = Object.create(null);
+    }
+  }
+
+  _set(key, value) {
+    if(this.current[key] !== undefined) {
+      this.current[key] = value;
+    } else {
+      this._update(key, value);
+    }
+    return value;
+  }
+
+  _get(key) {
+    const fromCurrent = this.current[key];
+
+    if(fromCurrent !== undefined) {
+      return fromCurrent;
+    }
+
+    const fromPrevious = this.previous[key];
+
+    if(fromPrevious) {
+      this._update(key, fromPrevious);
+      return fromPrevious;
+    }
+  }
+
+  lock(key, run) {
+    if (!this.locks[key]) {
+      this.locks[key] = [];
+      run(() => {
+        const waiting = this.locks[key];
+        var cb;
+        while((cb = waiting.shift())) {
+          cb(_.noop);
+        }
+        this.locks[key] = false;
+      });
+    } else {
+      this.locks[key].push(run);
+    }
+  }
+
+  get(key, typeParams, callback) {
+    this.flushed = false;
+
+    this.lock(key, release => {
+      const fromBuffer = this._get(key);
+      if (fromBuffer) {
+        callback(null, fromBuffer);
+        return release();
+      }
+
+      this._db.get(key, (err, bucket) => {
+        if (err && err.name !== 'NotFoundError') {
+          callback(err);
+          return release();
+        }
+
+        if (typeof bucket === 'string') {
+          bucket = JSON.parse(bucket);
+        } else if (typeof bucket === 'undefined') {
+          bucket = {
+            lastDrip: Date.now(),
+            content: typeParams.size,
+            size: typeParams.size
+          };
+        }
+
+        this._set(key, bucket);
+        callback(null, bucket);
+        release();
+      });
+    });
+  }
+}
+
+module.exports = DbBuffer;