Merge pull request #623 from LiskHQ/541-block-propagation

Improve block propagation - Closes #541

logic/broadcaster.js

@@ -5,6 +5,9 @@ var constants = require('../helpers/constants.js');
 var jobsQueue = require('../helpers/jobsQueue.js');
 var extend = require('extend');
 var _ = require('lodash');
+var BSON = require('bson');
+
+var bson = new BSON();
 
 // Private fields
 var modules, library, self, __private = {};
@@ -152,6 +155,9 @@ Broadcaster.prototype.broadcast = function (params, options, cb) {
 			}
 		},
 		function sendToPeer (peers, waterCb) {
+			if (options.data.block) {
+				options.data.block = bson.deserialize(options.data.block);
+			}
 			library.logger.debug('Begin broadcast', options);
 			if (params.limit === self.config.peerLimit) {
 				peers = peers.slice(0, self.config.broadcastLimit);

modules/blocks/chain.js

@@ -324,13 +324,12 @@ __private.applyTransaction = function (block, transaction, sender, cb) {
  * @method applyBlock
  * @emits  SIGTERM
  * @param  {Object}   block Full normalized block
- * @param  {boolean}  broadcast Indicator that block needs to be broadcasted
- * @param  {Function} cb Callback function
  * @param  {boolean}  saveBlock Indicator that block needs to be saved to database
+ * @param  {Function} cb Callback function
  * @return {Function} cb Callback function from params (through setImmediate)
  * @return {Object}   cb.err Error if occurred
  */
-Chain.prototype.applyBlock = function (block, broadcast, cb, saveBlock) {
+Chain.prototype.applyBlock = function (block, saveBlock, cb) {
 	// Prevent shutdown during database writes.
 	modules.blocks.isActive.set(true);
 
@@ -455,12 +454,10 @@ Chain.prototype.applyBlock = function (block, broadcast, cb, saveBlock) {
 					}
 
 					library.logger.debug('Block applied correctly with ' + block.transactions.length + ' transactions');
-					library.bus.message('newBlock', block, broadcast);
 
 					return seriesCb();
 				});
 			} else {
-				library.bus.message('newBlock', block, broadcast);
 				return seriesCb();
 			}
 		},
@@ -491,6 +488,15 @@ Chain.prototype.applyBlock = function (block, broadcast, cb, saveBlock) {
 	});
 };
 
+/**
+ * Broadcast reduced block to increase network performance.
+ * @param {Object} reducedBlock reduced block
+ * @param {boolean} broadcast Indicator that block needs to be broadcasted
+ */
+Chain.prototype.broadcastReducedBlock = function (reducedBlock, broadcast) {
+	library.bus.message('newBlock', reducedBlock, broadcast);
+	library.logger.debug(['reducedBlock', reducedBlock.id, 'broadcasted correctly'].join(' '));
+};
 
 /**
  * Deletes last block, undo transactions, recalculate round

modules/blocks/verify.js

@@ -7,6 +7,9 @@ var crypto = require('crypto');
 var slots = require('../../helpers/slots.js');
 var sql = require('../../sql/blocks.js');
 var exceptions = require('../../helpers/exceptions.js');
+var BSON = require('bson');
+
+var bson = new BSON();
 
 var modules, library, self, __private = {};
 
@@ -87,9 +90,77 @@ __private.checkTransaction = function (block, transaction, cb) {
 	});
 };
 
+/**
+ * Adds default properties to block.
+ * @private
+ * @param {Object} block Block object reduced
+ * @return {Object} Block object completed
+ */
+__private.addBlockProperties = function (block) {
+	if (block.version === undefined) {
+		block.version = 0;
+	}
+	if (block.numberOfTransactions === undefined) {
+		if (block.transactions === undefined) {
+			block.numberOfTransactions = 0;
+		} else {
+			block.numberOfTransactions = block.transactions.length;
+		}
+	}
+	if (block.totalAmount === undefined) {
+		block.totalAmount = 0;
+	}
+	if (block.totalFee === undefined) {
+		block.totalFee = 0;
+	}
+	if (block.payloadLength === undefined) {
+		block.payloadLength = 0;
+	}
+	if (block.reward === undefined) {
+		block.reward = 0;
+	}
+	if (block.transactions === undefined) {
+		block.transactions = [];
+	}
+	return block;
+};
+
+/**
+ * Deletes default properties from block.
+ * @private
+ * @param {Object} block Block object completed
+ * @return {Object} Block object reduced
+ */
+__private.deleteBlockProperties = function (block) {
+	var reducedBlock = JSON.parse(JSON.stringify(block));
+	if (reducedBlock.version === 0) {
+		delete reducedBlock.version;
+	}
+	// verifyBlock ensures numberOfTransactions is transactions.length
+	if (typeof(reducedBlock.numberOfTransactions) === 'number') {
+		delete reducedBlock.numberOfTransactions;
+	}
+	if (reducedBlock.totalAmount === 0) {
+		delete reducedBlock.totalAmount;
+	}
+	if (reducedBlock.totalFee === 0) {
+		delete reducedBlock.totalFee;
+	}
+	if (reducedBlock.payloadLength === 0) {
+		delete reducedBlock.payloadLength;
+	}
+	if (reducedBlock.reward === 0) {
+		delete reducedBlock.reward;
+	}
+	if (reducedBlock.transactions && reducedBlock.transactions.length === 0) {
+		delete reducedBlock.transactions;
+	}
+	delete reducedBlock.id;
+	return reducedBlock;
+};
+
 /**
  * Verify block and return all possible errors related to block
- * 
  * @public
  * @method verifyBlock
  * @param  {Object}  block Full block
@@ -266,6 +337,18 @@ Verify.prototype.processBlock = function (block, broadcast, cb, saveBlock) {
 	}
 
 	async.series({
+		addBlockProperties: function (seriesCb) {
+			if (!broadcast) {
+				try {
+					// set default properties
+					block = __private.addBlockProperties(block);
+				} catch (err) {
+					return setImmediate(seriesCb, err);
+				}
+			}
+
+			return setImmediate(seriesCb);
+		},
 		normalizeBlock: function (seriesCb) {
 			try {
 				block = library.logic.block.objectNormalize(block);
@@ -275,6 +358,20 @@ Verify.prototype.processBlock = function (block, broadcast, cb, saveBlock) {
 
 			return setImmediate(seriesCb);
 		},
+		deleteBlockProperties: function (seriesCb) {
+			if (broadcast) {
+				try {
+					// delete default properties
+					var blockReduced = __private.deleteBlockProperties(block);
+					var serializedBlockReduced = bson.serialize(blockReduced);
+					modules.blocks.chain.broadcastReducedBlock(serializedBlockReduced, broadcast);
+				} catch (err) {
+					return setImmediate(seriesCb, err);
+				}
+			}
+
+			return setImmediate(seriesCb);
+		},
 		verifyBlock: function (seriesCb) {
 			// Sanity check of the block, if values are coherent.
 			// No access to database
@@ -328,7 +425,7 @@ Verify.prototype.processBlock = function (block, broadcast, cb, saveBlock) {
 			// * Block and transactions have valid values (signatures, block slots, etc...)
 			// * The check against database state passed (for instance sender has enough LSK, votes are under 101, etc...)
 			// We thus update the database with the transactions values, save the block and tick it.
-			modules.blocks.chain.applyBlock(block, broadcast, cb, saveBlock);
+			modules.blocks.chain.applyBlock(block, saveBlock, cb);
 		}
 	});
 };

package.json

@@ -23,6 +23,7 @@
     "bignumber.js": "=4.0.2",
     "bluebird": "=3.5.0",
     "body-parser": "=1.17.2",
+    "bson": "=1.0.4",
     "bytebuffer": "=5.0.1",
     "change-case": "=3.0.1",
     "colors": "=1.1.2",

test/common/initModule.js

@@ -40,7 +40,16 @@ var modulesLoader = new function () {
 		schema: new z_schema(),
 		ed: ed,
 		bus: {
-			message: function () {}
+			argsMessages: [],
+			message: function () {
+				Array.prototype.push.apply(this.argsMessages, arguments);
+			},
+			getMessages: function () {
+				return this.argsMessages;
+			},
+			clearMessages: function () {
+				this.argsMessages = [];
+			}
 		},
 		nonce: randomString.generate(16),
 		dbSequence: new Sequence({