Added clustering and moved blocknotify from stratum module to portal

README.md

@@ -1,3 +1,42 @@
+# Stratum Portal
+
+## Goal of this project
+When ready, this portal will be able to spawn pools for all configured coins/cryptocurrencies.
+Each pool will take advantage of clustering to load balance across multiple CPU cores and be
+extremely efficient.
+
+For reward/payment processing, shares will be inserted into a fast NoSQL database such as Redis.
+Each coin will have a processor that monitors for confirmed submitted blocks then send out payments
+according to shares accumulated in the database.
+
+For now the plan is to not have user accounts, but rather, have miners use their coin address for
+stratum authentication. This portal will come with a minimalistic HTML5 front-end that displays
+statistics from from each pool such as connected miners, network/pool difficulty/hash rate, etc.
+
+To reduce variance for pools just starting out which have little to no hashing power a feature
+could be added that connects upstream to a larger pool server. After receiving work from the larger
+pool it would then be redistributed to our connected miners.
+
+Another great feature would be utilizing the multi-pool ability of this portal to do coin
+auto-switching using an coin profitability API such as CoinChoose.com
+
+
+#### [Optional, recommended] Setting up blocknotify
+  * In `config.json` set the port and password for `blockNotifyListener`
+  * For the blocknotify arguments in your daemon startup parameters or conf file, use:
+
+    ```
+    [path to blockNotify.js]
+    [pool host]:[pool blockNotifyListener port]
+    [blockNotifyListener password]
+    [coin symbol set in coin's json config]
+    %s"
+    ```
+
+    * Example: `dogecoind -blocknotify="scripts/blockNotify.js localhost:8117 mySuperSecurePassword doge %s"`
+    * If your daemon is on a different host you will have to copy over `scripts/blockNotify.js`
+
+
 
 Setup for development of stratum-pool
 =====================================

config.json

@@ -1,6 +1,10 @@
 {
+    "clustering": {
+        "enabled": true,
+        "forks": "auto"
+    },
     "blockNotifyListener": {
-        "enabled": false,
+        "enabled": true,
         "port": 8117,
         "password": "test"
     }

init.js

@@ -1,11 +1,18 @@
 var fs = require('fs');
+var os = require('os');
+var cluster = require('cluster');
+
+
 var posix = require('posix');
 var Stratum = require('stratum-pool');
 var PoolLogger = require('./libs/logutils.js');
+var BlocknotifyListener = require('./libs/blocknotifyListener.js');
 
 JSON.minify = JSON.minify || require("node-json-minify");
 
 
+
+//Try to give process ability to handle 100k concurrent connections
 try{
     posix.setrlimit('nofile', { soft: 100000, hard: 100000 });
 }
@@ -13,6 +20,8 @@ catch(e){
     console.error(e);
 }
 
+
+
 var loggerInstance = new PoolLogger({
     'default': true,
     'keys': {
@@ -26,73 +35,146 @@ var logDebug   = loggerInstance.logDebug;
 var logWarning = loggerInstance.logWarning;
 var logError   = loggerInstance.logError;
 
-var config = JSON.parse(JSON.minify(fs.readFileSync("config.json", {encoding: 'utf8'})));
 
 
-var stratum = new Stratum(config);
-stratum.on('log', function(logText){
-    logDebug(logText);
-});
+if (cluster.isMaster){
 
 
-var coinProfiles = (function(){
-    var profiles = {};
-    fs.readdirSync('coins').forEach(function(file){
-        var coinProfile = JSON.parse(JSON.minify(fs.readFileSync('coins/' + file, {encoding: 'utf8'})));
-        profiles[coinProfile.name.toLowerCase()] = coinProfile;
-    });
-    return profiles;
-})();
+    var config = JSON.parse(JSON.minify(fs.readFileSync("config.json", {encoding: 'utf8'})));
 
-fs.readdirSync('pool_configs').forEach(function(file){
+    //Read all coin profile json files from coins directory and build object where key is name of coin
+    var coinProfiles = (function(){
+        var profiles = {};
+        fs.readdirSync('coins').forEach(function(file){
+            var coinProfile = JSON.parse(JSON.minify(fs.readFileSync('coins/' + file, {encoding: 'utf8'})));
+            profiles[coinProfile.name.toLowerCase()] = coinProfile;
+        });
+        return profiles;
+    })();
 
-    var poolOptions = JSON.parse(JSON.minify(fs.readFileSync('pool_configs/' + file, {encoding: 'utf8'})));
-    if (poolOptions.disabled) return;
 
-    if (!(poolOptions.coin.toLowerCase() in coinProfiles)){
-        logError(poolOptions.coin, 'system', 'could not find coin profile');
-        return;
+    //Read all pool configs from pool_configs and join them with their coin profile
+    var poolConfigs = (function(){
+        var configs = [];
+        fs.readdirSync('pool_configs').forEach(function(file){
+            var poolOptions = JSON.parse(JSON.minify(fs.readFileSync('pool_configs/' + file, {encoding: 'utf8'})));
+            if (poolOptions.disabled) return;
+            if (!(poolOptions.coin.toLowerCase() in coinProfiles)){
+                logError(poolOptions.coin, 'system', 'could not find coin profile');
+                return;
+            }
+            poolOptions.coin = coinProfiles[poolOptions.coin.toLowerCase()];
+            configs.push(poolOptions);
+        });
+        return configs;
+    })();
+
+
+    var serializedConfigs = JSON.stringify(poolConfigs);
+
+
+    var numForks = (function(){
+        if (!config.clustering || !config.clustering.enabled)
+            return 1;
+        if (config.clustering.forks === 'auto')
+            return os.cpus().length;
+        if (!config.clustering.forks || isNaN(config.clustering.forks))
+            return 1;
+        return config.clustering.forks;
+    })();
+
+    for (var i = 0; i < numForks; i++) {
+        cluster.fork({
+            fork: i,
+            pools: serializedConfigs
+        });
     }
 
-    poolOptions.coin = coinProfiles[poolOptions.coin.toLowerCase()];
+    cluster.on('exit', function(worker, code, signal) {
+        console.log('worker fork with PID ' + worker.process.pid + ' died');
+    });
 
-    var authorizeFN = function (ip, workerName, password, callback) {
-        // Default implementation just returns true
-        logDebug(poolOptions.coin.name, 'client', "Authorize ["+ip+"] "+workerName+":"+password);
-        callback({
-            error: null,
-            authorized: true,
-            disconnect: false
+
+    //block notify options
+    //setup block notify here and use IPC to tell appropriate pools
+    var listener = new BlocknotifyListener(config.blockNotifyListener);
+    listener.on('log', function(text){
+        logDebug('blocknotify', 'system', text);
+    });
+    listener.on('hash', function(message){
+
+        var serializedMessage = JSON.stringify({'blocknotify': message.hash});
+        Object.keys(cluster.workers).forEach(function(id) {
+            cluster.workers[id].send(serializedMessage);
         });
-    };
 
+    });
+    listener.start();
 
+}
 
-    var pool = stratum.createPool(poolOptions, authorizeFN);
-    pool.on('share', function(isValidShare, isValidBlock, data){
+else{
 
-        var shareData = JSON.stringify(data);
+    var poolConfigs = JSON.parse(process.env.pools);
+    var fork = process.env.fork;
 
-        if (data.solution && !isValidBlock)
-            logDebug(poolOptions.coin.name, 'client', 'We thought a block solution was found but it was rejected by the daemon, share data: ' + shareData);
-        else if (isValidBlock)
-            logDebug(poolOptions.coin.name, 'client', 'Block found, share data: ' + shareData);
-        else if (isValidShare)
-            logDebug(poolOptions.coin.name, 'client', 'Valid share submitted, share data: ' + shareData);
-        else
-            logDebug(poolOptions.coin.name, 'client', 'Invalid share submitted, share data: ' + shareData)
+    var stratum = new Stratum();
 
-
-    }).on('log', function(severity, logKey, logText) {
-        if (severity == 'debug') {
-            logDebug(poolOptions.coin.name, logKey, logText);
-        } else if (severity == 'warning') {
-            logWarning(poolOptions.coin.name, logKey, logText);
-        } else if (severity == 'error') {
-            logError(poolOptions.coin.name, logKey, logText);
+    //Handle blocknotify message from master process sent via IPC
+    process.on('message', function(msg) {
+        var message = JSON.parse(msg);
+        if (message.blocknotify){
+            for (var i = 0; i < stratum.pools.length; i++){
+                if (stratum.pools[i].options.coin.name.toLowerCase() === message.coin.toLowerCase()){
+                    stratum.pools[i].processBlockNotify(message.blockHash)
+                    return;
+                }
+            }
         }
     });
-    pool.start();
 
-});
 
+
+
+    poolConfigs.forEach(function(poolOptions){
+
+        var logIdentify = poolOptions.coin.name + ' (Fork ' + fork + ')';
+
+        var authorizeFN = function (ip, workerName, password, callback) {
+            // Default implementation just returns true
+            logDebug(logIdentify, 'client', "Authorize [" + ip + "] " + workerName + ":" + password);
+            callback({
+                error: null,
+                authorized: true,
+                disconnect: false
+            });
+        };
+
+
+        var pool = stratum.createPool(poolOptions, authorizeFN);
+        pool.on('share', function(isValidShare, isValidBlock, data){
+
+            var shareData = JSON.stringify(data);
+
+            if (data.solution && !isValidBlock)
+                logDebug(logIdentify, 'client', 'We thought a block solution was found but it was rejected by the daemon, share data: ' + shareData);
+            else if (isValidBlock)
+                logDebug(logIdentify, 'client', 'Block found, share data: ' + shareData);
+            else if (isValidShare)
+                logDebug(logIdentify, 'client', 'Valid share submitted, share data: ' + shareData);
+            else
+                logDebug(logIdentify, 'client', 'Invalid share submitted, share data: ' + shareData)
+
+
+        }).on('log', function(severity, logKey, logText) {
+                if (severity == 'debug') {
+                    logDebug(logIdentify, logKey, logText);
+                } else if (severity == 'warning') {
+                    logWarning(logIdentify, logKey, logText);
+                } else if (severity == 'error') {
+                    logError(logIdentify, logKey, logText);
+                }
+            });
+        pool.start();
+    });
+}

libs/blocknotifyListener.js

@@ -0,0 +1,53 @@
+var events = require('events');
+var net = require('net');
+
+var listener = module.exports = function listener(options){
+
+    var _this = this;
+
+    var emitLog = function(text){
+        _this.emit('log', text);
+    };
+
+
+    this.start = function(){
+        if (!options || !options.enabled){
+            emitLog('Blocknotify listener disabled');
+            return;
+        }
+
+        var blockNotifyServer = net.createServer(function(c) {
+            emitLog('Block listener has incoming connection');
+            var data = '';
+            c.on('data', function(d){
+                emitLog('Block listener received blocknotify data');
+                data += d;
+                if (data.slice(-1) === '\n'){
+                    c.end();
+                }
+            });
+            c.on('end', function() {
+
+                emitLog('Block listener connection ended');
+
+                var message = JSON.parse(data);
+                if (message.password === options.password){
+                    _this.emit('hash', message);
+                }
+                else
+                    emitLog('Block listener received notification with incorrect password');
+
+            });
+        });
+        blockNotifyServer.listen(options.port, function() {
+            emitLog('Block notify listener server started on port ' + options.port)
+        });
+
+        emitLog("Block listener is enabled, starting server on port " + options.port);
+    }
+
+
+
+};
+
+listener.prototype.__proto__ = events.EventEmitter.prototype;

scripts/blockNotify.js

@@ -0,0 +1,38 @@
+#!/usr/bin/env node
+/**
+ * This script should be hooked to the coin daemon as follow: 
+ * 
+ * litecoind -blocknotify="/path/to/this/script/blockNotify.js localhost:8117 password litecoin %s"
+ * 
+ * The above will send tell litecoin to launch this script with those parameters every time
+ * a block is found.
+ * This script will then send the blockhash along with other informations to a listening tcp socket 
+**/
+
+var net       = require('net');
+var config    = process.argv[1];
+var parts     = config.split(':');
+var host      = parts[0];
+var port      = parts[1];
+var password  = process.argv[2];
+var coin      = process.argv[3];
+var blockHash = process.argv[4];
+
+var client = net.connect(port, host, function() {
+    console.log('client connected');
+    client.write(JSON.stringify({
+        password: password,
+        coin: coin,
+        blockHash: blockHash
+    }) + '\n');
+});
+
+client.on('data', function(data) {
+    console.log(data.toString());
+    //client.end();
+});
+
+client.on('end', function() {
+    console.log('client disconnected');
+    //process.exit();
+});