experimental flat file datastore.
This commit is contained in:
Christopher Jeffrey
parent
3564052340
commit
f79d2cfa89
@ -14,6 +14,7 @@ var utils = bcoin.utils;
|
||||
var assert = utils.assert;
|
||||
var pad32 = utils.pad32;
|
||||
var DUMMY = new Buffer([]);
|
||||
var DataStore = require('./datastore');
|
||||
|
||||
/**
|
||||
* ChainDB
|
||||
@ -74,6 +75,7 @@ ChainDB.prototype._init = function _init() {
|
||||
cacheSize: 16 * 1024 * 1024,
|
||||
writeBufferSize: 8 * 1024 * 1024
|
||||
});
|
||||
this.db = new DataStore(this.db);
|
||||
};
|
||||
|
||||
ChainDB.prototype.load = function load(callback) {
|
||||
|
||||
661
lib/bcoin/datastore.js
Normal file
661
lib/bcoin/datastore.js
Normal file
@ -0,0 +1,661 @@
|
||||
/**
|
||||
* datastore.js - storage
|
||||
* Copyright (c) 2014-2015, Fedor Indutny (MIT License)
|
||||
* https://github.com/indutny/bcoin
|
||||
*/
|
||||
|
||||
var EventEmitter = require('events').EventEmitter;
|
||||
var bcoin = require('../bcoin');
|
||||
var levelup = require('levelup');
|
||||
var constants = bcoin.protocol.constants;
|
||||
var network = bcoin.protocol.network;
|
||||
var utils = bcoin.utils;
|
||||
var assert = utils.assert;
|
||||
var fs = bcoin.fs;
|
||||
var pad32 = utils.pad32;
|
||||
|
||||
var MAX_FILE_SIZE = 1 * 1024 * 1024;
|
||||
var nullBuffer = new Buffer([0xff, 0xff, 0xff, 0xff]);
|
||||
|
||||
/**
|
||||
* DataStore
|
||||
*/
|
||||
|
||||
function DataStore(db, options) {
|
||||
if (!(this instanceof DataStore))
|
||||
return new DataStore(db, options);
|
||||
|
||||
EventEmitter.call(this);
|
||||
|
||||
if (!options)
|
||||
options = {};
|
||||
|
||||
this.options = options;
|
||||
this.dir = options.name;
|
||||
|
||||
bcoin.ensurePrefix();
|
||||
|
||||
if (!this.dir)
|
||||
this.dir = bcoin.prefix + '/store-' + network.type + '.db';
|
||||
|
||||
// Keep a pool of FDs open for disk cache benefits
|
||||
this._db = db;
|
||||
this.db = this;
|
||||
this.pool = new bcoin.lru(50, 1, function(key, value) {
|
||||
fs.close(value.fd, function(err) {
|
||||
if (err)
|
||||
self.emit('error', err);
|
||||
});
|
||||
});
|
||||
this.busy = false;
|
||||
this.jobs = [];
|
||||
this.fileIndex = -1;
|
||||
|
||||
this._init();
|
||||
}
|
||||
|
||||
utils.inherits(DataStore, EventEmitter);
|
||||
|
||||
DataStore.prototype._lock = function _lock(func, args, force) {
|
||||
var self = this;
|
||||
var called;
|
||||
|
||||
if (force) {
|
||||
assert(this.busy);
|
||||
return function unlock() {
|
||||
assert(!called);
|
||||
called = true;
|
||||
};
|
||||
}
|
||||
|
||||
if (this.busy) {
|
||||
this.jobs.push([func, args]);
|
||||
return;
|
||||
}
|
||||
|
||||
this.busy = true;
|
||||
|
||||
return function unlock() {
|
||||
var item;
|
||||
|
||||
assert(!called);
|
||||
called = true;
|
||||
|
||||
self.busy = false;
|
||||
|
||||
if (self.jobs.length === 0) {
|
||||
self.emit('flush');
|
||||
return;
|
||||
}
|
||||
|
||||
item = self.jobs.shift();
|
||||
item[0].apply(self, item[1]);
|
||||
};
|
||||
};
|
||||
|
||||
DataStore.prototype._exists = function _exists(callback) {
|
||||
fs.stat(this.dir, function(err) {
|
||||
if (err && err.code !== 'ENOENT')
|
||||
return callback(err);
|
||||
return callback(null, err == null);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype._ensure = function _ensure(callback) {
|
||||
var self = this;
|
||||
this._exists(function(err, result) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
if (result)
|
||||
return callback();
|
||||
|
||||
return fs.mkdir(self.dir, 0750, callback);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype._init = function _init(callback) {
|
||||
var self = this;
|
||||
callback = utils.ensure(callback);
|
||||
return this._ensure(function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
self.getLastIndex(function(err, index) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
self.fileIndex = index;
|
||||
|
||||
return callback();
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.allocatePage = function allocatePage(callback) {
|
||||
var self = this;
|
||||
var index = this.fileIndex + 1;
|
||||
fs.writeFile(this.dir + '/f' + pad32(index), new Buffer([]), function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
self.openFile(index, function(err, fd, size, index) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
self.fileIndex = index;
|
||||
return callback(null, fd, size, index);
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.openFile = function openFile(index, callback) {
|
||||
var self = this;
|
||||
var entry = this.pool.get(index);
|
||||
if (entry)
|
||||
return callback(null, entry.fd, entry.size, index);
|
||||
fs.open(this.dir + '/f' + pad32(index), 'r+', function(err, fd) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
fs.fstat(fd, function(err, stat) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
self.pool.set(index, { fd: fd, size: stat.size });
|
||||
return callback(null, fd, stat.size, index);
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.currentFile = function currentFile(callback) {
|
||||
if (this.fileIndex === -1)
|
||||
return this.allocatePage(callback);
|
||||
return this.openFile(this.fileIndex, callback);
|
||||
};
|
||||
|
||||
DataStore.prototype.getLastIndex = function getLastIndex(callback) {
|
||||
var i, max, index;
|
||||
|
||||
fs.readdir(this.dir, function(err, list) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
max = -1;
|
||||
for (i = 0; i < list.length; i++) {
|
||||
if (!/^f\d{10}$/.test(list[i]))
|
||||
continue;
|
||||
index = +list[i].substring(1);
|
||||
if (index > max)
|
||||
max = index;
|
||||
}
|
||||
|
||||
return callback(null, max);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.close = function close(callback) {
|
||||
var self = this;
|
||||
return callback();
|
||||
};
|
||||
|
||||
DataStore.prototype.getData = function get(off, callback) {
|
||||
var self = this;
|
||||
|
||||
callback = utils.ensure(callback);
|
||||
|
||||
off = this.parseOffset(off);
|
||||
|
||||
this.openFile(off.fileIndex, function(err, fd, fsize) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
return self.read(fd, off.offset, off.size, callback);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.get = function get(key, callback) {
|
||||
var self = this;
|
||||
return this._db.get(key, function(err, offset) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
if (isDirect(offset))
|
||||
return callback(null, offset);
|
||||
return self.getData(offset, callback);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.batch = function batch(ops, options, callback) {
|
||||
var batch = new Batch(this);
|
||||
if (ops) {
|
||||
if (!callback) {
|
||||
callback = options;
|
||||
options = null;
|
||||
}
|
||||
if (!options)
|
||||
options = {};
|
||||
batch.ops = ops;
|
||||
return batch.write(callback);
|
||||
}
|
||||
return batch;
|
||||
};
|
||||
|
||||
function Batch(store) {
|
||||
this.ops = [];
|
||||
this.store = store;
|
||||
this._db = store._db;
|
||||
}
|
||||
|
||||
Batch.prototype.put = function(key, value) {
|
||||
this.ops.push({ type: 'put', key: key, value: value });
|
||||
};
|
||||
|
||||
Batch.prototype.del = function del(key) {
|
||||
this.ops.push({ type: 'del', key: key });
|
||||
};
|
||||
|
||||
Batch.prototype.write = function write(callback) {
|
||||
var self = this;
|
||||
var batch;
|
||||
|
||||
if (!this._db)
|
||||
return callback(new Error('Already written.'));
|
||||
|
||||
batch = this._db.batch();
|
||||
|
||||
utils.forEachSerial(this.ops, function(op, next) {
|
||||
if (op.type === 'put') {
|
||||
if (isDirect(op.value)) {
|
||||
batch.put(op.key, op.value);
|
||||
return next();
|
||||
}
|
||||
return self.store.putData(op.value, function(err, offset) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
batch.put(op.key, offset);
|
||||
next();
|
||||
});
|
||||
}
|
||||
|
||||
if (op.type === 'del') {
|
||||
return self._db.get(op.key, function(err, offset) {
|
||||
if (err && err.type !== 'NotFoundError')
|
||||
return callback(err);
|
||||
if (!offset)
|
||||
return next();
|
||||
batch.del(op.key);
|
||||
if (isDirect(offset))
|
||||
return next();
|
||||
self.store.delData(offset, next);
|
||||
});
|
||||
}
|
||||
|
||||
return callback(new Error('Bad op type.'));
|
||||
}, function(err) {
|
||||
self.ops.length = 0;
|
||||
|
||||
delete self.ops;
|
||||
delete self._db;
|
||||
delete self.store;
|
||||
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return batch.write(callback);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.iterator = function iterator(options) {
|
||||
return new Iterator(this, options);
|
||||
};
|
||||
|
||||
function Iterator(store, options) {
|
||||
this.store = store;
|
||||
this._db = store._db;
|
||||
this.iterator = this._db.db.iterator(options);
|
||||
}
|
||||
|
||||
Iterator.prototype.seek = function seek(key) {
|
||||
return this.iterator.seek(key);
|
||||
};
|
||||
|
||||
// Store coins, chain entries, dummies, lookup
|
||||
// hashes directly in the db (unless they're
|
||||
// the same length as offset).
|
||||
function isDirect(data) {
|
||||
return data.length !== 12 && data.length <= 116;
|
||||
}
|
||||
|
||||
Iterator.prototype.next = function next(callback) {
|
||||
return this.iterator.next(function(err, key, value) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
if (value) {
|
||||
if (isDirect(value))
|
||||
return callback(null, key, value);
|
||||
return self.getData(value, function(err, data) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
return callback(null, key, data);
|
||||
});
|
||||
}
|
||||
|
||||
return callback(null, key, value);
|
||||
});
|
||||
};
|
||||
|
||||
Iterator.prototype.end = function end(callback) {
|
||||
var ret = this.iterator.end(callback);
|
||||
delete this.iterator;
|
||||
delete this.store;
|
||||
delete this._db;
|
||||
return ret;
|
||||
};
|
||||
|
||||
utils.wrap = function wrap(callback, unlock) {
|
||||
return function(err, result) {
|
||||
unlock();
|
||||
if (callback)
|
||||
callback(err, result);
|
||||
};
|
||||
};
|
||||
|
||||
DataStore.prototype.putData = function putData(data, callback, force) {
|
||||
var self = this;
|
||||
var offset, undo;
|
||||
|
||||
var unlock = this._lock(putData, [data, callback], force);
|
||||
if (!unlock)
|
||||
return;
|
||||
|
||||
callback = utils.wrap(callback, unlock);
|
||||
|
||||
this.currentFile(function(err, fd, fsize, index) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return self.write(fd, fsize, data, function(err, written) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
offset = self.createOffset(index, fsize, written);
|
||||
|
||||
fsize += written;
|
||||
|
||||
// tx1 -> tx1-start/undo -> tx2 -> tx2-start/undo
|
||||
return self.write(fd, fsize, offset, function(err, written) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
fsize += written;
|
||||
|
||||
if (self.pool.has(index))
|
||||
self.pool.get(index).size = fsize;
|
||||
|
||||
if (fsize > MAX_FILE_SIZE) {
|
||||
return self.allocatePage(function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
return callback(null, offset);
|
||||
});
|
||||
}
|
||||
|
||||
return callback(null, offset);
|
||||
});
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.put = function put(key, value, callback) {
|
||||
var self = this;
|
||||
|
||||
var unlock = this._lock(put, [key, value, callback]);
|
||||
if (!unlock)
|
||||
return;
|
||||
|
||||
callback = utils.wrap(callback, unlock);
|
||||
|
||||
if (isDirect(value))
|
||||
return self._db.put(key, value, callback);
|
||||
|
||||
return this.putData(value, function(err, offset) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return self._db.put(key, offset, callback);
|
||||
}, true);
|
||||
};
|
||||
|
||||
DataStore.prototype.readUndo = function readUndo(index, offset, callback) {
|
||||
var self = this;
|
||||
var undo;
|
||||
|
||||
return this.openFile(index, function(err, fd, fsize) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return self.read(fd, offset - 12, 12, function(err, data) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return callback(null, self.parseOffset(data));
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.isNull = function isNull(index, offset, size, callback) {
|
||||
var self = this;
|
||||
var undo;
|
||||
|
||||
return this.openFile(index, function(err, fd, fsize) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return self.read(fd, offset, 4, function(err, data) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return callback(null, utils.readU32(data, 0) === 0xffffffff);
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.delData = function delData(off, callback, force) {
|
||||
var self = this;
|
||||
var undo, index, offset, size;
|
||||
|
||||
var unlock = this._lock(delData, [off, callback], force);
|
||||
if (!unlock)
|
||||
return;
|
||||
|
||||
callback = utils.wrap(callback, unlock);
|
||||
|
||||
off = this.parseOffset(off);
|
||||
index = off.fileIndex;
|
||||
offset = off.offset;
|
||||
size = off.size;
|
||||
|
||||
return this.openFile(index, function(err, fd, fsize) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
return self.write(fd, offset, nullBuffer, function(err, data) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
if (offset + size !== fsize)
|
||||
return callback();
|
||||
|
||||
// If we're deleting the last record, traverse
|
||||
// through the reverse linked list of undo offsets
|
||||
// until we hit a record that isn't deleted.
|
||||
// Truncate to the last deleted record's offset.
|
||||
(function next() {
|
||||
self.readUndo(index, offset, function(err, undo) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
self.isNull(index, undo.offset, function(err, res) {
|
||||
if (!res)
|
||||
return done();
|
||||
offset = undo.offset;
|
||||
if (offset === 0)
|
||||
return done();
|
||||
return next();
|
||||
});
|
||||
});
|
||||
})();
|
||||
|
||||
function done() {
|
||||
// Delete the file if nothing is in it.
|
||||
if (offset === 0) {
|
||||
return fs.close(fd, function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
self.pool.remove(index);
|
||||
fs.unlink(self.dir + '/f' + pad32(index), callback);
|
||||
});
|
||||
}
|
||||
if (self.pool.has(index))
|
||||
self.pool.get(index).size = offset;
|
||||
self.truncate(index, offset, callback);
|
||||
}
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.del = function del(key, callback, force) {
|
||||
var self = this;
|
||||
|
||||
var unlock = this._lock(del, [key, callback], force);
|
||||
if (!unlock)
|
||||
return;
|
||||
|
||||
callback = utils.wrap(callback, unlock);
|
||||
|
||||
this._db.get(key, function(err, off) {
|
||||
if (err && err.type !== 'NotFoundError')
|
||||
return callback(err);
|
||||
if (!off)
|
||||
return next();
|
||||
if (isDirect(off))
|
||||
return self._db.del(key, callback);
|
||||
self.delData(off, function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
self._db.del(key, callback);
|
||||
}, true);
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype.createOffset = function createOffset(fileIndex, offset, size) {
|
||||
var buf = new Buffer(12);
|
||||
utils.writeU32(buf, fileIndex, 0);
|
||||
utils.writeU32(buf, offset, 4);
|
||||
utils.writeU32(buf, size, 8);
|
||||
return buf;
|
||||
};
|
||||
|
||||
DataStore.prototype.parseOffset = function parseOffset(data) {
|
||||
return {
|
||||
fileIndex: utils.readU32(data, 0),
|
||||
offset: utils.readU32(data, 4),
|
||||
size: utils.readU32(data, 8)
|
||||
};
|
||||
};
|
||||
|
||||
DataStore.prototype.truncate = function truncate(index, size, callback) {
|
||||
var self = this;
|
||||
|
||||
callback = utils.ensure(callback);
|
||||
|
||||
self.openFile(index, function(err, fd, fsize) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
fs.ftruncate(fd, size, function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
if (self.pool.has(index))
|
||||
self.pool.get(index).size = size;
|
||||
|
||||
return callback();
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
DataStore.prototype._ioError = function _ioError(name, size, offset) {
|
||||
return new Error(name
|
||||
+ '() failed at offset '
|
||||
+ offset
|
||||
+ ' with '
|
||||
+ size
|
||||
+ ' bytes left.');
|
||||
};
|
||||
|
||||
DataStore.prototype.read = function read(fd, offset, size, callback) {
|
||||
var self = this;
|
||||
var index = 0;
|
||||
var data;
|
||||
|
||||
callback = utils.ensure(callback);
|
||||
|
||||
assert(!(offset < 0 || offset == null))
|
||||
|
||||
data = new Buffer(size);
|
||||
|
||||
(function next() {
|
||||
fs.read(fd, data, index, size, offset, function(err, bytes) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
|
||||
if (!bytes)
|
||||
return callback(self._ioError('read', size, offset));
|
||||
|
||||
index += bytes;
|
||||
size -= bytes;
|
||||
offset += bytes;
|
||||
|
||||
if (index === data.length)
|
||||
return callback(null, data);
|
||||
|
||||
next();
|
||||
});
|
||||
})();
|
||||
};
|
||||
|
||||
DataStore.prototype.write = function write(fd, offset, data, callback) {
|
||||
var self = this;
|
||||
var size = data.length;
|
||||
var index = 0;
|
||||
|
||||
callback = utils.ensure(callback);
|
||||
|
||||
assert(!(offset < 0 || offset == null));
|
||||
|
||||
(function next() {
|
||||
fs.write(fd, data, index, size, offset, function(err, bytes) {
|
||||
if (err) {
|
||||
return callback(err, index);
|
||||
}
|
||||
|
||||
if (!bytes)
|
||||
return callback(self._ioError('write', size, offset));
|
||||
|
||||
index += bytes;
|
||||
size -= bytes;
|
||||
offset += bytes;
|
||||
|
||||
if (index === data.length) {
|
||||
if (!self.fsync)
|
||||
return callback(null, index);
|
||||
return fs.fsync(fd, function(err) {
|
||||
if (err)
|
||||
return callback(err);
|
||||
return callback(null, index);
|
||||
});
|
||||
}
|
||||
|
||||
next();
|
||||
});
|
||||
})();
|
||||
};
|
||||
|
||||
module.exports = DataStore;
|
||||
Loading…
Reference in New Issue
Block a user