const fs = require('graceful-fs');
|
const join = require('path').join;
|
const Readable = require('stream').Readable;
|
|
const _mkdirp = require('mkdirp');
|
const _rimraf = require('rimraf');
|
const writeJsonFile = require('write-json-file');
|
|
const entries = require('./util/Object.entries');
|
const values = require('./util/Object.values');
|
const promisify = require('./util/promisify');
|
|
const rimraf = promisify(_rimraf);
|
const open = promisify(fs.open);
|
const close = promisify(fs.close);
|
const read = promisify(fs.read);
|
const readFile = promisify(fs.readFile);
|
const write = promisify(fs.write);
|
const rename = promisify(fs.rename);
|
const unlink = promisify(fs.unlink);
|
const stat = promisify(fs.stat);
|
const mkdirp = promisify(_mkdirp);
|
|
const APPEND_VERSION = 1;
|
|
const _blockSize = 4 * 1024;
|
const _logSize = 2 * 1024 * 1024;
|
const _minCompactSize = 512 * 1024;
|
const _compactMultiplierThreshold = 1.5;
|
|
const value = (key, size, start) => ({
|
key,
|
size: size || 0,
|
start: start || 0,
|
});
|
|
const objFrom = map => {
|
if (map instanceof Map) {
|
const obj = {};
|
map.forEach((value, key) => {
|
obj[key] = value;
|
});
|
return obj;
|
}
|
return map;
|
};
|
|
const table = ({ nextByte, blockSize, logSize, map }) => ({
|
version: APPEND_VERSION,
|
nextByte: nextByte,
|
blockSize: blockSize,
|
logSize: logSize,
|
map: objFrom(map),
|
});
|
|
const modTable = ({ nextByte, blockSize, logSize, map }) => ({
|
version: APPEND_VERSION,
|
nextByte: nextByte,
|
blockSize: blockSize,
|
logSize: logSize,
|
map: new Map(entries(map)),
|
});
|
|
function putKey(_table, key, size) {
|
// _table.map[key] = value(key, size, _table.nextByte, Math.ceil(size / _table.blockSize));
|
_table.map.set(key, value(key, size, _table.nextByte));
|
_table.nextByte = _table.nextByte + size;
|
return _table;
|
}
|
|
function delKey(_table, key) {
|
// if (_table.map[key]) {
|
// delete _table.map[key];
|
if (_table.map.get(key)) {
|
_table.map.delete(key);
|
}
|
return _table;
|
}
|
|
const _tablepath = ({ path }) => join(path, 'table.json');
|
|
const _defaultTable = ({ blockSize, logSize }) =>
|
table({
|
nextByte: 0,
|
blockSize: blockSize || _blockSize,
|
logSize: logSize || _logSize,
|
map: {},
|
});
|
|
const timeout100 = () => new Promise(resolve => setTimeout(resolve, 100));
|
|
const _retry = (fn, n) => {
|
n = n || 5;
|
const _retryFn = value => {
|
if (n) {
|
n--;
|
return fn(value).catch(_retryFn);
|
}
|
return fn(value);
|
};
|
return _retryFn;
|
};
|
|
const _readTable = _this =>
|
readFile(_tablepath(_this), 'utf8')
|
.catch(e => JSON.stringify(_defaultTable(_this)))
|
.then(JSON.parse)
|
.then(_table => {
|
if (_table.version !== APPEND_VERSION) {
|
return _defaultTable(_this);
|
}
|
return _table;
|
});
|
|
const _writeTable = (_this, _table) => writeJsonFile(_tablepath(_this), _table);
|
|
const _logFilepath = ({ path }, { logSize }, index) => {
|
let logId = ((index / logSize) | 0).toString();
|
while (logId.length < 4) {
|
logId = `0${logId}`;
|
}
|
return join(path, `log${logId}`);
|
};
|
|
const _openLog = (_this, mode, _table, index) => {
|
if (_this._fd !== null) {
|
return Promise.resolve();
|
} else {
|
// If mode is 'a', stat the log to write to, if it should be empty and
|
// isn't, unlink before opening.
|
return Promise.resolve()
|
.then(() => {
|
if (mode === 'a' && index % _table.logSize === 0) {
|
return stat(_logFilepath(_this, _table, index))
|
.then(({ size }) => {
|
if (size > 0) {
|
return unlink(_logFilepath(_this, _table, index)).then(
|
timeout100,
|
);
|
}
|
})
|
.catch(() => {});
|
}
|
})
|
.then(() => open(_logFilepath(_this, _table, index), mode))
|
.then(fd => {
|
_this._fd = fd;
|
if (mode === 'a') {
|
_this._writeBuffer = new Buffer(_table.logSize);
|
_this._writeOffset = 0;
|
}
|
})
|
.catch(e => {
|
throw e;
|
});
|
}
|
};
|
|
const _closeLog = _this => {
|
if (_this._fd === null) {
|
return Promise.resolve();
|
} else {
|
return Promise.resolve()
|
.then(() => {
|
if (_this._writeBuffer) {
|
return write(_this._fd, _this._writeBuffer, 0, _this._writeOffset);
|
}
|
})
|
.then(() => close(_this._fd))
|
.then(() => {
|
_this._fd = null;
|
_this._writeBuffer = null;
|
_this._writeOffset = 0;
|
});
|
}
|
};
|
|
const _readBufferSize = (_this, { blockSize, logSize }) =>
|
Math.min(32 * blockSize, logSize);
|
|
const _readLog = (_this, _table) => {
|
let index = 0;
|
const out = new Readable({
|
read() {},
|
});
|
|
const rbSize = _table.logSize;
|
const _readBuffer = new Buffer(rbSize);
|
|
function _log() {
|
if (index >= _table.nextByte) {
|
out.push(null);
|
return _closeLog(_this);
|
}
|
|
const offset = 0;
|
function step() {
|
if (!_this._fd) {
|
index = _table.nextByte;
|
return _log();
|
}
|
|
return read(_this._fd, _readBuffer, 0, rbSize, 0).then(read => {
|
index += _table.logSize;
|
out.push(_readBuffer);
|
return _log();
|
});
|
}
|
|
return _closeLog(_this)
|
.then(() => _openLog(_this, 'r', _table, index))
|
.then(step);
|
}
|
Promise.resolve().then(_log);
|
|
return out;
|
};
|
|
const _appendBlock = (_this, _table, blockContent, index, next) => {
|
let prep;
|
if (_this._fd !== null && index % _table.logSize === 0) {
|
prep = _closeLog(_this).then(() => _openLog(_this, 'a', _table, index));
|
} else if (_this._fd === null) {
|
prep = _openLog(_this, 'a', _table, index);
|
}
|
function work() {
|
if (!_this._fd) {
|
return next(new Error());
|
}
|
if (blockContent.length > _table.logSize) {
|
return next(new Error('block longer than max size'));
|
}
|
const writeSlice = _this._writeBuffer.slice(
|
_this._writeOffset,
|
_this._writeOffset + blockContent.length,
|
);
|
// if (blockContent.length < _table.blockSize) {
|
// writeSlice.fill(0);
|
// }
|
blockContent.copy(writeSlice);
|
_this._writeOffset += blockContent.length;
|
if (_this._writeOffset > _this._writeBuffer.length) {
|
return next(
|
new Error(
|
`writeOffset ${_this._writeOffset} past writeBuffer length ${
|
_this._writeBuffer.length
|
}`,
|
),
|
);
|
}
|
if (_this._writeOffset > _table.logSize) {
|
return next(
|
new Error(
|
`writeOffset ${_this._writeOffset} past logSize ${_table.logSize}`,
|
),
|
);
|
}
|
next();
|
// return fs.write(_this._fd, blockContent, 0, _table.blockSize, next);
|
}
|
if (prep) {
|
prep.then(work);
|
} else {
|
work();
|
}
|
|
// return Promise.resolve()
|
// .then(function() {
|
// if (index % (_table.logSize / _table.blockSize) === 0) {
|
// return _closeLog(_this);
|
// }
|
// })
|
// .then(function() {
|
// return _openLog(_this, 'a', _table, index);
|
// })
|
// .then(function() {
|
// if (!_this._fd) {
|
// throw new Error();
|
// }
|
// if (blockContent.length > _table.blockSize) {
|
// throw new Error('block longer than max size');
|
// }
|
// if (blockContent.length < _table.blockSize) {
|
// var _blockContent = new Buffer(_table.blockSize);
|
// blockContent.copy(_blockContent);
|
// blockContent = _blockContent;
|
// }
|
// return write(_this._fd, blockContent, 0, _table.blockSize);
|
// });
|
};
|
|
const _sizeNeeded = (_this, { map }) =>
|
values(map).reduce((carry, { size }) => carry + size, 0);
|
|
const _sizeUsed = (_this, { nextByte }) => nextByte;
|
|
const _compactSize = (_this, _table) =>
|
Math.max(
|
_this.compactSizeThreshold,
|
_sizeNeeded(_this, _table) * _this.compactMultiplierThreshold,
|
);
|
|
const _lock = (_this, mustLock, promiseFn) => {
|
if (mustLock !== false) {
|
return (_this.lock = promiseFn(_this.lock));
|
}
|
return promiseFn(Promise.resolve());
|
};
|
|
const serialFsTask = (array, each) =>
|
new Promise((resolve, reject) => {
|
let queue = 0;
|
let index = 0;
|
let inNext = false;
|
function next(err) {
|
if (err) {
|
return reject(err);
|
}
|
if (index === array.length) {
|
return resolve();
|
}
|
queue++;
|
if (inNext) {
|
return;
|
}
|
inNext = true;
|
while (queue > index && index < array.length) {
|
try {
|
each(array[index++], next);
|
} catch (e) {
|
return next(e);
|
}
|
}
|
inNext = false;
|
}
|
next();
|
});
|
|
class AppendSerializer {
|
constructor(options) {
|
this.path = options.cacheDirPath;
|
this.autoParse = options.autoParse;
|
this.blockSize = options.blockSize || _blockSize;
|
this.logSize = options.logSize || _logSize;
|
this.compactSizeThreshold = options.compactSizeThreshold || _minCompactSize;
|
this.compactMultiplierThreshold =
|
options.compactMultiplierThreshold || _compactMultiplierThreshold;
|
|
this.lock = Promise.resolve();
|
this._fd = null;
|
}
|
|
read(mustLock) {
|
const start = Date.now();
|
const _this = this;
|
|
function _read() {
|
let activeTable;
|
return Promise.resolve()
|
.then(_retry(() => _readTable(_this)))
|
.then(_table => {
|
activeTable = _table;
|
})
|
.then(() => {
|
const map = new Map();
|
|
const valueStarts = [];
|
values(activeTable.map).forEach(value => {
|
valueStarts.push({
|
start: value.start,
|
end: value.start + value.size,
|
value,
|
});
|
});
|
valueStarts.sort((a, b) => a.start - b.start);
|
|
return new Promise((resolve, reject) => {
|
let valueIndex = 0;
|
let destBuffer = new Buffer(2 * 1024 * 1024);
|
let offset = 0;
|
let logOffset = 0;
|
const log = _readLog(_this, activeTable);
|
log.on('data', data => {
|
if (valueIndex >= valueStarts.length) {
|
return;
|
}
|
for (let bufferIndex = 0; bufferIndex < data.length; ) {
|
if (bufferIndex + logOffset >= valueStarts[valueIndex].end) {
|
valueIndex++;
|
}
|
if (valueIndex >= valueStarts.length) {
|
return;
|
}
|
const value = valueStarts[valueIndex].value;
|
if (bufferIndex + logOffset >= value.start) {
|
if (value.size > destBuffer.length) {
|
const newLength = Math.pow(
|
2,
|
Math.ceil(Math.log(value.size) / Math.log(2)),
|
);
|
destBuffer = new Buffer(newLength);
|
}
|
|
const readAmount = Math.min(
|
value.start + value.size - logOffset - bufferIndex,
|
activeTable.logSize - bufferIndex,
|
);
|
data
|
.slice(bufferIndex, bufferIndex + readAmount)
|
.copy(destBuffer.slice(offset, offset + readAmount));
|
bufferIndex += readAmount;
|
offset += readAmount;
|
|
if (offset >= value.size) {
|
offset = 0;
|
if (_this.autoParse) {
|
// console.log(value.size, destBuffer.utf8Slice(0, value.size))
|
map.set(
|
value.key,
|
JSON.parse(destBuffer.utf8Slice(0, value.size)),
|
);
|
} else {
|
map.set(value.key, destBuffer.utf8Slice(0, value.size));
|
}
|
}
|
} else if (bufferIndex + logOffset < value.start) {
|
bufferIndex += value.start - (bufferIndex + logOffset);
|
}
|
}
|
logOffset += activeTable.logSize;
|
});
|
log.on('end', resolve);
|
log.on('error', reject);
|
}).then(() => objFrom(map));
|
});
|
}
|
|
return _lock(_this, mustLock, promise =>
|
promise
|
.then(() => _read())
|
.catch(e =>
|
_closeLog(_this).then(() => {
|
throw e;
|
}),
|
),
|
);
|
}
|
|
write(ops, mustLock) {
|
if (ops.length === 0) {
|
return Promise.resolve();
|
}
|
|
const steps = 0;
|
const _this = this;
|
|
let activeTable;
|
let contentBuffer;
|
let contentLength;
|
function _write() {
|
return Promise.resolve()
|
.then(_retry(() => mkdirp(_this.path)))
|
.then(_retry(() => _readTable(_this)))
|
.then(_table => {
|
activeTable = modTable(_table);
|
const _ops = ops.slice();
|
function step(op, next) {
|
// steps++;
|
// var op = _ops.shift();
|
// if (!op) {
|
// return;
|
// }
|
|
let content = op.value;
|
if (content !== null) {
|
if (typeof content !== 'string') {
|
content = JSON.stringify(content);
|
}
|
|
if (
|
Buffer.byteLength &&
|
contentBuffer &&
|
Buffer.byteLength(content) <= contentBuffer.length
|
) {
|
contentLength = contentBuffer.utf8Write(content);
|
} else {
|
contentBuffer = new Buffer(content);
|
contentLength = contentBuffer.length;
|
}
|
|
const blockCount = Math.ceil(
|
((activeTable.nextByte % activeTable.logSize) + contentLength) /
|
activeTable.logSize,
|
);
|
let nextByte = activeTable.nextByte;
|
activeTable = putKey(activeTable, op.key, contentLength);
|
let bufferIndex = 0;
|
|
const bulk = Array.from(new Array(blockCount)).map((_, i) => i);
|
return serialFsTask(bulk, (_, next) => {
|
const blockSlice = contentBuffer.slice(
|
bufferIndex,
|
Math.min(
|
bufferIndex +
|
(activeTable.logSize - (nextByte % activeTable.logSize)),
|
contentLength,
|
),
|
);
|
_appendBlock(_this, activeTable, blockSlice, nextByte, next);
|
bufferIndex += blockSlice.length;
|
nextByte += blockSlice.length;
|
}).then(next);
|
|
// function append() {
|
// if (bufferIndex < contentBuffer.length) {
|
// var blockSlice = contentBuffer.slice(bufferIndex, bufferIndex + activeTable.blockSize);
|
// bufferIndex += activeTable.blockSize;
|
// return _appendBlock(_this, activeTable, blockSlice, nextByte++)
|
// .then(append);
|
// }
|
// }
|
// return append()
|
// .then(step);
|
} else {
|
activeTable = delKey(activeTable, op.key);
|
next();
|
}
|
}
|
|
return serialFsTask(_ops, step);
|
|
// return step();
|
})
|
.then(() => _closeLog(_this))
|
.then(
|
_retry(() => {
|
activeTable = table(activeTable);
|
return _writeTable(_this, activeTable);
|
}),
|
);
|
}
|
|
return _lock(_this, mustLock, promise =>
|
promise
|
.then(() => _write())
|
.catch(e =>
|
_closeLog(_this).then(() => {
|
throw e;
|
}),
|
)
|
.then(() => {
|
if (
|
_sizeUsed(_this, activeTable) > _compactSize(_this, activeTable)
|
) {
|
return _this.compact(false);
|
}
|
}),
|
);
|
}
|
|
compact(mustLock) {
|
const _this = this;
|
|
return _this
|
.read(mustLock)
|
.then(map => {
|
const ops = [];
|
Object.keys(map).forEach(key => {
|
ops.push({
|
key,
|
value: map[key],
|
});
|
});
|
return ops;
|
})
|
.then(ops =>
|
rimraf(`${_this.path}~`)
|
.then(timeout100)
|
.then(() => ops),
|
)
|
.then(ops => {
|
const copy = new AppendSerializer({
|
cacheDirPath: `${_this.path}~`,
|
|
blockSize: _this.blockSize,
|
logSize: _this.logSize,
|
compactSizeThreshold: _this.compactSizeThreshold,
|
compactMultiplierThreshold: _this.compactMultiplierThreshold,
|
});
|
|
return _lock(_this, mustLock, promise =>
|
promise
|
.then(() => copy.write(ops))
|
.then(() => rimraf(_this.path))
|
.then(timeout100)
|
.then(_retry(() => rename(copy.path, _this.path), 10)),
|
);
|
});
|
}
|
}
|
|
module.exports = AppendSerializer;
|
