{"action":"created","issue":{"url":"https://api.github.com/repos/hyperledger/caliper/issues/124","repository_url":"https://api.github.com/repos/hyperledger/caliper","labels_url":"https://api.github.com/repos/hyperledger/caliper/issues/124/labels{/name}","comments_url":"https://api.github.com/repos/hyperledger/caliper/issues/124/comments","events_url":"https://api.github.com/repos/hyperledger/caliper/issues/124/events","html_url":"https://github.com/hyperledger/caliper/pull/124","id":340508571,"node_id":"MDExOlB1bGxSZXF1ZXN0MjAwOTAyMzEx","number":124,"title":"Split caliper into transaction sender and event listener","user":{"login":"nitesh7sid","id":18530954,"node_id":"MDQ6VXNlcjE4NTMwOTU0","avatar_url":"https://avatars3.githubusercontent.com/u/18530954?v=4","gravatar_id":"","url":"https://api.github.com/users/nitesh7sid","html_url":"https://github.com/nitesh7sid","followers_url":"https://api.github.com/users/nitesh7sid/followers","following_url":"https://api.github.com/users/nitesh7sid/following{/other_user}","gists_url":"https://api.github.com/users/nitesh7sid/gists{/gist_id}","starred_url":"https://api.github.com/users/nitesh7sid/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nitesh7sid/subscriptions","organizations_url":"https://api.github.com/users/nitesh7sid/orgs","repos_url":"https://api.github.com/users/nitesh7sid/repos","events_url":"https://api.github.com/users/nitesh7sid/events{/privacy}","received_events_url":"https://api.github.com/users/nitesh7sid/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":22,"created_at":"2018-07-12T06:44:40Z","updated_at":"2018-08-01T14:19:26Z","closed_at":null,"author_association":"NONE","pull_request":{"url":"https://api.github.com/repos/hyperledger/caliper/pulls/124","html_url":"https://github.com/hyperledger/caliper/pull/124","diff_url":"https://github.com/hyperledger/caliper/pull/124.diff","patch_url":"https://github.com/hyperledger/caliper/pull/124.patch"},"body":"PR for spitting caliper process into transaction sender and event listener.\r\n## Short Description\r\nCurrent version of the caliper relies on a single threaded process to send the transactions as well as listen for the transaction confirmation event. This can cause an issue for clients with higher send rates since node.js is itself a single threaded. The callback of the eventhub may get queued in the event loop and hence may result in incorrect confirmation time.\r\n\r\nTo tacke this issue, caliper can be split into 2 sub processes namely,\r\n* sender who is just responsible for generating the transaction load\r\n* listener which connects to the event peer, listens for the block event, note the timestamp and then publish it to Apache Kafka MQ. \r\n\r\nA persistence layer is required so that after the sender has finished generating the load, it can then consume blocks and associated confirmation time from the Queue and compute `time_final` property for every transaction."},"comment":{"url":"https://api.github.com/repos/hyperledger/caliper/issues/comments/409591545","html_url":"https://github.com/hyperledger/caliper/pull/124#issuecomment-409591545","issue_url":"https://api.github.com/repos/hyperledger/caliper/issues/124","id":409591545,"node_id":"MDEyOklzc3VlQ29tbWVudDQwOTU5MTU0NQ==","user":{"login":"nitesh7sid","id":18530954,"node_id":"MDQ6VXNlcjE4NTMwOTU0","avatar_url":"https://avatars3.githubusercontent.com/u/18530954?v=4","gravatar_id":"","url":"https://api.github.com/users/nitesh7sid","html_url":"https://github.com/nitesh7sid","followers_url":"https://api.github.com/users/nitesh7sid/followers","following_url":"https://api.github.com/users/nitesh7sid/following{/other_user}","gists_url":"https://api.github.com/users/nitesh7sid/gists{/gist_id}","starred_url":"https://api.github.com/users/nitesh7sid/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nitesh7sid/subscriptions","organizations_url":"https://api.github.com/users/nitesh7sid/orgs","repos_url":"https://api.github.com/users/nitesh7sid/repos","events_url":"https://api.github.com/users/nitesh7sid/events{/privacy}","received_events_url":"https://api.github.com/users/nitesh7sid/received_events","type":"User","site_admin":false},"created_at":"2018-08-01T14:19:26Z","updated_at":"2018-08-01T14:19:26Z","author_association":"NONE","body":"Hi @haojun - I am trying to move the code from demoOptional.js to client-util.js file. I could achieve the following:\r\n1) add a new status field 'NeedVerifyWithMQ' in the TxStatus object and set it true if MQ mode is running.\r\n2) from local-client.js send the un-verified transactions to the client-util.js using a different type of message.\r\n3) In client-util.js, handle the new message and push the un-verified transactions in some buffer.\r\n4) In the init function of client.js, invoke a function from client-util.js 'cosumeEvents' which connects to kafka and stores the confirmation time in a hash map. the hash map is defined in client-util.js\r\n4) in the startTest function of client-util.js, start the interval object to traverse the unconfirmed transactions buffer and find the time_final for each transactions. But now i am facing couple of issues:\r\n\r\n* before all the transactions are verified from the buffer, the round exists and the tool starts with the next round. how should i handle this?\r\n* how do I confirm that the round is over and clear the interval object ?\r\n\r\nbelow is the code of client-util.js. I am not sure if I am missing anything. Could you please check ? Thanks\r\n```/**\r\n* Copyright 2017 HUAWEI. All Rights Reserved.\r\n*\r\n* SPDX-License-Identifier: Apache-2.0\r\n*\r\n*/\r\n\r\n\r\n'use strict';\r\nconst log          = require('../util.js').log;\r\nlet processes  = {}; // {pid:{obj, promise}}\r\nlet txUpdateTime = 1000;\r\nconst kafka = require('kafka-node');\r\nconst listener_config = require(\"../../listener/listener-config.json\");\r\nlet confirmedTransactions = [];\r\nlet cachedEvents = new Map();\r\nlet unConfirmedTransactions;\r\nvar txUpdateInter = null;\r\nvar globalConsumer;\r\nconst TxStatus = require('../transaction.js');\r\nvar updateTail;\r\n\r\nfunction _consumeEvents(){\r\n    var Consumer = kafka.Consumer;\r\n    var KafkaClient = new kafka.KafkaClient({ kafkaHost: listener_config.broker_urls, requestTimeout: 300000000 });\r\n    var options = {\r\n    autoCommit: true,\r\n    fetchMaxWaitMs: 1000,\r\n    fetchMaxBytes: 5120 * 5120,\r\n    encoding: 'buffer',\r\n    groupId: \"groupID\" + Math.floor(Math.random() * Math.floor(100))};\r\n \r\n    var topics = [{\r\n        topic: listener_config.topic\r\n    }];\r\n \r\n    var consumer = new Consumer(KafkaClient, topics, options);\r\n    globalConsumer = consumer;\r\n    consumer.on('message', function (message) {\r\n        var buf = new Buffer(message.value); // Read string into a buffer.\r\n        var data = buf.toString('utf-8')\r\n        var block = JSON.parse(data).block\r\n        \r\n        for (var index = 0; index < block.data.data.length; index++) {\r\n            var channel_header = block.data.data[index].payload.header.channel_header;\r\n            var transaction_id = channel_header.tx_id\r\n            var confirmation_time = JSON.parse(data).validTime;\r\n            if (cachedEvents.get(transaction_id)  == undefined)\r\n           {\r\n                cachedEvents.set(transaction_id, confirmation_time)\r\n           }\r\n            else if (cachedEvents.get(transaction_id) != undefined){\r\n \r\n                var transactionObject = cachedEvents.get(transaction_id);\r\n                transactionObject.Set('time_final', confirmation_time);\r\n                transactionObject.SetVerification(true);\r\n                transactionObject.Set('status','success');\r\n                cachedEvents.set(transaction_id, transactionObject);\r\n                confirmedTransactions.push(transactionObject);\r\n            }\r\n        }\r\n    });\r\n \r\n    consumer.on('error', function(err){   \r\n        // retry if unable to connect to kafka\r\n        _consumeEvents()\r\n    })\r\n} \r\nmodule.exports._consumeEvents = _consumeEvents;\r\n/**\r\n * Call the Promise function for a process\r\n * @param {String} pid pid of the process\r\n * @param {Boolean} isResolve indicates resolve(true) or reject(false)\r\n * @param {Object} msg input for the Promise function\r\n */\r\nfunction setPromise(pid, isResolve, msg) {\r\n    let p = processes[pid];\r\n    if(p && p.promise && typeof p.promise !== 'undefined') {\r\n        if(isResolve) {\r\n               \r\n            p.promise.resolve(msg);\r\n        }\r\n        else {\r\n            p.promise.reject(msg);\r\n        }\r\n    }\r\n}\r\n\r\n/**\r\n * Push test result from a child process into the global array\r\n * @param {String} pid pid of the child process\r\n * @param {Object} data test result\r\n */\r\nfunction pushResult(pid, data) {\r\n    let p = processes[pid];\r\n    if(p && p.results && typeof p.results !== 'undefined') {\r\n        p.results.push(data);\r\n    }\r\n}\r\n\r\n/**\r\n * Push update value from a child process into the global array\r\n * @param {String} pid pid of the child process\r\n * @param {Object} data update value\r\n */\r\nfunction pushUpdate(pid, data) {\r\n    let p = processes[pid];\r\n    if(p && p.updates && typeof p.updates !== 'undefined') {\r\n\r\n        p.updates.push(data);\r\n    }\r\n}\r\n\r\n/**\r\n * Push update value from a child process into the global array\r\n * @param {String} pid pid of the child process\r\n * @param {Object} data update value\r\n */\r\nfunction pushUpdateForMQ(pid, data) {\r\n    unConfirmedTransactions.push(data);\r\n    /*let p = processes[pid];\r\n    if(p && p.updatesWithMQ && typeof p.updatesWithMQ !== 'undefined') {\r\n        p.updatesWithMQ = data;\r\n    }*/\r\n}\r\n\r\n\r\n/**\r\n * Launch a child process to do the test\r\n * @param {Array} updates array to save txUpdate results\r\n * @param {Array} results array to save the test results\r\n */\r\nfunction launchClient(updates, results) {\r\n    let path = require('path');\r\n    let childProcess = require('child_process');\r\n    let child = childProcess.fork(path.join(__dirname, 'local-client.js'));\r\n    let pid   = child.pid.toString();\r\n    processes[pid] = {obj: child, results: results, updates: updates};\r\n\r\n    child.on('message', function(msg) {\r\n        if(msg.type === 'testResult') {\r\n\r\n            pushResult(pid, msg.data);\r\n            setPromise(pid, true, msg.data);\r\n        }\r\n        else if(msg.type === 'error') {\r\n            setPromise(pid, false, new Error('Client encountered error:' + msg.data));\r\n        }\r\n        else if(msg.type === 'txUpdated') {\r\n            pushUpdate(pid, msg.data);\r\n        }\r\n        else if(msg.type === 'txUpdatedWithMQ') {\r\n            pushUpdateForMQ(pid, msg.data);\r\n        }\r\n    });\r\n\r\n    child.on('error', function(){\r\n        setPromise(pid, false, new Error('Client encountered unexpected error'));\r\n    });\r\n\r\n    child.on('exit', function(){\r\n        log('Client exited');\r\n        setPromise(pid, true, null);\r\n    });\r\n}\r\n\r\n/**\r\n * Start a test\r\n * @param {Number} number test clients' count\r\n * @param {JSON} message start message\r\n * @param {Array} clientArgs each element contains specific arguments for a client\r\n * @param {Array} updates array to save txUpdate results\r\n * @param {Array} results array to save the test results\r\n * @return {Promise} promise object\r\n */\r\n\r\nfunction update() {\r\n    console.log(\"Pending transactions length\", unConfirmedTransactions.length)\r\n    var data = [];\r\n    var len  = unConfirmedTransactions.length;\r\n    if(len > updateTail) {\r\n        data = unConfirmedTransactions.slice(updateTail, len);\r\n        updateTail = len;\r\n    }\r\n     demoRefreshData(data);\r\n}\r\n\r\nfunction demoRefreshData(updates) {\r\n\r\n    for (let i = 0; i < updates.length; i++) {\r\n        var sub = updates[i].submitted;\r\n        var submitted_transactions = updates[i].committed;\r\n        for (let j =0; j < submitted_transactions.length; j++) {\r\n            var transactionStatus = submitted_transactions[j].status\r\n            var TransactionStatus = new TxStatus(transactionStatus.id, transactionStatus.status, transactionStatus.time_create, transactionStatus.time_final, \r\n                transactionStatus.result, transactionStatus.verified, transactionStatus.flags, transactionStatus.error_messages);\r\n                \r\n            if (cachedEvents.get(TransactionStatus.GetID()) == undefined) {\r\n                // make an entry into map only if it is not a query OR the error flag == 0\r\n                if (TransactionStatus.GetFlag() == 0) {\r\n                    cachedEvents.set(TransactionStatus.GetID(), TransactionStatus)\r\n                } else {\r\n                    confirmedTransactions.push(TransactionStatus)\r\n                }\r\n            }\r\n            else {\r\n                TransactionStatus.Set('time_final', cachedEvents.get(TransactionStatus.GetID()))\r\n                TransactionStatus.SetVerification(true)\r\n                TransactionStatus.Set('status', 'success')\r\n                cachedEvents.set(TransactionStatus.GetID(), TransactionStatus)\r\n                confirmedTransactions.push(TransactionStatus)\r\n                //getDefaultStats(totalSubmitted, confirmedTransactions, label)\r\n            }   \r\n        }  \r\n    }\r\n    console.log(\"confirmed\", confirmedTransactions.length)\r\n    //newStats = blockchain.getDefaultTxStats(confirmedTransactions, false);\r\n   //getDefaultStats(totalSubmitted, label)\r\n}\r\n\r\nfunction startTest(number, message, clientArgs, updates, results) {\r\n    let count = 0;\r\n    for(let i in processes) {\r\n        i;  // avoid eslint error\r\n        count++;\r\n    }\r\n    if(count === number) {\r\n        txUpdateInter = null;  \r\n        unConfirmedTransactions = [];\r\n        updateTail = 0;\r\n    if(txUpdateInter === null) {\r\n        // start a interval to query updates\r\n        txUpdateInter = setInterval(update, 1 * 1000);\r\n    } \r\n\r\n        // already launched clients\r\n        let txPerClient;\r\n        if (message.numb) {\r\n            // Run specified number of transactions\r\n            txPerClient  = Math.floor(message.numb / number);\r\n\r\n            // trim should be based on client number if specified with txNumber\r\n            if (message.trim) {\r\n                message.trim = Math.floor(message.trim / number);\r\n            }\r\n\r\n            if(txPerClient < 1) {\r\n                txPerClient = 1;\r\n            }\r\n            message.numb = txPerClient;\r\n        } else if (message.txDuration) {\r\n            // Run for time specified txDuration based on clients\r\n            // Do nothing, we run for the time specified within message.txDuration\r\n        } else {\r\n            return Promise.reject(new Error('Unconditioned transaction rate driving mode'));\r\n        }\r\n\r\n        message.clients = number;\r\n\r\n        let promises = [];\r\n        let idx = 0;\r\n\r\n        for(let id in processes) {\r\n            let client = processes[id];\r\n            let p = new Promise((resolve, reject) => {\r\n                client.promise = {\r\n                    resolve: resolve,\r\n                    reject:  reject\r\n                };\r\n            });\r\n            promises.push(p);\r\n            client.results = results;\r\n            client.updates = updates;\r\n            message.clientargs = clientArgs[idx];\r\n            message.clientIdx = idx;\r\n            idx++;\r\n            client.obj.send(message);\r\n        }\r\n\r\n        return Promise.all(promises).then(()=>{\r\n            // clear promises\r\n            \r\n                for(let client in processes) {\r\n                    delete client.promise;\r\n                }   \r\n                return Promise.resolve();\r\n         \r\n        }).catch((err)=>{\r\n            return Promise.reject(err);\r\n        });\r\n\r\n    }\r\n\r\n    // launch clients\r\n    processes = {};\r\n    for(let i = 0 ; i < number ; i++) {\r\n        launchClient(updates, results);\r\n    }\r\n\r\n    // start test\r\n    return startTest(number, message, clientArgs, updates, results);\r\n}\r\nmodule.exports.startTest = startTest;\r\n\r\n/**\r\n * Send message to all child processes\r\n * @param {JSON} message message\r\n * @return {Number} number of child processes\r\n */\r\nfunction sendMessage(message) {\r\n    for(let pid in processes) {\r\n        processes[pid].obj.send(message);\r\n    }\r\n    return processes.length;\r\n}\r\nmodule.exports.sendMessage = sendMessage;\r\n\r\n/**\r\n * Stop all test clients(child processes)\r\n */\r\nfunction stop() {\r\n    for(let pid in processes) {\r\n        processes[pid].obj.kill();\r\n    }\r\n    processes = {};\r\n}\r\nmodule.exports.stop = stop;\r\n```\r\n"}}
{"actor":{"display_login":"nitesh7sid"}}