mempool: persist to disk.

2017-02-28 11:32:32 -08:00 · 2017-02-28 11:32:32 -08:00 · 7bd89b35c3
commit 7bd89b35c3
parent 08936c8970
8 changed files with 334 additions and 33 deletions
--- a/etc/sample.conf
+++ b/etc/sample.conf
@ -46,6 +46,7 @@ limit-free: true
 limit-free-relay: 15
 reject-absurd-fees: true
 replace-by-fee: false
+persistent-mempool: false

 #
 # Pool
@ -89,6 +90,9 @@ known-peers: ./known-peers

 coinbase-flags: mined by bcoin
 # payout-address: 1111111111111111111114oLvT2,1111111111111111111114oLvT2
+max-block-weight: 4000000
+reserved-block-weight: 4000
+reserved-block-sigops: 400

 #
 # HTTP
--- a/lib/mempool/layout-browser.js
+++ b/lib/mempool/layout-browser.js
@ -6,13 +6,14 @@

 'use strict';

-var util = require('../utils/util');
-var pad32 = util.pad32;
-
 var layout = {
  R: 'R',
-  e: function e(id, hash) {
-    return 'e' + pad32(id) + hex(hash);
+  V: 'V',
+  e: function e(hash) {
+    return 'e' + hex(hash);
+  },
+  ee: function ee(key) {
+    return key.slice(1, 65);
  }
 };

--- a/lib/mempool/layout.js
+++ b/lib/mempool/layout.js
@ -9,17 +9,21 @@
 /*
 * Database Layout:
 *   R -> tip hash
+ *   V -> db version
 *   e[id][hash] -> entry
 */

 var layout = {
  R: new Buffer([0x52]),
-  e: function e(id, hash) {
-    var key = new Buffer(37);
+  V: new Buffer([0x76]),
+  e: function e(hash) {
+    var key = new Buffer(33);
    key[0] = 0x65;
-    key.writeUInt32BE(id, 1, true);
-    write(key, hash, 5);
+    write(key, hash, 1);
    return key;
+  },
+  ee: function ee(key) {
+    return key.toString('hex', 1, 33);
  }
 };

--- a/lib/mempool/mempool.js
+++ b/lib/mempool/mempool.js
@ -24,6 +24,9 @@ var Coin = require('../primitives/coin');
 var TXMeta = require('../primitives/txmeta');
 var MempoolEntry = require('./mempoolentry');
 var Network = require('../protocol/network');
+var encoding = require('../utils/encoding');
+var layout = require('./layout');
+var LDB = require('../db/ldb');
 var VerifyError = errors.VerifyError;
 var VerifyResult = errors.VerifyResult;

@ -73,11 +76,14 @@ function Mempool(options) {

  this.locker = this.chain.locker;

+  this.cache = new MempoolCache(this.options);
+
  this.size = 0;
  this.totalOrphans = 0;
  this.totalTX = 0;
  this.freeCount = 0;
  this.lastTime = 0;
+  this.lastUpdate = 0;

  this.waiting = {};
  this.orphans = {};
@ -100,7 +106,20 @@ util.inherits(Mempool, AsyncObject);

 Mempool.prototype._open = co(function* open() {
  var size = (this.options.maxSize / 1024).toFixed(2);
+  var i, entries, entry;
+
  yield this.chain.open();
+  yield this.cache.open();
+
+  if (this.options.persistent) {
+    entries = yield this.cache.load();
+
+    for (i = 0; i < entries.length; i++) {
+      entry = entries[i];
+      this.trackEntry(entry);
+    }
+  }
+
  this.logger.info('Mempool loaded (maxsize=%dkb).', size);
 });

@ -110,9 +129,9 @@ Mempool.prototype._open = co(function* open() {
 * @returns {Promise}
 */

-Mempool.prototype._close = function close() {
-  return Promise.resolve();
-};
+Mempool.prototype._close = co(function* close() {
+  yield this.cache.close();
+});

 /**
 * Notify the mempool that a new block has come
@ -127,7 +146,7 @@ Mempool.prototype._close = function close() {
 Mempool.prototype.addBlock = co(function* addBlock(block, txs) {
  var unlock = yield this.locker.lock();
  try {
-    return this._addBlock(block, txs);
+    return yield this._addBlock(block, txs);
  } finally {
    unlock();
  }
@ -142,7 +161,7 @@ Mempool.prototype.addBlock = co(function* addBlock(block, txs) {
 * @returns {Promise}
 */

-Mempool.prototype._addBlock = function addBlock(block, txs) {
+Mempool.prototype._addBlock = co(function* addBlock(block, txs) {
  var entries = [];
  var i, entry, tx, hash;

@ -172,13 +191,17 @@ Mempool.prototype._addBlock = function addBlock(block, txs) {
  // There may be a locktime in a TX that is now valid.
  this.rejects.reset();

+  this.cache.sync(block.hash);
+
+  yield this.cache.flush();
+
  if (entries.length === 0)
    return;

  this.logger.debug(
    'Removed %d txs from mempool for block %d.',
    entries.length, block.height);
-};
+});

 /**
 * Notify the mempool that a block has been disconnected
@ -220,7 +243,7 @@ Mempool.prototype._removeBlock = co(function* removeBlock(block, txs) {
      continue;

    try {
-      yield this._addTX(tx);
+      yield this.insertTX(tx);
      total++;
    } catch (e) {
      this.emit('error', e);
@ -232,6 +255,10 @@ Mempool.prototype._removeBlock = co(function* removeBlock(block, txs) {

  this.rejects.reset();

+  this.cache.sync(block.prevBlock);
+
+  yield this.cache.flush();
+
  if (total === 0)
    return;

@ -249,7 +276,7 @@ Mempool.prototype._removeBlock = co(function* removeBlock(block, txs) {
 Mempool.prototype.reset = co(function* reset() {
  var unlock = yield this.locker.lock();
  try {
-    return this._reset();
+    return yield this._reset();
  } finally {
    unlock();
  }
@ -260,7 +287,7 @@ Mempool.prototype.reset = co(function* reset() {
 * @private
 */

-Mempool.prototype._reset = function reset() {
+Mempool.prototype._reset = co(function* reset() {
  this.logger.info('Mempool reset (%d txs removed).', this.totalTX);

  this.size = 0;
@ -281,7 +308,12 @@ Mempool.prototype._reset = function reset() {
    this.fees.reset();

  this.rejects.reset();
-};
+
+  if (this.options.persistent) {
+    yield this.cache.wipe();
+    this.cache.clear();
+  }
+});

 /**
 * Ensure the size of the mempool stays below 300mb.
@ -620,12 +652,6 @@ Mempool.prototype.addTX = co(function* addTX(tx) {
  var unlock = yield this.locker.lock(hash);
  try {
    return yield this._addTX(tx);
-  } catch (err) {
-    if (err.type === 'VerifyError') {
-      if (!tx.hasWitness() && !err.malleated)
-        this.rejects.add(tx.hash());
-    }
-    throw err;
  } finally {
    unlock();
  }
@ -640,6 +666,33 @@ Mempool.prototype.addTX = co(function* addTX(tx) {
 */

 Mempool.prototype._addTX = co(function* _addTX(tx) {
+  var result;
+
+  try {
+    result = yield this.insertTX(tx);
+  } catch (err) {
+    if (err.type === 'VerifyError') {
+      if (!tx.hasWitness() && !err.malleated)
+        this.rejects.add(tx.hash());
+    }
+    throw err;
+  }
+
+  if (util.now() - this.lastUpdate > 10) {
+    yield this.cache.flush();
+    this.lastUpdate = util.now();
+  }
+});
+
+/**
+ * Add a transaction to the mempool without a lock.
+ * @method
+ * @private
+ * @param {TX} tx
+ * @returns {Promise}
+ */
+
+Mempool.prototype.insertTX = co(function* insertTX(tx) {
  var lockFlags = common.lockFlags.STANDARD_LOCKTIME_FLAGS;
  var hash = tx.hash('hex');
  var ret = new VerifyResult();
@ -987,6 +1040,8 @@ Mempool.prototype.addEntry = co(function* addEntry(entry, view) {

  this.logger.debug('Added tx %s to mempool.', tx.txid());

+  this.cache.save(entry);
+
  yield this.handleOrphans(tx);
 });

@ -1017,6 +1072,8 @@ Mempool.prototype.removeEntry = function removeEntry(entry, limit) {
  if (this.fees)
    this.fees.removeTX(hash);

+  this.cache.remove(tx.hash());
+
  this.emit('remove entry', entry);
 };

@ -1373,7 +1430,7 @@ Mempool.prototype.handleOrphans = co(function* handleOrphans(tx) {
    orphan = resolved[i];

    try {
-      yield this._addTX(orphan);
+      yield this.insertTX(orphan);
    } catch (err) {
      if (err.type === 'VerifyError') {
        this.logger.debug(
@ -1621,8 +1678,10 @@ Mempool.prototype.trackEntry = function trackEntry(entry, view) {
    this.spents[key] = entry;
  }

-  if (this.options.indexAddress)
+  if (this.options.indexAddress) {
+    assert(view, 'No view passed to trackEntry.');
    this.indexEntry(entry, view);
+  }

  this.size += this.memUsage(tx);
  this.totalTX++;
@ -1865,6 +1924,15 @@ function MempoolOptions(options) {
  this.expiryTime = policy.MEMPOOL_EXPIRY_TIME;
  this.minRelay = this.network.minRelay;

+  this.location = null;
+  this.db = 'memory';
+  this.maxFiles = 64;
+  this.cacheSize = 32 << 20;
+  this.compression = true;
+  this.bufferKeys = !util.isBrowser;
+
+  this.persistent = false;
+
  this.fromOptions(options);
 }

@ -1962,6 +2030,38 @@ MempoolOptions.prototype.fromOptions = function fromOptions(options) {
    this.minRelay = options.minRelay;
  }

+  if (options.location != null) {
+    assert(typeof options.location === 'string');
+    this.location = options.location;
+  }
+
+  if (options.db != null) {
+    assert(typeof options.db === 'string');
+    this.db = options.db;
+  }
+
+  if (options.maxFiles != null) {
+    assert(util.isNumber(options.maxFiles));
+    this.maxFiles = options.maxFiles;
+  }
+
+  if (options.cacheSize != null) {
+    assert(util.isNumber(options.cacheSize));
+    this.cacheSize = options.cacheSize;
+  }
+
+  if (options.compression != null) {
+    assert(typeof options.compression === 'boolean');
+    this.compression = options.compression;
+  }
+
+  if (options.persistent != null) {
+    assert(typeof options.persistent === 'boolean');
+    this.persistent = options.persistent;
+    assert(this.persistent !== this.indexAddress,
+      'Cannot have persistent mempool with address indexing.');
+  }
+
  return this;
 };

@ -2162,8 +2262,12 @@ CoinIndex.prototype.remove = function remove(outpoint) {
  delete this.map[key];
 };

-/*
- * Helpers
+/**
+ * Orphan
+ * @constructor
+ * @ignore
+ * @param {TX} tx
+ * @param {Hash[]} missing
 */

 function Orphan(tx, missing) {
@ -2175,6 +2279,185 @@ Orphan.prototype.toTX = function toTX() {
  return TX.fromRaw(this.raw);
 };

+/**
+ * Mempool Cache
+ * @ignore
+ * @constructor
+ * @param {Object} options
+ */
+
+function MempoolCache(options) {
+  if (!(this instanceof MempoolCache))
+    return new MempoolCache(options);
+
+  this.logger = options.logger;
+  this.chain = options.chain;
+  this.db = null;
+  this.batch = null;
+
+  if (options.persistent)
+    this.db = LDB(options);
+}
+
+MempoolCache.VERSION = 0;
+
+MempoolCache.prototype.getVersion = co(function* getVersion() {
+  var data = yield this.db.get(layout.V);
+
+  if (!data)
+    return -1;
+
+  return data.readUInt32LE(0, true);
+});
+
+MempoolCache.prototype.getTip = co(function* getTip() {
+  var hash = yield this.db.get(layout.R);
+
+  if (!hash)
+    return;
+
+  return hash.toString('hex');
+});
+
+MempoolCache.prototype.getEntries = function getEntries() {
+  return this.db.values({
+    gte: layout.e(encoding.ZERO_HASH),
+    lte: layout.e(encoding.MAX_HASH),
+    parse: MempoolEntry.fromRaw
+  });
+};
+
+MempoolCache.prototype.getKeys = function getKeys() {
+  return this.db.keys({
+    gte: layout.e(encoding.ZERO_HASH),
+    lte: layout.e(encoding.MAX_HASH)
+  });
+};
+
+MempoolCache.prototype.open = co(function* open() {
+  if (!this.db)
+    return;
+
+  yield this.db.open();
+
+  this.batch = this.db.batch();
+});
+
+MempoolCache.prototype.close = co(function* close() {
+  if (!this.db)
+    return;
+
+  yield this.db.close();
+
+  this.batch = null;
+});
+
+MempoolCache.prototype.save = function save(entry) {
+  if (!this.db)
+    return;
+
+  this.batch.put(layout.e(entry.tx.hash()), entry.toRaw());
+};
+
+MempoolCache.prototype.remove = function remove(hash) {
+  if (!this.db)
+    return;
+
+  this.batch.del(layout.e(hash));
+};
+
+MempoolCache.prototype.sync = function sync(hash) {
+  if (!this.db)
+    return;
+
+  this.batch.put(layout.R, new Buffer(hash, 'hex'));
+};
+
+MempoolCache.prototype.clear = function clear() {
+  this.batch.clear();
+  this.batch = this.db.batch();
+};
+
+MempoolCache.prototype.flush = co(function* flush() {
+  if (!this.db)
+    return;
+
+  yield this.batch.write();
+
+  this.batch = this.db.batch();
+});
+
+MempoolCache.prototype.init = co(function* init(hash) {
+  var batch = this.db.batch();
+  batch.put(layout.V, encoding.U32(MempoolCache.VERSION));
+  batch.put(layout.R, new Buffer(hash, 'hex'));
+  yield batch.write();
+});
+
+MempoolCache.prototype.load = co(function* load() {
+  var version = yield this.getVersion();
+  var i, tip, entries, entry;
+
+  if (version === -1) {
+    version = MempoolCache.VERSION;
+    tip = this.chain.tip.hash;
+
+    this.logger.info(
+      'Mempool cache is empty. Writing tip %s.',
+      util.revHex(tip));
+
+    yield this.init(tip);
+  }
+
+  if (version !== MempoolCache.VERSION) {
+    this.logger.warning(
+      'Mempool cache version mismatch (%d != %d)!',
+      version,
+      MempoolCache.VERSION);
+    this.logger.warning('Invalidating mempool cache.');
+    yield this.wipe();
+    return [];
+  }
+
+  tip = yield this.getTip();
+
+  if (tip !== this.chain.tip.hash) {
+    this.logger.warning(
+      'Mempool tip not consistent with chain tip (%s != %s)!',
+      util.revHex(tip),
+      this.chain.tip.rhash());
+    this.logger.warning('Invalidating mempool cache.');
+    yield this.wipe();
+    return [];
+  }
+
+  entries = yield this.getEntries();
+
+  this.logger.info(
+    'Loaded mempool from disk (%d entries).',
+    entries.length);
+
+  return entries;
+});
+
+MempoolCache.prototype.wipe = co(function* wipe() {
+  var batch = this.db.batch();
+  var keys = yield this.getKeys();
+  var i, key;
+
+  for (i = 0; i < keys.length; i++) {
+    key = keys[i];
+    batch.del(key);
+  }
+
+  batch.put(layout.V, encoding.U32(MempoolCache.VERSION));
+  batch.put(layout.R, new Buffer(this.chain.tip.hash, 'hex'));
+
+  yield batch.write();
+
+  this.logger.info('Removed %d mempool entries from disk.', keys.length);
+});
+
 /*
 * Expose
 */
--- a/lib/mempool/mempoolentry.js
+++ b/lib/mempool/mempoolentry.js
@ -195,7 +195,7 @@ MempoolEntry.prototype.isFree = function isFree(height) {
 */

 MempoolEntry.prototype.getSize = function getSize() {
-  return this.tx.getSize() + 49;
+  return this.tx.getSize() + 53;
 };

 /**
@ -210,7 +210,7 @@ MempoolEntry.prototype.toRaw = function toRaw() {
  bw.writeU32(this.size);
  bw.writeU32(this.sigops);
  bw.writeDouble(this.priority);
-  bw.writeU32(this.fee);
+  bw.writeU64(this.fee);
  bw.writeU32(this.ts);
  bw.writeU64(this.value);
  bw.writeU8(this.dependencies ? 1 : 0);
@ -233,7 +233,7 @@ MempoolEntry.prototype.fromRaw = function fromRaw(data) {
  this.size = br.readU32();
  this.sigops = br.readU32();
  this.priority = br.readDouble();
-  this.fee = br.readU32();
+  this.fee = br.readU64();
  this.ts = br.readU32();
  this.value = br.readU64();
  this.dependencies = br.readU8() === 1;
--- a/lib/node/config.js
+++ b/lib/node/config.js
@ -256,6 +256,7 @@ config.toOptions = function toOptions(data) {
  options.requireStandard = bool(data.requirestandard);
  options.rejectAbsurdFees = bool(data.rejectabsurdfees);
  options.replaceByFee = bool(data.replacebyfee);
+  options.persistentMempool = bool(data.persistentmempool);

  // Pool
  options.selfish = bool(data.selfish);
@ -282,6 +283,8 @@ config.toOptions = function toOptions(data) {
  options.payoutAddress = list(data.payoutaddress);
  options.coinbaseFlags = str(data.coinbaseflags);
  options.maxBlockWeight = num(data.maxblockweight);
+  options.reservedBlockWeight = num(data.reservedblockweight);
+  options.reservedBlockSigops = num(data.reservedblocksigops);

  // HTTP
  options.sslCert = file(data.sslcert, prefix);
--- a/lib/node/fullnode.js
+++ b/lib/node/fullnode.js
@ -75,6 +75,9 @@ function FullNode(options) {
    logger: this.logger,
    chain: this.chain,
    fees: this.fees,
+    db: this.options.db,
+    location: this.location('mempool'),
+    persistent: this.options.persistentMempool,
    maxSize: this.options.mempoolSize,
    limitFree: this.options.limitFree,
    limitFreeRelay: this.options.limitFreeRelay,
@ -120,7 +123,9 @@ function FullNode(options) {
    fees: this.fees,
    address: this.options.payoutAddress,
    coinbaseFlags: this.options.coinbaseFlags,
-    maxWeight: this.options.maxBlockWeight
+    maxWeight: this.options.maxBlockWeight,
+    reservedWeight: this.options.reservedBlockWeight,
+    reservedSigops: this.options.reservedBlockSigops
  });

  // Wallet database needs access to fees.
--- a/package.json
+++ b/package.json
@ -76,6 +76,7 @@
    "./lib/http/rpcclient": "./browser/empty.js",
    "./lib/http/server": "./browser/empty.js",
    "./lib/http/wallet": "./browser/empty.js",
+    "./lib/mempool/layout": "./lib/mempool/layout-browser.js",
    "./lib/net/dns": "./lib/net/dns-browser.js",
    "./lib/net/tcp": "./lib/net/tcp-browser.js",
    "./lib/utils/native": "./browser/empty.js",