RanchiMall/fcoin
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mempool: rewrite.

Browse Source
This commit is contained in:
Christopher Jeffrey 2016-08-15 15:46:37 -07:00
parent c794c17077
commit c8bc9fb8b6
No known key found for this signature in database
GPG Key ID: 8962AB9DE6666BBD
9 changed files with 727 additions and 870 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
lib/bcoin/bip152.js
View File

@ -171,51 +171,42 @@ CompactBlock.prototype.toRequest = function toRequest() {
return TXRequest.fromCompact(this); return TXRequest.fromCompact(this);
}; };
CompactBlock.prototype.fillMempool = function fillMempool(mempool, callback) { CompactBlock.prototype.fillMempool = function fillMempool(mempool) {
var self = this;
var have = {}; var have = {};
var id, index; var hashes = mempool.getSnapshot();
var i, id, index, hash, tx;
mempool.getSnapshot(function(err, hashes) { for (i = 0; i < hashes.length; i++) {
if (err) hash = hashes[i];
return callback(err); id = this.sid(hash);
index = this.idMap[id];
utils.forEachSerial(hashes, function(hash, next) { if (index == null)
id = self.sid(hash); continue;
index = self.idMap[id];
if (index == null) if (have[index]) {
return next(); // Siphash collision, just request it.
this.available[index] = null;
this.count--;
continue;
}
if (have[index]) { tx = mempool.getTX(hash);
// Siphash collision, just request it.
self.available[index] = null;
self.count--;
return next();
}
mempool.getTX(hash, function(err, tx) { if (!tx)
if (err) continue;
return callback(err);
// Race condition: tx this.available[index] = tx;
// fell out of mempool. have[index] = true;
if (!tx) this.count++;
return next();
self.available[index] = tx; // We actually may have a siphash collision
have[index] = true; // here, but exit early anyway for perf.
self.count++; if (this.count === this.totalTX)
return true;
}
// We actually may have a siphash collision return false;
// here, but exit early anyway for perf.
if (self.count === self.totalTX)
return callback(null, true);
next();
});
}, callback);
});
}; };
CompactBlock.prototype.fillMissing = function fillMissing(res) { CompactBlock.prototype.fillMissing = function fillMissing(res) {

117
lib/bcoin/fullnode.js
View File

@ -483,32 +483,15 @@ Fullnode.prototype.getFullBlock = function getFullBlock(hash, callback) {
*/ */
Fullnode.prototype.getCoin = function getCoin(hash, index, callback) { Fullnode.prototype.getCoin = function getCoin(hash, index, callback) {
var self = this; var coin = this.mempool.getCoin(hash, index);
this.mempool.getCoin(hash, index, function(err, coin) {
if (err)
return callback(err);
if (coin) if (coin)
return callback(null, coin); return callback(null, coin);
self.chain.db.getCoin(hash, index, function(err, coin) { if (this.mempool.isSpent(hash, index))
if (err) return callback();
return callback(err);
if (!coin) this.chain.db.getCoin(hash, index, callback);
return callback();
self.mempool.isSpent(hash, index, function(err, spent) {
if (err)
return callback(err);
if (spent)
return callback();
return callback(null, coin);
});
});
});
}; };
/** /**
@ -520,29 +503,23 @@ Fullnode.prototype.getCoin = function getCoin(hash, index, callback) {
Fullnode.prototype.getCoinsByAddress = function getCoinsByAddress(addresses, callback) { Fullnode.prototype.getCoinsByAddress = function getCoinsByAddress(addresses, callback) {
var self = this; var self = this;
this.mempool.getCoinsByAddress(addresses, function(err, coins) { var coins = this.mempool.getCoinsByAddress(addresses);
this.chain.db.getCoinsByAddress(addresses, function(err, blockCoins) {
if (err) if (err)
return callback(err); return callback(err);
self.chain.db.getCoinsByAddress(addresses, function(err, blockCoins) { utils.forEach(blockCoins, function(coin, next) {
var spent = self.mempool.isSpent(coin.hash, coin.index);
if (!spent)
coins.push(coin);
return next();
}, function(err) {
if (err) if (err)
return callback(err); return callback(err);
return callback(null, coins);
utils.forEach(blockCoins, function(coin, next) {
self.mempool.isSpent(coin.hash, coin.index, function(err, spent) {
if (err)
return callback(err);
if (!spent)
coins.push(coin);
return next();
});
}, function(err) {
if (err)
return callback(err);
return callback(null, coins);
});
}); });
}); });
}; };
@ -554,25 +531,12 @@ Fullnode.prototype.getCoinsByAddress = function getCoinsByAddress(addresses, cal
*/ */
Fullnode.prototype.getTX = function getTX(hash, callback) { Fullnode.prototype.getTX = function getTX(hash, callback) {
var self = this; var tx = this.mempool.getTX(hash);
this.mempool.getTX(hash, function(err, tx) { if (tx)
if (err) return callback(null, tx);
return callback(err);
if (tx) this.chain.db.getTX(hash, callback);
return callback(null, tx);
self.chain.db.getTX(hash, function(err, tx) {
if (err)
return callback(err);
if (!tx)
return callback();
return callback(null, tx);
});
});
}; };
/** /**
@ -582,17 +546,10 @@ Fullnode.prototype.getTX = function getTX(hash, callback) {
*/ */
Fullnode.prototype.hasTX = function hasTX(hash, callback) { Fullnode.prototype.hasTX = function hasTX(hash, callback) {
var self = this; if (this.mempool.hasTX(hash))
return callback(null, true);
this.mempool.hasTX(hash, function(err, result) { this.chain.db.hasTX(hash, callback);
if (err)
return callback(err);
if (result)
return callback(null, true);
self.chain.db.hasTX(hash, callback);
});
}; };
/** /**
@ -603,17 +560,10 @@ Fullnode.prototype.hasTX = function hasTX(hash, callback) {
*/ */
Fullnode.prototype.isSpent = function isSpent(hash, index, callback) { Fullnode.prototype.isSpent = function isSpent(hash, index, callback) {
var self = this; if (this.mempool.isSpent(hash, index))
return callback(null, true);
this.mempool.isSpent(hash, index, function(err, spent) { this.chain.db.isSpent(hash, index, callback);
if (err)
return callback(err);
if (spent)
return callback(null, true);
self.chain.db.isSpent(hash, index, callback);
});
}; };
/** /**
@ -624,18 +574,13 @@ Fullnode.prototype.isSpent = function isSpent(hash, index, callback) {
*/ */
Fullnode.prototype.getTXByAddress = function getTXByAddress(addresses, callback) { Fullnode.prototype.getTXByAddress = function getTXByAddress(addresses, callback) {
var self = this; var mempool = this.mempool.getTXByAddress(addresses);
this.mempool.getTXByAddress(addresses, function(err, mempool) { this.chain.db.getTXByAddress(addresses, function(err, txs) {
if (err) if (err)
return callback(err); return callback(err);
self.chain.db.getTXByAddress(addresses, function(err, txs) { return callback(null, mempool.concat(txs));
if (err)
return callback(err);
return callback(null, mempool.concat(txs));
});
}); });
}; };

102
lib/bcoin/http/rpc.js
View File

@ -1094,50 +1094,41 @@ RPC.prototype.getrawmempool = function getrawmempool(args, callback) {
RPC.prototype.mempoolToJSON = function mempoolToJSON(verbose, callback) { RPC.prototype.mempoolToJSON = function mempoolToJSON(verbose, callback) {
var self = this; var self = this;
var out = {}; var out = {};
var tx; var i, tx, hashes, hash, entry;
if (verbose) { if (verbose) {
return this.mempool.getSnapshot(function(err, hashes) { hashes = this.mempool.getSnapshot();
if (err)
return callback(err);
utils.forEachSerial(hashes, function(hash, next) { for (i = 0; i < hashes.length; i++) {
self.mempool.getEntry(hash, function(err, entry) { hash = hashes[i];
if (err) entry = this.mempool.getEntry(hash);
return callback(err);
tx = entry.tx; if (!entry)
continue;
out[tx.rhash] = { tx = entry.tx;
size: entry.size,
fee: entry.fee,
modifiedfee: entry.fees,
time: entry.ts,
height: entry.height,
startingpriority: entry.priority,
currentpriority: entry.getPriority(self.chain.height),
descendantcount: 0,
descendantsize: entry.size,
descendantfees: entry.fees,
depends: []
};
next(); out[tx.rhash] = {
}); size: entry.size,
}, function(err) { fee: entry.fee,
if (err) modifiedfee: entry.fees,
return callback(err); time: entry.ts,
return callback(null, out); height: entry.height,
}); startingpriority: entry.priority,
}); currentpriority: entry.getPriority(self.chain.height),
descendantcount: 0,
descendantsize: entry.size,
descendantfees: entry.fees,
depends: []
};
}
return callback(null, out);
} }
this.mempool.getSnapshot(function(err, hashes) { hashes = this.mempool.getSnapshot();
if (err)
return callback(err);
return callback(null, hashes.map(utils.revHex)); return callback(null, hashes.map(utils.revHex));
});
}; };
RPC.prototype.gettxout = function gettxout(args, callback) { RPC.prototype.gettxout = function gettxout(args, callback) {
@ -1573,7 +1564,7 @@ RPC.prototype.getnetworkhashps = function getnetworkhashps(args, callback) {
RPC.prototype.prioritisetransaction = function prioritisetransaction(args, callback) { RPC.prototype.prioritisetransaction = function prioritisetransaction(args, callback) {
var self = this; var self = this;
var hash, pri, fee; var hash, pri, fee, entry;
if (args.help || args.length !== 3) { if (args.help || args.length !== 3) {
return callback(new RPCError('prioritisetransaction' return callback(new RPCError('prioritisetransaction'
@ -1592,35 +1583,32 @@ RPC.prototype.prioritisetransaction = function prioritisetransaction(args, callb
hash = utils.revHex(hash); hash = utils.revHex(hash);
this.mempool.getEntry(hash, function(err, entry) { entry = this.mempool.getEntry(hash);
if (!entry)
return callback(new RPCError('Transaction not in mempool.'));
entry.priority += pri;
entry.fees += fee;
if (entry.priority < 0)
entry.priority = 0;
if (entry.fees < 0)
entry.fees = 0;
this.mempool.fillAllCoins(entry.tx, function(err) {
if (err) if (err)
return callback(err); return callback(err);
if (!entry) if (!entry.tx.hasCoins())
return callback(new RPCError('Transaction not in mempool.')); return callback(new RPCError('Transaction not in mempool.'));
entry.priority += pri; self.mempool.addUnchecked(entry, function(err) {
entry.fees += fee;
if (entry.priority < 0)
entry.priority = 0;
if (entry.fees < 0)
entry.fees = 0;
this.mempool.fillAllCoins(entry.tx, function(err) {
if (err) if (err)
return callback(err); return callback(err);
if (!entry.tx.hasCoins()) callback(null, true);
return callback(new RPCError('Transaction not in mempool.'));
self.mempool.addUnchecked(entry, function(err) {
if (err)
return callback(err);
callback(null, true);
});
}); });
}); });
}; };

1145
lib/bcoin/mempool.js
View File

File diff suppressed because it is too large Load Diff

60
lib/bcoin/peer.js
View File

@ -1124,7 +1124,7 @@ Peer.prototype._handleGetUTXOs = function _handleGetUTXOs(payload) {
if (!payload.mempool) if (!payload.mempool)
return callback(); return callback();
self.mempool.getCoin(hash, index, callback); callback(null, self.mempool.getCoin(hash, index));
} }
function isSpent(hash, index, callback) { function isSpent(hash, index, callback) {
@ -1134,7 +1134,7 @@ Peer.prototype._handleGetUTXOs = function _handleGetUTXOs(payload) {
if (!payload.mempool) if (!payload.mempool)
return callback(null, false); return callback(null, false);
self.mempool.isSpent(hash, index, callback); callback(null, self.mempool.isSpent(hash, index));
} }
if (payload.prevout.length > 15) if (payload.prevout.length > 15)
@ -1427,7 +1427,7 @@ Peer.prototype._handleVersion = function _handleVersion(version) {
Peer.prototype._handleMempool = function _handleMempool() { Peer.prototype._handleMempool = function _handleMempool() {
var self = this; var self = this;
var items = []; var items = [];
var i; var i, hashes;
var unlock = this.locker.lock(_handleMempool, []); var unlock = this.locker.lock(_handleMempool, []);
if (!unlock) if (!unlock)
@ -1450,18 +1450,14 @@ Peer.prototype._handleMempool = function _handleMempool() {
if (this.pool.options.selfish) if (this.pool.options.selfish)
return done(); return done();
this.mempool.getSnapshot(function(err, hashes) { hashes = this.mempool.getSnapshot();
if (err)
return done(err);
for (i = 0; i < hashes.length; i++) for (i = 0; i < hashes.length; i++)
items.push(new InvItem(constants.inv.TX, hashes[i])); items.push(new InvItem(constants.inv.TX, hashes[i]));
self.logger.debug('Sending mempool snapshot (%s).', self.hostname); self.logger.debug('Sending mempool snapshot (%s).', self.hostname);
self.sendInv(items); self.sendInv(items);
done();
});
}; };
/** /**
@ -1494,7 +1490,7 @@ Peer.prototype._getItem = function _getItem(item, callback) {
if (item.isTX()) { if (item.isTX()) {
if (!this.mempool) if (!this.mempool)
return callback(); return callback();
return this.mempool.getEntry(item.hash, callback); return callback(null, this.mempool.getEntry(item.hash));
} }
if (this.chain.db.options.spv) if (this.chain.db.options.spv)
@ -1941,6 +1937,7 @@ Peer.prototype._handleSendCmpct = function _handleSendCmpct(payload) {
Peer.prototype._handleCmpctBlock = function _handleCmpctBlock(block) { Peer.prototype._handleCmpctBlock = function _handleCmpctBlock(block) {
var self = this; var self = this;
var hash = block.hash('hex'); var hash = block.hash('hex');
var result;
function done(err) { function done(err) {
if (err) { if (err) {
@ -1970,31 +1967,28 @@ Peer.prototype._handleCmpctBlock = function _handleCmpctBlock(block) {
// Sort of a lock too. // Sort of a lock too.
this.compactBlocks[hash] = block; this.compactBlocks[hash] = block;
block.fillMempool(this.mempool, function(err, result) { result = block.fillMempool(this.mempool);
if (err)
return done(err);
if (result) { if (result) {
delete self.compactBlocks[hash]; delete this.compactBlocks[hash];
self.emit('block', block.toBlock()); this.fire('block', block.toBlock());
self.logger.debug( this.logger.debug(
'Received full compact block %s (%s).', 'Received full compact block %s (%s).',
block.rhash, self.hostname); block.rhash, this.hostname);
return; return;
} }
self.write(self.framer.getBlockTxn(block.toRequest())); this.write(this.framer.getBlockTxn(block.toRequest()));
this.logger.debug(
'Received semi-full compact block %s (%s).',
block.rhash, this.hostname);
block.startTimeout(10000, function() {
self.logger.debug( self.logger.debug(
'Received semi-full compact block %s (%s).', 'Compact block timed out: %s (%s).',
block.rhash, self.hostname); block.rhash, self.hostname);
delete self.compactBlocks[hash];
block.startTimeout(10000, function() {
self.logger.debug(
'Compact block timed out: %s (%s).',
block.rhash, self.hostname);
delete self.compactBlocks[hash];
});
}); });
}; };

2
lib/bcoin/pool.js
View File

@ -1563,7 +1563,7 @@ Pool.prototype.has = function has(type, hash, force, callback) {
} }
// Check the mempool. // Check the mempool.
return this.mempool.has(hash, check); return check(null, this.mempool.has(hash));
} }
// Check the chain. // Check the chain.

3
lib/bcoin/rbt.js
View File

@ -37,6 +37,9 @@ function RBT(location, options) {
if (!options) if (!options)
options = {}; options = {};
if (typeof options === 'function')
options = { compare: options };
this.options = options; this.options = options;
this.root = SENTINEL; this.root = SENTINEL;
this.compare = options.compare || utils.cmp; this.compare = options.compare || utils.cmp;

56
test/block-test.js
View File

@ -224,23 +224,21 @@ describe('Block', function() {
var fakeMempool = { var fakeMempool = {
getSnapshot: function(callback) { getSnapshot: function(callback) {
callback(null, Object.keys(map)); return Object.keys(map);
}, },
getTX: function(hash, callback) { getTX: function(hash, callback) {
callback(null, map[hash]); return map[hash];
} }
}; };
assert.equal(cblock.sid(block.txs[1].hash()), 125673511480291); assert.equal(cblock.sid(block.txs[1].hash()), 125673511480291);
cblock.fillMempool(fakeMempool, function(err, result) { var result = cblock.fillMempool(fakeMempool);
assert.ifError(err); assert(result);
assert(result); for (var i = 0; i < cblock.available.length; i++)
for (var i = 0; i < cblock.available.length; i++) assert(cblock.available[i]);
assert(cblock.available[i]); assert.equal(cblock.toBlock().toRaw().toString('hex'), block.toRaw().toString('hex'));
assert.equal(cblock.toBlock().toRaw().toString('hex'), block.toRaw().toString('hex')); cb();
cb();
});
}); });
it('should handle half-full compact block', function(cb) { it('should handle half-full compact block', function(cb) {
@ -262,39 +260,37 @@ describe('Block', function() {
var fakeMempool = { var fakeMempool = {
getSnapshot: function(callback) { getSnapshot: function(callback) {
callback(null, keys); return keys;
}, },
getTX: function(hash, callback) { getTX: function(hash, callback) {
callback(null, map[hash]); return map[hash];
} }
}; };
assert.equal(cblock.sid(block.txs[1].hash()), 125673511480291); assert.equal(cblock.sid(block.txs[1].hash()), 125673511480291);
cblock.fillMempool(fakeMempool, function(err, result) { var result = cblock.fillMempool(fakeMempool);
assert.ifError(err); assert(!result);
assert(!result);
var req = cblock.toRequest(); var req = cblock.toRequest();
assert.equal(req.hash, cblock.hash('hex')); assert.equal(req.hash, cblock.hash('hex'));
assert.deepEqual(req.indexes, [5, 6, 7, 8, 9]); assert.deepEqual(req.indexes, [5, 6, 7, 8, 9]);
req = bip152.TXRequest.fromRaw(req.toRaw()); req = bip152.TXRequest.fromRaw(req.toRaw());
assert.equal(req.hash, cblock.hash('hex')); assert.equal(req.hash, cblock.hash('hex'));
assert.deepEqual(req.indexes, [5, 6, 7, 8, 9]); assert.deepEqual(req.indexes, [5, 6, 7, 8, 9]);
var res = bip152.TXResponse.fromBlock(block, req); var res = bip152.TXResponse.fromBlock(block, req);
res = bip152.TXResponse.fromRaw(res.toRaw()); res = bip152.TXResponse.fromRaw(res.toRaw());
var result = cblock.fillMissing(res); var result = cblock.fillMissing(res);
assert(result); assert(result);
for (var i = 0; i < cblock.available.length; i++) for (var i = 0; i < cblock.available.length; i++)
assert(cblock.available[i]); assert(cblock.available[i]);
assert.equal(cblock.toBlock().toRaw().toString('hex'), block.toRaw().toString('hex')); assert.equal(cblock.toBlock().toRaw().toString('hex'), block.toRaw().toString('hex'));
cb(); cb();
});
}); });
}); });

49
test/mempool-test.js
View File

@ -126,41 +126,30 @@ describe('Mempool', function() {
assert.ifError(err); assert.ifError(err);
mempool.addTX(t4, function(err) { mempool.addTX(t4, function(err) {
assert.ifError(err); assert.ifError(err);
mempool.getBalance(function(err, balance) { var balance = mempool.getBalance();
assert.equal(balance, 0);
mempool.addTX(t1, function(err) {
assert.ifError(err); assert.ifError(err);
assert.equal(balance.total, 0); var balance = mempool.getBalance();
mempool.addTX(t1, function(err) { assert.equal(balance, 60000);
mempool.addTX(t2, function(err) {
assert.ifError(err); assert.ifError(err);
mempool.getBalance(function(err, balance) { var balance = mempool.getBalance();
assert.equal(balance, 50000);
mempool.addTX(t3, function(err) {
assert.ifError(err); assert.ifError(err);
assert.equal(balance.total, 60000); var balance = mempool.getBalance();
mempool.addTX(t2, function(err) { assert.equal(balance, 22000);
mempool.addTX(f1, function(err) {
assert.ifError(err); assert.ifError(err);
mempool.getBalance(function(err, balance) { var balance = mempool.getBalance();
assert.ifError(err); assert.equal(balance, 20000);
assert.equal(balance.total, 50000); var txs = mempool.getHistory();
mempool.addTX(t3, function(err) { assert(txs.some(function(tx) {
assert.ifError(err); return tx.hash('hex') === f1.hash('hex');
mempool.getBalance(function(err, balance) { }));
assert.ifError(err);
assert.equal(balance.total, 22000);
mempool.addTX(f1, function(err) {
assert.ifError(err);
mempool.getBalance(function(err, balance) {
assert.ifError(err);
assert.equal(balance.total, 20000);
mempool.getHistory(function(err, txs) {
assert(txs.some(function(tx) {
return tx.hash('hex') === f1.hash('hex');
}));
cb(); cb();
});
});
});
});
});
});
}); });
}); });
}); });