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

peer: refactor.

Browse Source
This commit is contained in:
Christopher Jeffrey 2016-12-16 15:03:39 -08:00
parent 4269d16fee
commit 8ccefb8e71
No known key found for this signature in database
GPG Key ID: 8962AB9DE6666BBD
2 changed files with 48 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
lib/net/packets.js
View File

@ -1147,6 +1147,8 @@ InvPacket.prototype.getSize = function getSize() {
InvPacket.prototype.toWriter = function toWriter(bw) { InvPacket.prototype.toWriter = function toWriter(bw) {
var i, item; var i, item;
assert(this.items.length <= 50000);
bw.writeVarint(this.items.length); bw.writeVarint(this.items.length);
for (i = 0; i < this.items.length; i++) { for (i = 0; i < this.items.length; i++) {
@ -1178,6 +1180,8 @@ InvPacket.prototype.fromReader = function fromReader(br) {
count = br.readVarint(); count = br.readVarint();
assert(count <= 50000, 'Inv item count too high.');
for (i = 0; i < count; i++) for (i = 0; i < count; i++)
this.items.push(InvItem.fromReader(br)); this.items.push(InvItem.fromReader(br));

82
lib/net/peer.js
View File

@ -189,7 +189,7 @@ Peer.prototype._init = function init() {
this.parser.on('packet', co(function* (packet) { this.parser.on('packet', co(function* (packet) {
try { try {
yield self._onPacket(packet); yield self.handlePacket(packet);
} catch (e) { } catch (e) {
self.destroy(); self.destroy();
self.error(e); self.error(e);
@ -330,19 +330,16 @@ Peer.prototype.connect = function connect(port, host) {
*/ */
Peer.prototype.open = co(function* open() { Peer.prototype.open = co(function* open() {
yield this._wait(); yield this.initConnect();
yield this._stallify(); yield this.initStall();
yield this._bip151(); yield this.initBIP151();
yield this._bip150(); yield this.initBIP150();
yield this._handshake(); yield this.initVerack();
yield this._finalize(); yield this.finalize();
if (this.destroyed) if (this.destroyed)
throw new Error('Peer was destroyed.'); throw new Error('Peer was destroyed.');
clearTimeout(this.connectTimeout);
this.connectTimeout = null;
// Finally we can let the pool know // Finally we can let the pool know
// that this peer is ready to go. // that this peer is ready to go.
this.emit('open'); this.emit('open');
@ -367,7 +364,7 @@ Peer.prototype.tryOpen = co(function* tryOpen() {
* @private * @private
*/ */
Peer.prototype._wait = function _wait() { Peer.prototype.initConnect = function initConnect() {
var self = this; var self = this;
if (this.connected) { if (this.connected) {
@ -397,7 +394,7 @@ Peer.prototype._wait = function _wait() {
* @private * @private
*/ */
Peer.prototype._stallify = function _stallify() { Peer.prototype.initStall = function initStall() {
var self = this; var self = this;
assert(!this.stallTimer); assert(!this.stallTimer);
this.stallTimer = setInterval(function() { this.stallTimer = setInterval(function() {
@ -412,7 +409,7 @@ Peer.prototype._stallify = function _stallify() {
* @private * @private
*/ */
Peer.prototype._bip151 = co(function* _bip151() { Peer.prototype.initBIP151 = co(function* initBIP151() {
// Send encinit. Wait for handshake to complete. // Send encinit. Wait for handshake to complete.
if (!this.bip151) if (!this.bip151)
return; return;
@ -442,7 +439,7 @@ Peer.prototype._bip151 = co(function* _bip151() {
* @private * @private
*/ */
Peer.prototype._bip150 = co(function* _bip150() { Peer.prototype.initBIP150 = co(function* initBIP150() {
if (!this.bip151 || !this.bip150) if (!this.bip151 || !this.bip150)
return; return;
@ -475,7 +472,7 @@ Peer.prototype._bip150 = co(function* _bip150() {
* @private * @private
*/ */
Peer.prototype._handshake = co(function* _handshake() { Peer.prototype.initVerack = co(function* initVerack() {
// Say hello. // Say hello.
this.sendVersion(); this.sendVersion();
@ -509,7 +506,7 @@ Peer.prototype._handshake = co(function* _handshake() {
* @private * @private
*/ */
Peer.prototype._finalize = co(function* _finalize() { Peer.prototype.finalize = co(function* finalize() {
var self = this; var self = this;
// Setup the ping interval. // Setup the ping interval.
@ -591,7 +588,7 @@ Peer.prototype.announceBlock = function announceBlock(blocks) {
// they're using compact block mode 1. // they're using compact block mode 1.
if (this.compactMode && this.compactMode.mode === 1) { if (this.compactMode && this.compactMode.mode === 1) {
this.invFilter.add(block.hash()); this.invFilter.add(block.hash());
this._sendCompactBlock(block, this.compactWitness); this.sendCompactBlock(block, this.compactWitness);
continue; continue;
} }
@ -1083,24 +1080,29 @@ Peer.prototype.getData = function getData(items) {
* @param {Packet} packet * @param {Packet} packet
*/ */
Peer.prototype._onPacket = co(function* onPacket(packet) { Peer.prototype.handlePacket = co(function* handlePacket(packet) {
var unlock; var unlock;
// We stop reads and lock the peer for any
// packet with significant IO/asynchronocity.
switch (packet.type) { switch (packet.type) {
case packetTypes.VERSION: case packetTypes.GETDATA:
case packetTypes.CMPCTBLOCK: case packetTypes.GETBLOCKS:
// These can't have locks or stop reads. case packetTypes.GETHEADERS:
return yield this.__onPacket(packet); case packetTypes.GETUTXOS:
default:
unlock = yield this.locker.lock(); unlock = yield this.locker.lock();
this.socket.pause(); this.socket.pause();
try { try {
return yield this.__onPacket(packet); return yield this.onPacket(packet);
} finally { } finally {
this.socket.resume(); this.socket.resume();
unlock(); unlock();
} }
break; break;
default:
return yield this.onPacket(packet);
} }
}); });
@ -1110,7 +1112,7 @@ Peer.prototype._onPacket = co(function* onPacket(packet) {
* @param {Packet} packet * @param {Packet} packet
*/ */
Peer.prototype.__onPacket = co(function* onPacket(packet) { Peer.prototype.onPacket = co(function* onPacket(packet) {
this.lastRecv = util.ms(); this.lastRecv = util.ms();
if (this.destroyed) if (this.destroyed)
@ -1611,11 +1613,12 @@ Peer.prototype.handleMempool = co(function* handleMempool(packet) {
/** /**
* Get a block/tx from the broadcast map. * Get a block/tx from the broadcast map.
* @private
* @param {InvItem} item * @param {InvItem} item
* @returns {Promise} * @returns {Promise}
*/ */
Peer.prototype._getBroadcasted = function _getBroadcasted(item) { Peer.prototype.getBroadcasted = function getBroadcasted(item) {
var entry = this.pool.invMap[item.hash]; var entry = this.pool.invMap[item.hash];
if (!entry) if (!entry)
@ -1646,12 +1649,13 @@ Peer.prototype._getBroadcasted = function _getBroadcasted(item) {
/** /**
* Get a block/tx either from the broadcast map, mempool, or blockchain. * Get a block/tx either from the broadcast map, mempool, or blockchain.
* @private
* @param {InvItem} item * @param {InvItem} item
* @returns {Promise} * @returns {Promise}
*/ */
Peer.prototype._getItem = co(function* _getItem(item) { Peer.prototype.getItem = co(function* getItem(item) {
var entry = this._getBroadcasted(item); var entry = this.getBroadcasted(item);
if (entry) if (entry)
return entry; return entry;
@ -1676,12 +1680,13 @@ Peer.prototype._getItem = co(function* _getItem(item) {
/** /**
* Send a block from the broadcast list or chain. * Send a block from the broadcast list or chain.
* @private
* @param {InvItem} item * @param {InvItem} item
* @returns {Boolean} * @returns {Boolean}
*/ */
Peer.prototype._sendBlock = co(function* _sendBlock(item, witness) { Peer.prototype.sendBlock = co(function* sendBlock(item, witness) {
var block = this._getBroadcasted(item); var block = this.getBroadcasted(item);
// Check for a broadcasted item first. // Check for a broadcasted item first.
if (block) { if (block) {
@ -1720,12 +1725,13 @@ Peer.prototype._sendBlock = co(function* _sendBlock(item, witness) {
/** /**
* Send a compact block. * Send a compact block.
* @private
* @param {Block} block * @param {Block} block
* @param {Boolean} witness * @param {Boolean} witness
* @returns {Boolean} * @returns {Boolean}
*/ */
Peer.prototype._sendCompactBlock = function _sendCompactBlock(block, witness) { Peer.prototype.sendCompactBlock = function sendCompactBlock(block, witness) {
// Try again with a new nonce // Try again with a new nonce
// if we get a siphash collision. // if we get a siphash collision.
for (;;) { for (;;) {
@ -1761,7 +1767,7 @@ Peer.prototype.handleGetData = co(function* handleGetData(packet) {
item = items[i]; item = items[i];
if (item.isTX()) { if (item.isTX()) {
tx = yield this._getItem(item); tx = yield this.getItem(item);
if (!tx) { if (!tx) {
notFound.push(item); notFound.push(item);
@ -1788,7 +1794,7 @@ Peer.prototype.handleGetData = co(function* handleGetData(packet) {
switch (item.type) { switch (item.type) {
case constants.inv.BLOCK: case constants.inv.BLOCK:
case constants.inv.WITNESS_BLOCK: case constants.inv.WITNESS_BLOCK:
result = yield this._sendBlock(item, item.hasWitness()); result = yield this.sendBlock(item, item.hasWitness());
if (!result) { if (!result) {
notFound.push(item); notFound.push(item);
continue; continue;
@ -1802,7 +1808,7 @@ Peer.prototype.handleGetData = co(function* handleGetData(packet) {
continue; continue;
} }
block = yield this._getItem(item); block = yield this.getItem(item);
if (!block) { if (!block) {
notFound.push(item); notFound.push(item);
@ -1826,7 +1832,7 @@ Peer.prototype.handleGetData = co(function* handleGetData(packet) {
// Fallback to full block. // Fallback to full block.
height = yield this.chain.db.getHeight(item.hash); height = yield this.chain.db.getHeight(item.hash);
if (height < this.chain.tip.height - 10) { if (height < this.chain.tip.height - 10) {
result = yield this._sendBlock(item, this.compactWitness); result = yield this.sendBlock(item, this.compactWitness);
if (!result) { if (!result) {
notFound.push(item); notFound.push(item);
continue; continue;
@ -1835,14 +1841,14 @@ Peer.prototype.handleGetData = co(function* handleGetData(packet) {
break; break;
} }
block = yield this._getItem(item); block = yield this.getItem(item);
if (!block) { if (!block) {
notFound.push(item); notFound.push(item);
continue; continue;
} }
yield this._sendCompactBlock(block, this.compactWitness); yield this.sendCompactBlock(block, this.compactWitness);
blocks++; blocks++;
@ -2387,7 +2393,7 @@ Peer.prototype.handleGetBlockTxn = co(function* handleGetBlockTxn(packet) {
item = new InvItem(constants.inv.BLOCK, req.hash); item = new InvItem(constants.inv.BLOCK, req.hash);
block = yield this._getItem(item); block = yield this.getItem(item);
if (!block) { if (!block) {
this.logger.debug( this.logger.debug(