RanchiMall/node-stratum-pool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restructured pool object - much cleaner and easier to follow flow.

Browse Source
This commit is contained in:
Matt 2014-03-26 22:41:25 -06:00
parent 5b849dce59
commit b05f7689f3
2 changed files with 311 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
lib/jobManager.js
View File

@ -84,15 +84,16 @@ var JobManager = module.exports = function JobManager(maxDifficulty, hashDigest,
(_this.currentJob.rpcData.transactions.length != rpcData.transactions.length)
if (updatedTransactions && (Date.now() - lastTransactionUpdateCheck >= options.txRefreshInterval)){
lastTransactionUpdateCheck = Date.now();
}
else
if (updatedTransactions && (Date.now() - lastTransactionUpdateCheck <= options.txRefreshInterval)){
updatedTransactions = false;
}
//Update current job if new block or new transactions
if (isNewBlock || updatedTransactions){
lastTransactionUpdateCheck = Date.now();
var tmpBlockTemplate = new blockTemplate(
maxDifficulty,
jobCounter.next(),

522
lib/pool.js
View File

@ -37,7 +37,6 @@ var pool = module.exports = function pool(options, authorizeFn){
this.options = options;
var _this = this;
var publicKeyBuffer;
var blockPollingIntervalId;
@ -53,12 +52,13 @@ var pool = module.exports = function pool(options, authorizeFn){
case 'sha256':
case 'skein':
case 'hefty1':
case 'keccak':
return '00000000ffff0000000000000000000000000000000000000000000000000000';
case 'scrypt':
case 'scrypt-jane':
case 'x11':
case 'quark':
case 'keccak':
case 'max':
return '0000ffff00000000000000000000000000000000000000000000000000000000';
default:
emitErrorLog('The ' + options.coin.algorithm + ' hashing algorithm is not supported.');
@ -114,10 +114,112 @@ var pool = module.exports = function pool(options, authorizeFn){
emitLog('Starting pool for ' + options.coin.name + ' [' + options.coin.symbol.toUpperCase() + ']');
SetupJobManager();
SetupVarDiff();
SetupDaemonInterface();
SetupApi();
SetupDaemonInterface(function(){
DetectCoinData(function(){
OnBlockchainSynced(function(){
GetFirstJob(function(){
OutputPoolInfo();
SetupBlockPolling();
StartStratumServer();
SetupPeer();
});
});
});
});
};
function GetFirstJob(finishedCallback){
GetBlockTemplate(function(error, result){
if (error) {
emitErrorLog('Error with getblocktemplate on creating first job, server cannot start');
return;
}
Object.keys(options.ports).forEach(function(port){
var portDiff = options.ports[port].diff;
if (portDiff > _this.jobManager.currentJob.difficulty)
emitWarningLog('Pool difficulty of ' + portDiff + ' on port ' + port +
' was set higher than network difficulty of ' + _this.jobManager.currentJob.difficulty);
});
finishedCallback();
});
}
function OutputPoolInfo(){
if (!process.env.forkId || process.env.forkId === '0')
emitLog(['Pool Server Started...',
'Running Coin Pool:\t' + options.coin.name + ' [' + options.coin.symbol.toUpperCase() + ']',
'Network Connected:\t' + (options.testnet ? 'Testnet' : 'Live Blockchain'),
'Detected Reward Type:\t' + options.coin.reward,
'Current Block Height:\t' + _this.jobManager.currentJob.rpcData.height,
'Current Connect Peers:\t' + options.initStats.connections,
'Network Hash Rate:\t' + (options.initStats.networkHashRate / 1e3).toFixed(6) + ' KH/s',
'Network Difficulty:\t' + _this.jobManager.currentJob.difficulty
].join('\n\t\t\t\t\t\t')
);
}
function OnBlockchainSynced(syncedCallback){
var checkSynced = function(displayNotSynced){
_this.daemon.cmd('getblocktemplate', [], function(results){
var synced = results.every(function(r){
return !r.error || r.error.code !== -10;
});
if (synced){
syncedCallback();
}
else{
if (displayNotSynced) displayNotSynced();
setTimeout(checkSynced, 5000);
generateProgress(smallestHeight);
}
});
};
checkSynced(function(){
emitErrorLog('Daemon is still syncing with network (download blockchain) - server will be started once synced');
});
var generateProgress = function(){
_this.daemon.cmd('getinfo', [], function(results) {
var blockCount = results.sort(function (a, b) {
return b.response.blocks - a.response.blocks;
})[0].response.blocks;
//get list of peers and their highest block height to compare to ours
_this.daemon.cmd('getpeerinfo', [], function(results){
var peers = results[0].response;
var totalBlocks = peers.sort(function(a, b){
return b.startingheight - a.startingheight;
})[0].startingheight;
var percent = (blockCount / totalBlocks * 100).toFixed(2);
//Only let the first fork show synced status or the log wil look flooded with it
if (!process.env.forkId || process.env.forkId === '0')
emitWarningLog('Downloaded ' + percent + '% of blockchain from network');
});
});
};
}
function SetupApi() {
if (typeof(options.api) !== 'object' || typeof(options.api.start) !== 'function') {
return;
@ -126,6 +228,7 @@ var pool = module.exports = function pool(options, authorizeFn){
}
}
function SetupPeer(){
if (!options.p2p || !options.p2p.enabled)
return;
@ -139,10 +242,11 @@ var pool = module.exports = function pool(options, authorizeFn){
}).on('error', function(msg){
emitWarningLog(msg);
}).on('blockFound', function(hash){
this.processBlockNotify(hash);
_this.processBlockNotify(hash);
});
}
function SetupVarDiff(){
_this.varDiff = {};
Object.keys(options.ports).forEach(function(port) {
@ -151,6 +255,7 @@ var pool = module.exports = function pool(options, authorizeFn){
});
}
/*
Coin daemons either use submitblock or getblocktemplate for submitting new blocks
*/
@ -195,13 +300,13 @@ var pool = module.exports = function pool(options, authorizeFn){
_this.jobManager.on('newBlock', function(blockTemplate){
//Check if stratumServer has been initialized yet
if ( typeof(_this.stratumServer ) !== 'undefined') {
emitLog('Detected new block');
//emitLog('Detected new block');
_this.stratumServer.broadcastMiningJobs(blockTemplate.getJobParams());
}
}).on('updatedBlock', function(blockTemplate){
//Check if stratumServer has been initialized yet
if ( typeof(_this.stratumServer ) !== 'undefined') {
emitLog('Detected updated block transactions');
//emitLog('Detected updated block transactions');
var job = blockTemplate.getJobParams();
job[8] = false;
_this.stratumServer.broadcastMiningJobs(job);
@ -235,181 +340,12 @@ var pool = module.exports = function pool(options, authorizeFn){
}
function SetupDaemonInterface(){
function SetupDaemonInterface(finishedCallback){
if (!_this.daemon) _this.daemon = new daemon.interface(options.daemons);
_this.daemon = new daemon.interface(options.daemons);
_this.daemon.once('online', function(){
async.parallel({
addressInfo: function(callback){
_this.daemon.cmd('validateaddress', [options.address], function(results){
//Make sure address is valid with each daemon
var allValid = results.every(function(result){
if (result.error || !result.response){
emitErrorLog('validateaddress rpc error on daemon instance ' +
result.instance.index + ', error +' + JSON.stringify(result.error));
}
else if (!result.response.isvalid)
emitErrorLog('Daemon instance ' + result.instance.index +
' reports address is not valid');
return result.response && result.response.isvalid;
});
if (allValid)
callback(null, results[0].response);
else{
callback('not all addresses are valid')
}
});
},
info: function(callback){
_this.daemon.cmd('getinfo', [], function(results){
// Print which network each daemon is running on
var isTestnet;
var allValid = results.every(function(result){
if (result.error){
emitErrorLog('getmininginfo on init failed with daemon instance ' +
result.instance.index + ', error ' + JSON.stringify(result.error)
);
return false;
}
if (typeof isTestnet === 'undefined'){
isTestnet = result.response.testnet;
return true;
}
else if (isTestnet !== result.response.testnet){
emitErrorLog('not all daemons are on same network');
return false;
}
else
return true;
});
if (!allValid){
callback('could not getmininginfo correctly on each daemon');
return;
}
//Find and return the response with the largest block height (most in-sync)
var largestHeight = results.sort(function(a, b){
return b.response.blocks - a.response.blocks;
})[0].response;
callback(null, largestHeight);
});
},
rewardType: function(callback){
_this.daemon.cmd('getdifficulty', [], function(results){
var isPos = results.every(function(result){
if (result.error){
emitErrorLog('getinfo on init failed with daemon instance ' +
result.instance.index + ', error ' + JSON.stringify(result.error)
);
return false;
}
return isNaN(result.response) && 'proof-of-stake' in result.response;
});
options.coin.reward = isPos ? 'POS' : 'POW'
callback(null);
});
},
submitMethod: function(callback){
/* This checks to see whether the daemon uses submitblock
or getblocktemplate for submitting new blocks */
_this.daemon.cmd('submitblock', [], function(results){
var couldNotDetectMethod = results.every(function(result){
if (result.error && result.error.message === 'Method not found'){
options.hasSubmitMethod = false;
callback(null);
return false;
}
else if (result.error && result.error.code === -1){
options.hasSubmitMethod = true;
callback(null);
return false;
}
else
return true;
});
if (couldNotDetectMethod){
emitErrorLog('Could not detect block submission RPC method, ' + JSON.stringify(results));
callback('block submission detection failed');
}
});
}
}, function(err, results){
if (err){
emitErrorLog('Could not start pool, ' + JSON.stringify(err));
return;
}
if (options.coin.reward === 'POS' && typeof(results.addressInfo.pubkey) == 'undefined') {
// address provided is not of the wallet.
emitErrorLog('The address provided is not from the daemon wallet.');
return;
}
publicKeyBuffer = (function(){
switch(options.coin.reward){
case 'POS':
return util.pubkeyToScript(results.addressInfo.pubkey);
case 'POW':
if (options.coin.algorithm)
return util.addressToScript(results.addressInfo.address);
}
})();
var networkType = results.info.testnet ? 'testnet' : 'live blockchain';
GetBlockTemplate(function(error, result){
if (error) {
if (error.code === -10){
emitErrorLog('Daemon is still syncing with network (download blocks)');
WaitForSync(results.info.blocks);
}
else
emitErrorLog('Error with getblocktemplate on initializing');
} else {
Object.keys(options.ports).forEach(function(port){
var portDiff = options.ports[port].diff;
if (portDiff > _this.jobManager.currentJob.difficulty)
emitWarningLog('Pool difficulty of ' + portDiff + ' on port ' + port +
' was set higher than network difficulty of ' + _this.jobManager.currentJob.difficulty);
});
emitLog('Connected to ' + networkType +
'; detected ' + options.coin.reward +
' reward type; block height of ' + results.info.blocks +
'; difficulty of ' + _this.jobManager.currentJob.difficulty);
SetupBlockPolling();
StartStratumServer();
SetupPeer();
}
});
});
finishedCallback();
}).on('connectionFailed', function(error){
emitErrorLog('Failed to connect daemon(s): ' + JSON.stringify(error));
@ -423,54 +359,197 @@ var pool = module.exports = function pool(options, authorizeFn){
}
function DetectCoinData(finishedCallback){
function WaitForSync(currentBlocks){
async.waterfall([
var generateProgress = function(currentBlocks){
//get list of peers and their highest block height to compare to ours
function(callback){
_this.daemon.cmd('validateaddress', [options.address], function(results){
_this.daemon.cmd('getpeerinfo', [], function(results){
var peers = results[0].response;
var totalBlocks = peers.sort(function(a, b){
return b.startingheight - a.startingheight;
})[0].startingheight;
var percent = (currentBlocks / totalBlocks * 100).toFixed(2);
//Only let the first fork show synced status or the log wil look flooded with it
if (!process.env.forkId || process.env.forkId === '0')
emitWarningLog('Downloaded ' + percent + '% of blockchain from network');
});
};
generateProgress(currentBlocks);
setInterval(function(){
_this.daemon.cmd('getblocktemplate', [], function(results){
var synced = results.every(function(r){
return !r.error || r.error.code !== -10;
});
if (synced){
SetupDaemonInterface();
}
else{
_this.daemon.cmd('getinfo', [], function(results){
var smallestHeight = results.sort(function(a, b){
return b.response.blocks - a.response.blocks;
})[0].response.blocks;
generateProgress(smallestHeight);
//Make sure address is valid with each daemon
var allValid = results.every(function(result){
if (result.error || !result.response){
emitErrorLog('validateaddress rpc error on daemon instance ' +
result.instance.index + ', error +' + JSON.stringify(result.error));
}
else if (!result.response.isvalid)
emitErrorLog('Daemon instance ' + result.instance.index +
' reports address is not valid');
return result.response && result.response.isvalid;
});
}
});
if (!allValid){
callback('not all addresses are valid');
return;
}
}, 5000);
//Try to find result that owns address in case of POS coin with multi daemons
var ownedInfo = results.filter(function(r){
return r.response.ismine;
});
callback(null, !!ownedInfo.length ? ownedInfo[0].response : results[0].response);
});
},
function(addressInfo, callback){
_this.daemon.cmd('getdifficulty', [], function(results){
//This detects if a coin is POS because getdiff returns an object instead of a number
var isPos = results.every(function(result){
if (result.error){
emitErrorLog('getinfo on init failed with daemon instance ' +
result.instance.index + ', error ' + JSON.stringify(result.error)
);
return false;
}
return isNaN(result.response) && 'proof-of-stake' in result.response;
});
options.coin.reward = isPos ? 'POS' : 'POW';
/* POS coins must use the pubkey in coinbase transaction, and pubkey is
only given if address is owned by wallet.*/
if (options.coin.reward === 'POS' && typeof(addressInfo.pubkey) == 'undefined') {
emitErrorLog('The address provided is not from the daemon wallet - this is required for POS coins.');
return;
}
options.publicKeyBuffer = (function(){
switch(options.coin.reward){
case 'POS':
return util.pubkeyToScript(addressInfo.pubkey);
case 'POW':
return util.addressToScript(addressInfo.address);
}
})();
callback(null);
});
},
function(callback){
_this.daemon.cmd('getinfo', [], function(results){
// Print which network each daemon is running on
var isTestnet;
var allValid = results.every(function(result){
if (result.error){
emitErrorLog('getinfo on init failed with daemon instance ' +
result.instance.index + ', error ' + JSON.stringify(result.error)
);
return false;
}
//Make sure every daemon is on the correct network or the config is wrong
if (typeof isTestnet === 'undefined'){
isTestnet = result.response.testnet;
return true;
}
else if (isTestnet !== result.response.testnet){
emitErrorLog('not all daemons are on same network');
return false;
}
else
return true;
});
if (!allValid){
callback('could not getinfo correctly on each daemon');
return;
}
//Find and return the response with the largest block height (most in-sync)
var infoResult = results.sort(function(a, b){
return b.response.blocks - a.response.blocks;
})[0].response;
options.testnet = infoResult.testnet;
options.initStats = { connections: infoResult.connections};
callback(null);
});
},
function(callback){
_this.daemon.cmd('getmininginfo', [], function(results){
var allValid = results.every(function(result){
if (result.error){
emitErrorLog('getmininginfo on init failed with daemon instance ' +
result.instance.index + ', error ' + JSON.stringify(result.error)
);
return false;
}
return true;
});
if (!allValid){
callback('could not getmininginfo correctly on each daemon');
return;
}
//Find and return the response with the largest block height (most in-sync)
var miningInfoResult = results.sort(function(a, b){
return b.response.blocks - a.response.blocks;
})[0].response;
options.initStats.networkHashRate = miningInfoResult.networkhashps;
callback(null);
});
},
function(callback){
/* This checks to see whether the daemon uses submitblock
or getblocktemplate for submitting new blocks */
_this.daemon.cmd('submitblock', [], function(results){
var couldNotDetectMethod = results.every(function(result){
if (result.error && result.error.message === 'Method not found'){
options.hasSubmitMethod = false;
callback(null);
return false;
}
else if (result.error && result.error.code === -1){
options.hasSubmitMethod = true;
callback(null);
return false;
}
else
return true;
});
if (couldNotDetectMethod){
emitErrorLog('Could not detect block submission RPC method, ' + JSON.stringify(results));
callback('block submission detection failed');
}
});
}
], function(err, results){
if (err){
emitErrorLog('Could not start pool, ' + JSON.stringify(err));
return;
}
finishedCallback();
});
}
function StartStratumServer(){
_this.stratumServer = new stratum.Server(options.ports, options.connectionTimeout, options.banning, authorizeFn);
@ -540,6 +619,8 @@ var pool = module.exports = function pool(options, authorizeFn){
});
}
function SetupBlockPolling(){
if (typeof options.blockRefreshInterval !== "number" || options.blockRefreshInterval <= 0){
@ -555,6 +636,8 @@ var pool = module.exports = function pool(options, authorizeFn){
emitLog('Block polling every ' + pollingInterval + ' milliseconds');
}
function GetBlockTemplate(callback){
_this.daemon.cmd('getblocktemplate',
[{"capabilities": [ "coinbasetxn", "workid", "coinbase/append" ]}],
@ -564,7 +647,7 @@ var pool = module.exports = function pool(options, authorizeFn){
result.instance.index + ' with error ' + JSON.stringify(result.error));
callback(result.error);
} else {
var processedNewBlock = _this.jobManager.processTemplate(result.response, publicKeyBuffer);
var processedNewBlock = _this.jobManager.processTemplate(result.response, options.publicKeyBuffer);
if (processedNewBlock) {
@ -581,6 +664,8 @@ var pool = module.exports = function pool(options, authorizeFn){
);
}
function CheckBlockAccepted(blockHash, callback){
_this.daemon.cmd('getblock',
[blockHash],
@ -600,6 +685,8 @@ var pool = module.exports = function pool(options, authorizeFn){
}
/**
* This method is being called from the blockNotify so that when a new block is discovered by the daemon
* We can inform our miners about the newly found block
@ -646,6 +733,7 @@ var pool = module.exports = function pool(options, authorizeFn){
)
};
this.attachMiners = function(miners) {
miners.forEach(function (clientObj) {
_this.stratumServer.manuallyAddStratumClient(clientObj);
@ -654,10 +742,12 @@ var pool = module.exports = function pool(options, authorizeFn){
};
this.getStratumServer = function() {
return _this.stratumServer;
};
this.setVarDiff = function(port, varDiffInstance) {
if (typeof(_this.varDiff[port]) != 'undefined' ) {
_this.varDiff[port].removeAllListeners();