diff --git a/lib/local_comms.js b/lib/local_comms.js index 714e228..3ca0829 100644 --- a/lib/local_comms.js +++ b/lib/local_comms.js @@ -179,6 +179,114 @@ function Database(){ callback(true); }; + this.storeBulkShares = function(shareObject) { + let cachedData = { + global_stats: global.database.getCache('global_stats'), + pplns_stats: global.database.getCache('pplns_stats'), + pps_stats: global.database.getCache('pps_stats'), + solo_stats: global.database.getCache('solo_stats') + }; + let shares = {}; // Shares keyed by blockID + let shareCount = 0; + shareObject.forEach(function(data){ + //Data is the share object at this point. + let share = global.protos.Share.decode(data); + if (!shares.hasOwnProperty(share.blockHeight)){ + shares[share.blockHeight] = []; + } + shares[share.blockHeight].push(data); + shareCount += 1; + let minerID = share.paymentAddress; + if (typeof(share.paymentID) !== 'undefined' && share.paymentID.length > 10) { + minerID = minerID + '.' + share.paymentID; + } + let minerIDWithIdentifier = minerID + "_" + share.identifier; + if (!cachedData.global_stats.hasOwnProperty('totalHashes')){ + cachedData.global_stats.totalHashes = 0; + } + if (!cachedData.global_stats.hasOwnProperty('roundHashes')){ + cachedData.global_stats.roundHashes = 0; + } + cachedData.global_stats.totalHashes += share.shares; + cachedData.global_stats.roundHashes += share.shares; + let stats_type = 'pplns_stats'; + switch (share.poolType) { + case global.protos.POOLTYPE.PPLNS: + stats_type = 'pplns_stats'; + break; + case global.protos.POOLTYPE.PPS: + stats_type = 'pps_stats'; + break; + case global.protos.POOLTYPE.SOLO: + stats_type = 'solo_stats'; + break; + } + if (!cachedData[stats_type].hasOwnProperty('totalHashes')){ + cachedData[stats_type].totalHashes = 0; + } + if (!cachedData[stats_type].hasOwnProperty('roundHashes')){ + cachedData[stats_type].roundHashes = 0; + } + cachedData[stats_type].totalHashes += share.shares; + cachedData[stats_type].roundHashes += share.shares; + if (!cachedData.hasOwnProperty(minerID)){ + let minerCache = global.database.getCache(minerID); + if (minerCache === false){ + minerCache = {totalHashes: 0, goodShares: 0}; + } + cachedData[minerID] = minerCache; + } + if (!cachedData.hasOwnProperty(minerIDWithIdentifier)){ + let minerCache = global.database.getCache(minerIDWithIdentifier); + if (minerCache === false){ + minerCache = {totalHashes: 0, goodShares: 0}; + } + cachedData[minerIDWithIdentifier] = minerCache; + } + if (!cachedData[minerIDWithIdentifier].hasOwnProperty('totalHashes')){ + cachedData[minerIDWithIdentifier].totalHashes = 0; + } + if (!cachedData[minerIDWithIdentifier].hasOwnProperty('goodShares')){ + cachedData[minerIDWithIdentifier].goodShares = 0; + } + if (!cachedData[minerID].hasOwnProperty('totalHashes')){ + cachedData[minerID].totalHashes = 0; + } + if (!cachedData[minerID].hasOwnProperty('goodShares')){ + cachedData[minerID].goodShares = 0; + } + cachedData[minerIDWithIdentifier].totalHashes += share.shares; + cachedData[minerID].totalHashes += share.shares; + cachedData[minerIDWithIdentifier].goodShares += 1; + cachedData[minerID].goodShares += 1; + if(shareObject.length === shareCount){ + // Perform insert and return + let txn = global.database.env.beginTxn(); + for (let key in cachedData){ + if (cachedData.hasOwnProperty(key)){ + txn.putString(global.database.cacheDB, key, JSON.stringify(cachedData[key])); + } + } + let blocksSeen = 0; + for (let key in shares){ + if (shares.hasOwnProperty(key)){ + blocksSeen += 1; + let sharesSeen = 0; + shares[key].forEach(function(share){ + sharesSeen += 1; + txn.putBinary(global.database.shareDB, key, share); + if (Object.keys(shares).length === blocksSeen && sharesSeen === shares[key].length){ + txn.commit(); + return true; + } + }); + } + } + } + }); + return false; + }; + this.storeInvalidShare = function(shareData, callback){ try { let share = global.protos.InvalidShare.decode(shareData); diff --git a/lib/remoteShare.js b/lib/remoteShare.js index dd291f1..ed02ae8 100644 --- a/lib/remoteShare.js +++ b/lib/remoteShare.js @@ -12,6 +12,7 @@ if (cluster.isMaster) { } else { threadName = "(Worker " + cluster.worker.id + " - " + process.pid + ") "; } +let shareData = []; // Websocket Stuffs. app.use(function(req, res, next){ @@ -21,6 +22,13 @@ app.use(function(req, res, next){ })); }); +// Master/Slave communication Handling +function messageHandler(message) { + shareData.push(message.data); +} + +process.on('message', messageHandler); + app.post('/leafApi', function (req, res) { try { let msgData = global.protos.WSData.decode(req.body); @@ -29,14 +37,8 @@ app.post('/leafApi', function (req, res) { } switch (msgData.msgType) { case global.protos.MESSAGETYPE.SHARE: - global.database.storeShare(msgData.exInt, msgData.msg, function(data){ - if (!data){ - return res.status(400).end(); - } else { - return res.json({'success': true}); - } - }); - break; + process.send({block: msgData.exInt, data: msgData.msg}); + return res.json({'success': true}); case global.protos.MESSAGETYPE.BLOCK: global.database.storeBlock(msgData.exInt, msgData.msg, function(data){ if (!data){ @@ -68,6 +70,14 @@ if (cluster.isMaster) { let numWorkers = require('os').cpus().length; console.log('Master cluster setting up ' + numWorkers + ' workers...'); + setInterval(function(){ + if (Object.keys(shareData).length > 0){ + if (global.database.storeBulkShares(shareData)){ + shareData = []; + } + } + }, 1000); + for (let i = 0; i < numWorkers; i++) { let worker = cluster.fork(); workerList.push(worker);