diff --git a/init.js b/init.js index 053cee8..1a2f257 100644 --- a/init.js +++ b/init.js @@ -5,11 +5,13 @@ let argv = require('minimist')(process.argv.slice(2)); let config = fs.readFileSync("./config.json"); let coinConfig = fs.readFileSync("./coinConfig.json"); let protobuf = require('protocol-buffers'); +let path = require('path'); global.support = require("./lib/support.js")(); global.config = JSON.parse(config); global.mysql = mysql.createPool(global.config.mysql); global.protos = protobuf(fs.readFileSync('./lib/data.proto')); +global.argv = argv; let comms; let coinInc; @@ -52,46 +54,60 @@ global.mysql.query("SELECT * FROM config").then(function (rows) { global.database.initEnv(); global.coinFuncs.blockedAddresses.push(global.config.pool.address); global.coinFuncs.blockedAddresses.push(global.config.payout.feeAddress); - switch(argv.module){ - case 'pool': - global.config.ports = []; - global.mysql.query("SELECT * FROM port_config").then(function(rows){ - rows.forEach(function(row){ - row.hidden = row.hidden === 1; - row.ssl = row.ssl === 1; - global.config.ports.push({ - port: row.poolPort, - difficulty: row.difficulty, - desc: row.portDesc, - portType: row.portType, - hidden: row.hidden, - ssl: row.ssl + if (argv.hasOwnProperty('tool') && fs.existsSync('./tools/'+argv.tool+'.js')) { + require('./tools/'+argv.tool+'.js'); + } else if (argv.hasOwnProperty('module')){ + switch(argv.module){ + case 'pool': + global.config.ports = []; + global.mysql.query("SELECT * FROM port_config").then(function(rows){ + rows.forEach(function(row){ + row.hidden = row.hidden === 1; + row.ssl = row.ssl === 1; + global.config.ports.push({ + port: row.poolPort, + difficulty: row.difficulty, + desc: row.portDesc, + portType: row.portType, + hidden: row.hidden, + ssl: row.ssl + }); }); + }).then(function(){ + require('./lib/pool.js'); }); - }).then(function(){ - require('./lib/pool.js'); - }); - break; - case 'blockManager': - require('./lib/blockManager.js'); - break; - case 'payments': - require('./lib/payments.js'); - break; - case 'api': - require('./lib/api.js'); - break; - case 'remoteShare': - require('./lib/remoteShare.js'); - break; - case 'worker': - require('./lib/worker.js'); - break; - case 'longRunner': - require('./lib/longRunner.js'); - break; - default: - console.error("Invalid module provided. Please provide a valid module"); - process.exit(1); + break; + case 'blockManager': + require('./lib/blockManager.js'); + break; + case 'payments': + require('./lib/payments.js'); + break; + case 'api': + require('./lib/api.js'); + break; + case 'remoteShare': + require('./lib/remoteShare.js'); + break; + case 'worker': + require('./lib/worker.js'); + break; + case 'longRunner': + require('./lib/longRunner.js'); + break; + default: + console.error("Invalid module provided. Please provide a valid module"); + process.exit(1); + } + } else { + console.error("Invalid module/tool provided. Please provide a valid module/tool"); + console.error("Valid Modules: pool, blockManager, payments, api, remoteShare, worker, longRunner"); + let valid_tools = "Valid Tools: "; + fs.readdirSync('./tools/').forEach(function(line){ + valid_tools += path.parse(line).name + ", "; + }); + valid_tools = valid_tools.slice(0, -2); + console.error(valid_tools); + process.exit(1); } }); \ No newline at end of file diff --git a/lib/local_comms.js b/lib/local_comms.js index 493bde7..da5d98a 100644 --- a/lib/local_comms.js +++ b/lib/local_comms.js @@ -18,9 +18,9 @@ function Database(){ path: global.config.db_storage_path, maxDbs: 10, mapSize: 24 * 1024 * 1024 * 1024, - noSync: true, + noSync: false, mapAsync: true, - useWritemap: true, + useWritemap: false, noMetaSync: true, maxReaders: 512 }); @@ -179,6 +179,124 @@ 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; + debug(shareObject.length + ' shares to process'); + shareObject.forEach(function(share){ + //Data is the share object at this point. + shareCount += 1; + if (typeof(share.shares) === "number") { + if (!shares.hasOwnProperty(share.blockHeight)) { + shares[share.blockHeight] = []; + } + shares[share.blockHeight].push(share); + let minerID = share.paymentAddress; + if (typeof(share.paymentID) !== 'undefined' && share.paymentID.length > 10) { + minerID = minerID + '.' + share.paymentID; + } + let minerIDWithIdentifier = minerID + "_" + share.identifier; + // this.incrementCacheData('global_stats', [{location: 'totalHashes', value: share.shares}, {location: 'roundHashes', value: share.shares}]); + if (!cachedData.global_stats.hasOwnProperty('totalHashes') || cachedData.global_stats.totalHashes === null) { + cachedData.global_stats.totalHashes = 0; + } + if (!cachedData.global_stats.hasOwnProperty('roundHashes') || cachedData.global_stats.roundHashes === null) { + 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 === null) { + cachedData[stats_type].totalHashes = 0; + } + if (!cachedData[stats_type].hasOwnProperty('roundHashes') || cachedData[stats_type].roundHashes === null) { + 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 === null) { + cachedData[minerIDWithIdentifier].totalHashes = 0; + } + if (!cachedData[minerIDWithIdentifier].hasOwnProperty('goodShares') || cachedData[minerIDWithIdentifier].goodShares === null) { + cachedData[minerIDWithIdentifier].goodShares = 0; + } + if (!cachedData[minerID].hasOwnProperty('totalHashes') || cachedData[minerID].totalHashes === null) { + cachedData[minerID].totalHashes = 0; + } + if (!cachedData[minerID].hasOwnProperty('goodShares') || cachedData[minerID].goodShares === null) { + cachedData[minerID].goodShares = 0; + } + cachedData[minerIDWithIdentifier].totalHashes += share.shares; + cachedData[minerID].totalHashes += share.shares; + cachedData[minerIDWithIdentifier].goodShares += 1; + cachedData[minerID].goodShares += 1; + } else { + console.error("Error in share parser: " + JSON.stringify(share)); + } + 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(final_share){ + sharesSeen += 1; + try { + txn.putBinary(global.database.shareDB, parseInt(key), global.protos.Share.encode(final_share)); + } catch (e) { + debug(final_share); + } + if (Object.keys(shares).length === blocksSeen && sharesSeen === shares[key].length){ + debug('Made it to where I can do the insert'); + txn.commit(); + return true; + } + }); + } + } + } + }); + return false; + }; + this.storeInvalidShare = function(shareData, callback){ try { let share = global.protos.InvalidShare.decode(shareData); @@ -218,6 +336,21 @@ function Database(){ return highestBlock; }; + this.getBlockByID = function(blockID){ + this.refreshEnv(); + debug("Getting the data for blockID: " + blockID); + let txn = this.env.beginTxn({readOnly: true}); + let data = txn.getBinary(this.blockDB, blockID); + if (data === null){ + debug("Unable to get block at height: "+ blockID); + return false; + } + let blockData = global.protos.Block.decode(data); + txn.commit(); + debug("Done getting the last block for: "+ blockData.poolType + " height of: "+ blockID); + return blockData; + }; + this.calculateShares = function(blockData, blockHeight){ debug("Calculating shares for "+ blockData.hash); this.refreshEnv(); @@ -376,6 +509,16 @@ function Database(){ txn.commit(); }; + this.bulkSetCache = function(cacheUpdates){ + let txn = this.env.beginTxn(); + for (let key in cacheUpdates){ + if (cacheUpdates.hasOwnProperty(key)){ + txn.putString(this.cacheDB, key, JSON.stringify(cacheUpdates[key])); + } + } + txn.commit(); + }; + this.cleanShareDB = function() { /* This function takes the difficulty of the current block, and the last PPS block. If it's 0, save everything, @@ -477,12 +620,14 @@ function Database(){ this.refreshEnv = function(){ if (this.dirtyenv === true){ + /* console.log("Database Worker: Reloading LMDB Env"); global.database.env.sync(function(){ }); clearInterval(this.intervalID); global.database.env.close(); this.initEnv(); + */ } }; diff --git a/lib/remoteShare.js b/lib/remoteShare.js index dd291f1..60b5fff 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,15 @@ app.use(function(req, res, next){ })); }); +// Master/Slave communication Handling +function messageHandler(message) { + if (typeof message.shares === "number"){ + shareData.push(message); + } +} + +process.on('message', messageHandler); + app.post('/leafApi', function (req, res) { try { let msgData = global.protos.WSData.decode(req.body); @@ -29,14 +39,11 @@ 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; + try { + process.send(global.protos.Share.decode(msgData.msg)); + } catch (e){ + } + return res.json({'success': true}); case global.protos.MESSAGETYPE.BLOCK: global.database.storeBlock(msgData.exInt, msgData.msg, function(data){ if (!data){ @@ -68,8 +75,16 @@ 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){ + global.database.storeBulkShares(shareData); + shareData = []; + } + }, 1000); + for (let i = 0; i < numWorkers; i++) { let worker = cluster.fork(); + worker.on('message', messageHandler); workerList.push(worker); } @@ -81,6 +96,7 @@ if (cluster.isMaster) { console.log('Worker ' + worker.process.pid + ' died with code: ' + code + ', and signal: ' + signal); console.log('Starting a new worker'); worker = cluster.fork(); + worker.on('message', messageHandler); workerList.push(worker); }); } else { diff --git a/lib/worker.js b/lib/worker.js index f8c0486..8e31caf 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -137,6 +137,7 @@ function updateShareStats() { if (globalMinerList === false) { globalMinerList = []; } + let cache_updates = {}; // pplns: 0, pps: 0, solo: 0, prop: 0, global: 0 ['pplns', 'pps', 'solo', 'prop', 'global'].forEach(function (key) { let cachedData = global.database.getCache(key + "_stats"); @@ -172,7 +173,7 @@ function updateShareStats() { minerHistory: [{ts: currentTime, cn: cachedData.hash}] }; } - global.database.setCache(key + "_stats", cachedData); + cache_updates[key + "_stats"] = cachedData; }); minerList.forEach(function (miner) { if (globalMinerList.indexOf(miner) === -1) { @@ -206,7 +207,7 @@ function updateShareStats() { badShares: 0 }; } - global.database.setCache(miner, cachedData); + cache_updates[miner] = cachedData; }); globalMinerList.forEach(function (miner) { if (minerList.indexOf(miner) === -1) { @@ -234,14 +235,15 @@ function updateShareStats() { }); } minerStats.hash = 0; - global.database.setCache(miner, minerStats); + cache_updates[miner] = minerStats; } } }); Object.keys(identifiers).forEach(function (key) { - global.database.setCache(key + '_identifiers', identifiers[key]); + cache_updates[key + '_identifiers'] = identifiers[key]; }); - global.database.setCache('minerList', globalMinerList); + cache_updates['minerList'] = globalMinerList; + global.database.bulkSetCache(cache_updates); callback(null); }); } @@ -449,6 +451,8 @@ function updateBlockHeader() { value: body.result.block_header.reward, ts: body.result.block_header.timestamp })]); + } else if (body.result.block_header.hash === lastBlockHash) { + console.log("Block headers identical to historical header. Ignoring"); } else { console.error("GetLastBlockHeader Error during block header update"); } diff --git a/package.json b/package.json index 0c3842f..db30ebe 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ "moment": "2.17.1", "multi-hashing": "git+https://github.com/Snipa22/node-multi-hashing-aesni.git", "mysql": "2.12.0", - "node-lmdb": "^0.4.3", + "node-lmdb": "0.4.12", "promise-mysql": "3.0.0", "protocol-buffers": "^3.2.1", "range": "0.0.3", diff --git a/tools/blocks.js b/tools/blocks.js new file mode 100644 index 0000000..1de4149 --- /dev/null +++ b/tools/blocks.js @@ -0,0 +1,44 @@ +const valid_actions = ['finder', 'stats', 'list']; +let error = 0; + +if (!global.argv.hasOwnProperty('action') || valid_actions.indexOf(global.argv.action) === -1) { + console.error("No action provided to block module."); + console.error("Valid actions: " + valid_actions.join(', ')); +} + +switch (global.argv.action) { + case 'finder': + if (!global.argv.hasOwnProperty('value')) { + console.error('No block provided in value field. Please use --value=blockID'); + error = 1; + break; + } + let blockID = parseInt(global.argv.value); + let block_data = global.database.getBlockByID(blockID); + /* + required string hash = 1; + required int64 difficulty = 2; + required int64 shares = 3; + required int64 timestamp = 4; + required POOLTYPE poolType = 5; + required bool unlocked = 6; + required bool valid = 7; + optional int64 value = 8; + */ + if (!block_data) { + console.error("Invalid blockID provided."); + error = 1; + break; + } + console.log("Data for block: " + blockID + '\n' + + 'Hash: ' + block_data.hash + '\n' + + 'Difficulty: ' + block_data.hash + '\n' + + 'Hashes Required: ' + block_data.hash + '\n' + + 'Find Time: ' + block_data.hash + '\n' + + 'Pool Type: ' + block_data.hash + '\n' + + 'Unlocked: ' + block_data.hash + '\n' + + 'Valid: ' + block_data.hash + '\n' + + 'Value: ' + block_data.hash + '\n'); +} + +process.exit(error); \ No newline at end of file