diff --git a/packages/caliper-core/lib/common/config/Config.js b/packages/caliper-core/lib/common/config/Config.js index 2a4011c98..dc3ce7add 100644 --- a/packages/caliper-core/lib/common/config/Config.js +++ b/packages/caliper-core/lib/common/config/Config.js @@ -91,7 +91,8 @@ const keys = { Communication: { Method: 'caliper-worker-communication-method', Address: 'caliper-worker-communication-address', - } + }, + MaxTxPromises: 'caliper-worker-maxtxpromises' }, Flow: { Skip: { diff --git a/packages/caliper-core/lib/common/config/default.yaml b/packages/caliper-core/lib/common/config/default.yaml index b4531f221..1b6605467 100644 --- a/packages/caliper-core/lib/common/config/default.yaml +++ b/packages/caliper-core/lib/common/config/default.yaml @@ -122,6 +122,7 @@ caliper: method: process # Address used for mqtt communications address: mqtt://localhost:1883 + maxtxpromises: 100 # Caliper flow options flow: # Skip options diff --git a/packages/caliper-core/lib/common/utils/circular-array.js b/packages/caliper-core/lib/common/utils/circular-array.js new file mode 100644 index 000000000..504ef24d5 --- /dev/null +++ b/packages/caliper-core/lib/common/utils/circular-array.js @@ -0,0 +1,46 @@ +/* +* Licensed under the Apache License, Version 2.0 (the "License"); +* you may not use this file except in compliance with the License. +* You may obtain a copy of the License at +* +* http://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, software +* distributed under the License is distributed on an "AS IS" BASIS, +* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +* See the License for the specific language governing permissions and +* limitations under the License. +*/ + +'use strict'; + +/** + * Create an Array that has a maximum length and will overwrite existing entries when additional items are added to the array + */ +class CircularArray extends Array { + + /** + * Constructor + * @param {number} maxLength maximum length of array + */ + constructor(maxLength) { + super(); + this.pointer = 0; + this.maxLength = maxLength; + } + + /** + * Add entry into array + * @param {any} element the element to add to the array + */ + add(element) { + if (this.length === this.maxLength) { + this[this.pointer] = element; + } else { + this.push(element); + } + this.pointer = (this.pointer + 1) % this.maxLength; + } +} + +module.exports = CircularArray; diff --git a/packages/caliper-core/lib/worker/client/caliper-local-client.js b/packages/caliper-core/lib/worker/client/caliper-local-client.js index c06956dc6..73298fd33 100644 --- a/packages/caliper-core/lib/worker/client/caliper-local-client.js +++ b/packages/caliper-core/lib/worker/client/caliper-local-client.js @@ -16,6 +16,7 @@ const Config = require('../../common/config/config-util.js'); const CaliperUtils = require('../../common/utils/caliper-utils.js'); +const CircularArray = require('../../common/utils/circular-array'); const bc = require('../../common/core/blockchain.js'); const RateControl = require('../rate-control/rateControl.js'); const PrometheusClient = require('../../common/prometheus/prometheus-push-client'); @@ -37,6 +38,8 @@ class CaliperLocalClient { this.clientIndex = clientIndex; this.messenger = messenger; this.context = undefined; + this.txUpdateTime = Config.get(Config.keys.TxUpdateTime, 5000); + this.maxTxPromises = Config.get(Config.keys.Worker.MaxTxPromises, 100); // Internal stats this.results = []; @@ -240,13 +243,13 @@ class CaliperLocalClient { Logger.info('Info: client ' + this.clientIndex + ' start test runFixedNumber()' + (cb.info ? (':' + cb.info) : '')); this.startTime = Date.now(); - let promises = []; - while(this.txNum < number) { + const circularArray = new CircularArray(this.maxTxPromises); + while (this.txNum < number) { // If this function calls cb.run() too quickly, micro task queue will be filled with unexecuted promises, // and I/O task(s) will get no chance to be execute and fall into starvation, for more detail info please visit: // https://snyk.io/blog/nodejs-how-even-quick-async-functions-can-block-the-event-loop-starve-io/ await this.setImmediatePromise(() => { - promises.push(cb.run().then((result) => { + circularArray.add(cb.run().then((result) => { this.addResult(result); return Promise.resolve(); })); @@ -254,7 +257,7 @@ class CaliperLocalClient { await rateController.applyRateControl(this.startTime, this.txNum, this.results, this.resultStats); } - await Promise.all(promises); + await Promise.all(circularArray); this.endTime = Date.now(); } @@ -269,13 +272,14 @@ class CaliperLocalClient { Logger.info('Info: client ' + this.clientIndex + ' start test runDuration()' + (cb.info ? (':' + cb.info) : '')); this.startTime = Date.now(); - let promises = []; + // Use a circular array of Promises so that the Promise.all() call does not exceed the maximum permissable Array size + const circularArray = new CircularArray(this.maxTxPromises); while ((Date.now() - this.startTime)/1000 < duration) { // If this function calls cb.run() too quickly, micro task queue will be filled with unexecuted promises, // and I/O task(s) will get no chance to be execute and fall into starvation, for more detail info please visit: // https://snyk.io/blog/nodejs-how-even-quick-async-functions-can-block-the-event-loop-starve-io/ await this.setImmediatePromise(() => { - promises.push(cb.run().then((result) => { + circularArray.add(cb.run().then((result) => { this.addResult(result); return Promise.resolve(); })); @@ -283,7 +287,7 @@ class CaliperLocalClient { await rateController.applyRateControl(this.startTime, this.txNum, this.results, this.resultStats); } - await Promise.all(promises); + await Promise.all(circularArray); this.endTime = Date.now(); } @@ -320,7 +324,6 @@ class CaliperLocalClient { Logger.debug('prepareTest() with:', test); let cb = require(CaliperUtils.resolvePath(test.cb)); - this.txUpdateTime = Config.get(Config.keys.TxUpdateTime, 5000); const self = this; let initUpdateInter = setInterval( () => { self.initUpdate(); } , self.txUpdateTime); @@ -374,7 +377,6 @@ class CaliperLocalClient { this.beforeTest(test); - this.txUpdateTime = Config.get(Config.keys.TxUpdateTime, 1000); Logger.info('txUpdateTime: ' + this.txUpdateTime); const self = this; let txUpdateInter = setInterval( () => { self.txUpdate(); } , self.txUpdateTime);