Merge branch 'stratumsrv' into bfgminer

Makefile.am

@@ -83,6 +83,10 @@ endif
 
 bfgminer_SOURCES	+= logging.c
 
+if NEED_BFG_DRIVER_PROXY
+bfgminer_SOURCES += driver-proxy.c driver-proxy.h
+endif
+
 if USE_LIBMICROHTTPD
 bfgminer_SOURCES += httpsrv.c httpsrv.h driver-getwork.c
 bfgminer_LDADD += $(libmicrohttpd_LIBS)
@@ -90,6 +94,13 @@ bfgminer_LDFLAGS += $(libmicrohttpd_LDFLAGS)
 bfgminer_CPPFLAGS += $(libmicrohttpd_CFLAGS)
 endif
 
+if USE_LIBEVENT
+bfgminer_SOURCES  += driver-stratum.c
+bfgminer_LDADD    += $(libevent_LIBS)
+bfgminer_LDFLAGS  += $(libevent_LDFLAGS)
+bfgminer_CPPFLAGS += $(libevent_CFLAGS)
+endif
+
 
 # GPU sources, TODO: make them selectable
 # the GPU portion extracted from original main.c

README

@@ -233,6 +233,7 @@ Options for both config file and command line:
 --show-processors   Show per processor statistics in summary
 --skip-security-checks <arg> Skip security checks sometimes to save bandwidth; only check 1/<arg>th of the time (default: never skip)
 --socks-proxy <arg> Set socks4 proxy (host:port) for all pools without a proxy specified
+--stratum-port <arg> Port number to listen on for stratum miners (-1 means disabled) (default: -1)
 --submit-threads    Minimum number of concurrent share submissions (default: 64)
 --syslog            Use system log for output messages (default: standard error)
 --temp-cutoff <arg> Maximum temperature devices will be allowed to reach before being disabled, one value or comma separated list

api.c

@@ -3066,6 +3066,8 @@ static void debugstate(struct io_data *io_data, __maybe_unused SOCKETTYPE c, cha
 		io_close(io_data);
 }
 
+extern void stratumsrv_change_port();
+
 static void setconfig(struct io_data *io_data, __maybe_unused SOCKETTYPE c, char *param, bool isjson, __maybe_unused char group)
 {
 	char *comma;
@@ -3113,6 +3115,13 @@ static void setconfig(struct io_data *io_data, __maybe_unused SOCKETTYPE c, char
 		if (httpsrv_port != -1)
 			httpsrv_start(httpsrv_port);
 	}
+#endif
+#ifdef USE_LIBEVENT
+	else if (strcasecmp(param, "stratum-port") == 0)
+	{
+		stratumsrv_port = value;
+		stratumsrv_change_port();
+	}
 #endif
 	else {
 		message(io_data, MSG_UNKCON, 0, param, isjson);

configure.ac

@@ -350,6 +350,27 @@ if test "x$httpsrv" != "xno"; then
 fi
 AM_CONDITIONAL([USE_LIBMICROHTTPD], [test x$httpsrv = xyes])
 
+libevent=auto
+AC_ARG_WITH([libevent],
+	[AC_HELP_STRING([--without-libevent],[Compile support for libevent stratum server (default enabled)])],
+	[libevent=$withval]
+)
+if test "x$libevent" != "xno"; then
+	PKG_CHECK_MODULES([libevent],[libevent >= 2],[
+		AC_DEFINE([USE_LIBEVENT],[1],[Defined to 1 if libevent support is wanted])
+		libevent=yes
+	],[
+		if test "x$libevent" = "xyes"; then
+			AC_MSG_ERROR([Unable to find libevent 2])
+		else
+			AC_MSG_WARN([libevent 2 not found; stratum proxy will be unavailable])
+		fi
+	])
+fi
+AM_CONDITIONAL([USE_LIBEVENT], [test x$libevent = xyes])
+
+AM_CONDITIONAL([NEED_BFG_DRIVER_PROXY], [test x$libevent$httpsrv != xnono])
+
 AC_ARG_ENABLE([modminer],
 	[AC_HELP_STRING([--disable-modminer],[Compile support for ModMiner (default enabled)])],
 	[modminer=$enableval],

driver-getwork.c

@@ -26,79 +26,10 @@
 #include <uthash.h>
 
 #include "deviceapi.h"
+#include "driver-proxy.h"
 #include "httpsrv.h"
 #include "miner.h"
 
-struct device_drv getwork_drv;
-
-struct getwork_client {
-	char *username;
-	struct cgpu_info *cgpu;
-	struct work *work;
-	struct timeval tv_hashes_done;
-	
-	UT_hash_handle hh;
-};
-
-static
-struct getwork_client *getwork_clients;
-static
-pthread_mutex_t getwork_clients_mutex;
-
-static
-void prune_worklog()
-{
-	struct getwork_client *client, *tmp;
-	struct work *work, *tmp2;
-	struct timeval tv_now;
-	
-	timer_set_now(&tv_now);
-	
-	mutex_lock(&getwork_clients_mutex);
-	HASH_ITER(hh, getwork_clients, client, tmp)
-	{
-		HASH_ITER(hh, client->work, work, tmp2)
-		{
-			if (timer_elapsed(&work->tv_work_start, &tv_now) <= opt_expiry)
-				break;
-			HASH_DEL(client->work, work);
-			free_work(work);
-		}
-	}
-	mutex_unlock(&getwork_clients_mutex);
-}
-
-static
-pthread_t prune_worklog_pth;
-
-static
-void *prune_worklog_thread(void *userdata)
-{
-	struct cgpu_info *cgpu = userdata;
-	
-	pthread_detach(pthread_self());
-	RenameThread("SGW_pruner");
-	
-	while (!cgpu->shutdown)
-	{
-		prune_worklog();
-		sleep(60);
-	}
-	return NULL;
-}
-
-static
-void getwork_init()
-{
-	mutex_init(&getwork_clients_mutex);
-}
-
-static
-void getwork_first_client(struct cgpu_info *cgpu)
-{
-	pthread_create(&prune_worklog_pth, NULL, prune_worklog_thread, cgpu);
-}
-
 static
 void getwork_prepare_resp(struct MHD_Response *resp)
 {
@@ -129,8 +60,7 @@ int getwork_error(struct MHD_Connection *conn, int16_t errcode, const char *errm
 
 int handle_getwork(struct MHD_Connection *conn, bytes_t *upbuf)
 {
-	static bool _init = false, b;
-	struct getwork_client *client;
+	struct proxy_client *client;
 	struct MHD_Response *resp;
 	char *user, *idstr = NULL;
 	const char *submit = NULL;
@@ -144,12 +74,6 @@ int handle_getwork(struct MHD_Connection *conn, bytes_t *upbuf)
 	const char *hashesdone = NULL;
 	int ret;
 	
-	if (unlikely(!_init))
-	{
-		_init = true;
-		getwork_init();
-	}
-	
 	if (bytes_len(upbuf))
 	{
 		bytes_nullterminate(upbuf);
@@ -182,44 +106,14 @@ int handle_getwork(struct MHD_Connection *conn, bytes_t *upbuf)
 		goto out;
 	}
 	
-	mutex_lock(&getwork_clients_mutex);
-	HASH_FIND_STR(getwork_clients, user, client);
+	client = proxy_find_or_create_client(user);
+	free(user);
 	if (!client)
 	{
-		cgpu = malloc(sizeof(*cgpu));
-		client = malloc(sizeof(*client));
-		*cgpu = (struct cgpu_info){
-			.drv = &getwork_drv,
-			.threads = 0,
-			.device_data = client,
-			.device_path = user,
-		};
-		if (unlikely(!create_new_cgpus(add_cgpu_live, cgpu)))
-		{
-			free(client);
-			free(cgpu);
-			ret = getwork_error(conn, -32603, "Failed creating new cgpu", idstr, idstr_sz);
-			goto out;
-		}
-		*client = (struct getwork_client){
-			.username = user,
-			.cgpu = cgpu,
-		};
-		
-		b = HASH_COUNT(getwork_clients);
-		HASH_ADD_KEYPTR(hh, getwork_clients, client->username, strlen(user), client);
-		mutex_unlock(&getwork_clients_mutex);
-		
-		if (!b)
-			getwork_first_client(cgpu);
-	}
-	else
-	{
-		mutex_unlock(&getwork_clients_mutex);
-		free(user);
-		cgpu = client->cgpu;
+		ret = getwork_error(conn, -32603, "Failed creating new cgpu", idstr, idstr_sz);
+		goto out;
 	}
-	user = NULL;
+	cgpu = client->cgpu;
 	thr = cgpu->thr[0];
 	
 	hashesdone = MHD_lookup_connection_value(conn, MHD_HEADER_KIND, "X-Hashes-Done");
@@ -310,26 +204,8 @@ out:
 		hashes_done(thr, lld, &tv_delta, NULL);
 	}
 	
-	free(user);
 	free(idstr);
 	if (json)
 		json_decref(json);
 	return ret;
 }
-
-#ifdef HAVE_CURSES
-static
-void getwork_wlogprint_status(struct cgpu_info *cgpu)
-{
-	struct getwork_client *client = cgpu->device_data;
-	wlogprint("Username: %s\n", client->username);
-}
-#endif
-
-struct device_drv getwork_drv = {
-	.dname = "getwork",
-	.name = "SGW",
-#ifdef HAVE_CURSES
-	.proc_wlogprint_status = getwork_wlogprint_status,
-#endif
-};

driver-proxy.c

@@ -0,0 +1,129 @@
+#include "config.h"
+
+#include <pthread.h>
+
+#include <uthash.h>
+
+#include "deviceapi.h"
+#include "driver-proxy.h"
+#include "miner.h"
+#include "util.h"
+
+struct device_drv proxy_drv;
+
+static
+struct proxy_client *proxy_clients;
+static
+pthread_mutex_t proxy_clients_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+static
+void prune_worklog()
+{
+	struct proxy_client *client, *tmp;
+	struct work *work, *tmp2;
+	struct timeval tv_now;
+	
+	timer_set_now(&tv_now);
+	
+	mutex_lock(&proxy_clients_mutex);
+	HASH_ITER(hh, proxy_clients, client, tmp)
+	{
+		HASH_ITER(hh, client->work, work, tmp2)
+		{
+			if (timer_elapsed(&work->tv_work_start, &tv_now) <= opt_expiry)
+				break;
+			HASH_DEL(client->work, work);
+			free_work(work);
+		}
+	}
+	mutex_unlock(&proxy_clients_mutex);
+}
+
+static
+pthread_t prune_worklog_pth;
+
+static
+void *prune_worklog_thread(void *userdata)
+{
+	struct cgpu_info *cgpu = userdata;
+	
+	pthread_detach(pthread_self());
+	RenameThread("PXY_pruner");
+	
+	while (!cgpu->shutdown)
+	{
+		prune_worklog();
+		sleep(60);
+	}
+	return NULL;
+}
+
+static
+void proxy_first_client(struct cgpu_info *cgpu)
+{
+	pthread_create(&prune_worklog_pth, NULL, prune_worklog_thread, cgpu);
+}
+
+struct proxy_client *proxy_find_or_create_client(const char *username)
+{
+	struct proxy_client *client;
+	struct cgpu_info *cgpu;
+	char *user;
+	int b;
+	
+	if (!username)
+		return NULL;
+	
+	mutex_lock(&proxy_clients_mutex);
+	HASH_FIND_STR(proxy_clients, username, client);
+	if (!client)
+	{
+		user = strdup(username);
+		cgpu = malloc(sizeof(*cgpu));
+		client = malloc(sizeof(*client));
+		*cgpu = (struct cgpu_info){
+			.drv = &proxy_drv,
+			.threads = 0,
+			.device_data = client,
+			.device_path = user,
+		};
+		if (unlikely(!create_new_cgpus(add_cgpu_live, cgpu)))
+		{
+			free(client);
+			free(cgpu);
+			free(user);
+			return NULL;
+		}
+		*client = (struct proxy_client){
+			.username = user,
+			.cgpu = cgpu,
+		};
+		
+		b = HASH_COUNT(proxy_clients);
+		HASH_ADD_KEYPTR(hh, proxy_clients, client->username, strlen(user), client);
+		mutex_unlock(&proxy_clients_mutex);
+		
+		if (!b)
+			proxy_first_client(cgpu);
+	}
+	else
+		mutex_unlock(&proxy_clients_mutex);
+	return client;
+}
+
+#ifdef HAVE_CURSES
+static
+void proxy_wlogprint_status(struct cgpu_info *cgpu)
+{
+	struct proxy_client *client = cgpu->device_data;
+	wlogprint("Username: %s\n", client->username);
+}
+#endif
+
+struct device_drv proxy_drv = {
+	.dname = "proxy",
+	.name = "PXY",
+#ifdef HAVE_CURSES
+	.proc_wlogprint_status = proxy_wlogprint_status,
+#endif
+};

driver-proxy.h

@@ -0,0 +1,19 @@
+#ifndef BFG_DRIVER_PROXY_H
+#define BFG_DRIVER_PROXY_H
+
+#include <uthash.h>
+
+#include "miner.h"
+
+struct proxy_client {
+	char *username;
+	struct cgpu_info *cgpu;
+	struct work *work;
+	struct timeval tv_hashes_done;
+	
+	UT_hash_handle hh;
+};
+
+extern struct proxy_client *proxy_find_or_create_client(const char *user);
+
+#endif

driver-stratum.c

@@ -0,0 +1,639 @@
+#include "config.h"
+
+#ifndef WIN32
+#include <arpa/inet.h>
+#else
+#include <winsock2.h>
+#endif
+
+#include <stdbool.h>
+#include <stdint.h>
+#include <string.h>
+
+#include <event2/buffer.h>
+#include <event2/bufferevent.h>
+#include <event2/event.h>
+#include <event2/listener.h>
+
+#include <jansson.h>
+
+#include "deviceapi.h"
+#include "driver-proxy.h"
+#include "miner.h"
+#include "util.h"
+
+#define MAX_CLIENTS 255
+
+static bool _ssm_xnonce1s[MAX_CLIENTS + 1] = { true };
+static uint8_t _ssm_client_octets;
+static uint8_t _ssm_client_xnonce2sz;
+static char *_ssm_notify;
+static int _ssm_notify_sz;
+static struct event *ev_notify;
+static notifier_t _ssm_update_notifier;
+
+struct stratumsrv_job {
+	char *my_job_id;
+	
+	struct pool *pool;
+	uint8_t work_restart_id;
+	uint8_t n2size;
+	struct timeval tv_prepared;
+	struct stratum_work swork;
+	char *nonce1;
+	
+	UT_hash_handle hh;
+};
+
+static struct stratumsrv_job *_ssm_jobs;
+static struct work _ssm_cur_job_work;
+static uint64_t _ssm_jobid;
+
+static struct event_base *_smm_evbase;
+static bool _smm_running;
+static struct evconnlistener *_smm_listener;
+
+struct stratumsrv_conn {
+	struct bufferevent *bev;
+	uint32_t xnonce1_le;
+	struct timeval tv_hashes_done;
+	bool hashes_done_ext;
+	
+	struct stratumsrv_conn *next;
+};
+
+static struct stratumsrv_conn *_ssm_connections;
+
+static
+void _ssm_gen_dummy_work(struct work *work, struct stratumsrv_job *ssj, const char * const extranonce2, uint32_t xnonce1)
+{
+	uint8_t *p, *s;
+	
+	*work = (struct work){
+		.pool = ssj->pool,
+		.work_restart_id = ssj->work_restart_id,
+		.tv_staged = ssj->tv_prepared,
+	};
+	bytes_resize(&work->nonce2, ssj->n2size);
+	s = bytes_buf(&work->nonce2);
+	p = &s[ssj->n2size - _ssm_client_xnonce2sz];
+	if (extranonce2)
+		hex2bin(p, extranonce2, _ssm_client_xnonce2sz);
+#ifndef __OPTIMIZE__
+	else
+		memset(p, '\0', _ssm_client_xnonce2sz);
+#endif
+	p -= _ssm_client_octets;
+	memcpy(p, &xnonce1, _ssm_client_octets);
+	if (p != s)
+		memset(s, '\xbb', p - s);
+	gen_stratum_work2(work, &ssj->swork, ssj->nonce1);
+}
+
+static
+bool stratumsrv_update_notify_str(struct pool * const pool, bool clean)
+{
+	cg_rlock(&pool->data_lock);
+	
+	struct stratumsrv_conn *conn;
+	const struct stratum_work * const swork = &pool->swork;
+	const int n2size = pool->n2size;
+	char my_job_id[33];
+	int i;
+	struct stratumsrv_job *ssj;
+	ssize_t n2pad = n2size - _ssm_client_octets - _ssm_client_xnonce2sz;
+	if (n2pad < 0)
+		return false;
+	size_t coinb1in_lenx = swork->nonce2_offset * 2;
+	size_t n2padx = n2pad * 2;
+	size_t coinb1_lenx = coinb1in_lenx + n2padx;
+	size_t coinb2_len = bytes_len(&swork->coinbase) - swork->nonce2_offset - n2size;
+	size_t coinb2_lenx = coinb2_len * 2;
+	sprintf(my_job_id, "%"PRIx64"-%"PRIx64, (uint64_t)time(NULL), _ssm_jobid++);
+	size_t bufsz = 166 + strlen(my_job_id) + coinb1_lenx + coinb2_lenx + (swork->merkles * 67);
+	char * const buf = malloc(bufsz);
+	char *p = buf;
+	char prevhash[65], coinb1[coinb1_lenx + 1], coinb2[coinb2_lenx], version[9], nbits[9], ntime[9];
+	uint32_t ntime_n;
+	bin2hex(prevhash, &swork->header1[4], 32);
+	bin2hex(coinb1, bytes_buf(&swork->coinbase), swork->nonce2_offset);
+	memset(&coinb1[coinb1in_lenx], 'B', n2padx);
+	coinb1[coinb1_lenx] = '\0';
+	bin2hex(coinb2, &bytes_buf(&swork->coinbase)[swork->nonce2_offset + n2size], coinb2_len);
+	p += sprintf(p, "{\"params\":[\"%s\",\"%s\",\"%s\",\"%s\",[", my_job_id, prevhash, coinb1, coinb2);
+	for (i = 0; i < swork->merkles; ++i)
+	{
+		if (i)
+			*p++ = ',';
+		*p++ = '"';
+		bin2hex(p, &bytes_buf(&swork->merkle_bin)[i * 32], 32);
+		p += 64;
+		*p++ = '"';
+	}
+	bin2hex(version, swork->header1, 4);
+	bin2hex(nbits, swork->diffbits, 4);
+	ntime_n = htobe32(swork->ntime + timer_elapsed(&swork->tv_received, NULL));
+	bin2hex(ntime, &ntime_n, 4);
+	p += sprintf(p, "],\"%s\",\"%s\",\"%s\",%s],\"method\":\"mining.notify\",\"id\":null}\n", version, nbits, ntime, clean ? "true" : "false");
+	
+	ssj = malloc(sizeof(*ssj));
+	*ssj = (struct stratumsrv_job){
+		.my_job_id = strdup(my_job_id),
+		
+		.pool = pool,
+		.work_restart_id = pool->work_restart_id,
+		.n2size = n2size,
+		.nonce1 = strdup(pool->nonce1),
+	};
+	timer_set_now(&ssj->tv_prepared);
+	stratum_work_cpy(&ssj->swork, swork);
+	
+	cg_runlock(&pool->data_lock);
+	
+	ssj->swork.data_lock_p = NULL;
+	HASH_ADD_KEYPTR(hh, _ssm_jobs, ssj->my_job_id, strlen(ssj->my_job_id), ssj);
+	_ssm_gen_dummy_work(&_ssm_cur_job_work, ssj, NULL, 0);
+	
+	_ssm_notify_sz = p - buf;
+	assert(_ssm_notify_sz <= bufsz);
+	_ssm_notify = buf;
+	
+	LL_FOREACH(_ssm_connections, conn)
+	{
+		if (unlikely(!conn->xnonce1_le))
+			continue;
+		bufferevent_write(conn->bev, _ssm_notify, _ssm_notify_sz);
+	}
+	
+	return true;
+}
+
+static
+void _ssj_free(struct stratumsrv_job * const ssj)
+{
+	free(ssj->my_job_id);
+	stratum_work_clean(&ssj->swork);
+	free(ssj->nonce1);
+	free(ssj);
+}
+
+static
+void stratumsrv_job_pruner()
+{
+	struct stratumsrv_job *ssj, *tmp_ssj;
+	struct timeval tv_now;
+	
+	timer_set_now(&tv_now);
+	
+	HASH_ITER(hh, _ssm_jobs, ssj, tmp_ssj)
+	{
+		if (timer_elapsed(&ssj->tv_prepared, &tv_now) <= opt_expiry)
+			break;
+		HASH_DEL(_ssm_jobs, ssj);
+		applog(LOG_DEBUG, "SSM: Pruning job_id %s", ssj->my_job_id);
+		_ssj_free(ssj);
+	}
+}
+
+static void stratumsrv_client_close(struct stratumsrv_conn *);
+
+static
+void stratumsrv_conn_close_completion_cb(struct bufferevent *bev, void *p)
+{
+	struct evbuffer * const output = bufferevent_get_output(bev);
+	if (evbuffer_get_length(output))
+		// Still have more data to write...
+		return;
+	stratumsrv_client_close(p);
+}
+
+static void stratumsrv_event(struct bufferevent *, short, void *);
+
+static
+void stratumsrv_boot(struct stratumsrv_conn * const conn, const char * const msg)
+{
+	struct bufferevent * const bev = conn->bev;
+	char buf[58 + strlen(msg)];
+	int bufsz = sprintf(buf, "{\"params\":[\"%s\"],\"method\":\"client.show_message\",\"id\":null}\n", msg);
+	bufferevent_write(bev, buf, bufsz);
+	bufferevent_setcb(bev, NULL, stratumsrv_conn_close_completion_cb, stratumsrv_event, conn);
+}
+
+static
+void stratumsrv_boot_all_subscribed(const char * const msg)
+{
+	struct stratumsrv_conn *conn, *tmp_conn;
+	
+	_ssm_notify = NULL;
+	
+	// Boot all connections
+	LL_FOREACH_SAFE(_ssm_connections, conn, tmp_conn)
+	{
+		if (!conn->xnonce1_le)
+			continue;
+		stratumsrv_boot(conn, msg);
+	}
+}
+
+static
+void _stratumsrv_update_notify(evutil_socket_t fd, short what, __maybe_unused void *p)
+{
+	struct pool *pool = current_pool();
+	bool clean;
+	
+	if (fd == _ssm_update_notifier[0])
+	{
+		evtimer_del(ev_notify);
+		notifier_read(_ssm_update_notifier);
+		applog(LOG_DEBUG, "SSM: Update triggered by notifier");
+	}
+	
+	clean = _ssm_cur_job_work.pool ? stale_work(&_ssm_cur_job_work, true) : true;
+	if (clean)
+	{
+		struct stratumsrv_job *ssj, *tmp;
+		
+		applog(LOG_DEBUG, "SSM: Current replacing job stale, pruning all jobs");
+		HASH_ITER(hh, _ssm_jobs, ssj, tmp)
+		{
+			HASH_DEL(_ssm_jobs, ssj);
+			_ssj_free(ssj);
+		}
+	}
+	else
+		stratumsrv_job_pruner();
+	
+	if (!pool->stratum_notify)
+	{
+		applog(LOG_WARNING, "SSM: Not using a stratum server upstream!");
+		if (clean)
+			stratumsrv_boot_all_subscribed("Current upstream pool does not have active stratum");
+		goto out;
+	}
+	
+	if (!stratumsrv_update_notify_str(pool, clean))
+	{
+		applog(LOG_WARNING, "SSM: Failed to subdivide upstream stratum notify!");
+		if (clean)
+			stratumsrv_boot_all_subscribed("Current upstream pool does not have active stratum");
+	}
+	
+out: ;
+	struct timeval tv_scantime = {
+		.tv_sec = opt_scantime,
+	};
+	evtimer_add(ev_notify, &tv_scantime);
+}
+
+static struct proxy_client *_stratumsrv_find_or_create_client(const char *);
+
+static
+struct proxy_client *(*stratumsrv_find_or_create_client)(const char *) = _stratumsrv_find_or_create_client;
+
+static
+struct proxy_client *_stratumsrv_find_or_create_client(const char *user)
+{
+	struct proxy_client * const client = proxy_find_or_create_client(user);
+	struct cgpu_info *cgpu;
+	struct thr_info *thr;
+	
+	if (!client)
+		return NULL;
+	
+	cgpu = client->cgpu;
+	thr = cgpu->thr[0];
+	
+	memcpy(thr->work_restart_notifier, _ssm_update_notifier, sizeof(thr->work_restart_notifier));
+	stratumsrv_find_or_create_client = proxy_find_or_create_client;
+	
+	return client;
+}
+
+static
+void _stratumsrv_failure(struct bufferevent * const bev, const char * const idstr, const int e, const char * const emsg)
+{
+	if (!idstr)
+		return;
+	
+	char buf[0x100];
+	size_t bufsz = snprintf(buf, sizeof(buf), "{\"error\":[%d,\"%s\",null],\"id\":%s,\"result\":null}\n", e, emsg, idstr);
+	bufferevent_write(bev, buf, bufsz);
+}
+#define return_stratumsrv_failure(e, emsg)  do{  \
+	_stratumsrv_failure(bev, idstr, e, emsg);    \
+	return;                                      \
+}while(0)
+
+static
+void _stratumsrv_success(struct bufferevent * const bev, const char * const idstr)
+{
+	if (!idstr)
+		return;
+	
+	size_t bufsz = 36 + strlen(idstr);
+	char buf[bufsz];
+	
+	bufsz = sprintf(buf, "{\"result\":true,\"id\":%s,\"error\":null}\n", idstr);
+	bufferevent_write(bev, buf, bufsz);
+}
+
+static
+void stratumsrv_mining_subscribe(struct bufferevent *bev, json_t *params, const char *idstr, uint32_t *xnonce1_p)
+{
+	char buf[90 + strlen(idstr) + (_ssm_client_octets * 2 * 2) + 0x10];
+	char xnonce1x[(_ssm_client_octets * 2) + 1];
+	int bufsz;
+	
+	if (!_ssm_notify)
+	{
+		evtimer_del(ev_notify);
+		_stratumsrv_update_notify(-1, 0, NULL);
+		if (!_ssm_notify)
+			return_stratumsrv_failure(20, "No notify set (upstream not stratum?)");
+	}
+	
+	if (!*xnonce1_p)
+	{
+		uint32_t xnonce1;
+		for (xnonce1 = MAX_CLIENTS; _ssm_xnonce1s[xnonce1]; --xnonce1)
+			if (!xnonce1)
+				return_stratumsrv_failure(20, "Maximum clients already connected");
+		*xnonce1_p = htole32(xnonce1);
+	}
+	
+	bin2hex(xnonce1x, xnonce1_p, _ssm_client_octets);
+	bufsz = sprintf(buf, "{\"id\":%s,\"result\":[[[\"mining.set_difficulty\",\"x\"],[\"mining.notify\",\"%s\"]],\"%s\",%d],\"error\":null}\n", idstr, xnonce1x, xnonce1x, _ssm_client_xnonce2sz);
+	bufferevent_write(bev, buf, bufsz);
+	bufferevent_write(bev, "{\"params\":[0.9999847412109375],\"id\":null,\"method\":\"mining.set_difficulty\"}\n", 75);
+	bufferevent_write(bev, _ssm_notify, _ssm_notify_sz);
+}
+
+static
+void stratumsrv_mining_authorize(struct bufferevent *bev, json_t *params, const char *idstr, uint32_t *xnonce1_p)
+{
+	struct proxy_client * const client = stratumsrv_find_or_create_client(__json_array_string(params, 0));
+	
+	if (unlikely(!client))
+		return_stratumsrv_failure(20, "Failed creating new cgpu");
+	
+	_stratumsrv_success(bev, idstr);
+}
+
+static
+void stratumsrv_mining_submit(struct bufferevent *bev, json_t *params, const char *idstr, struct stratumsrv_conn * const conn)
+{
+	uint32_t * const xnonce1_p = &conn->xnonce1_le;
+	struct work _work, *work;
+	struct stratumsrv_job *ssj;
+	struct proxy_client *client = stratumsrv_find_or_create_client(__json_array_string(params, 0));
+	struct cgpu_info *cgpu;
+	struct thr_info *thr;
+	const char * const job_id = __json_array_string(params, 1);
+	const char * const extranonce2 = __json_array_string(params, 2);
+	const char * const ntime = __json_array_string(params, 3);
+	const char * const nonce = __json_array_string(params, 4);
+	uint32_t nonce_n;
+	
+	if (unlikely(!client))
+		return_stratumsrv_failure(20, "Failed creating new cgpu");
+	if (unlikely(!(job_id && extranonce2 && ntime && nonce)))
+		return_stratumsrv_failure(20, "Couldn't understand parameters");
+	if (unlikely(strlen(nonce) < 8))
+		return_stratumsrv_failure(20, "nonce too short");
+	if (unlikely(strlen(ntime) < 8))
+		return_stratumsrv_failure(20, "ntime too short");
+	if (unlikely(strlen(extranonce2) < _ssm_client_xnonce2sz * 2))
+		return_stratumsrv_failure(20, "extranonce2 too short");
+	
+	cgpu = client->cgpu;
+	thr = cgpu->thr[0];
+	
+	// Lookup job_id
+	HASH_FIND_STR(_ssm_jobs, job_id, ssj);
+	if (!ssj)
+		return_stratumsrv_failure(21, "Job not found");
+	
+	// Generate dummy work
+	work = &_work;
+	_ssm_gen_dummy_work(work, ssj, extranonce2, *xnonce1_p);
+	
+	// Submit nonce
+	hex2bin(&work->data[68], ntime, 4);
+	hex2bin((void*)&nonce_n, nonce, 4);
+	nonce_n = le32toh(nonce_n);
+	if (!submit_nonce(thr, work, nonce_n))
+		_stratumsrv_failure(bev, idstr, 23, "H-not-zero");
+	else
+	if (stale_work(work, true))
+		_stratumsrv_failure(bev, idstr, 21, "stale");
+	else
+		_stratumsrv_success(bev, idstr);
+	
+	clean_work(work);
+	
+	if (!conn->hashes_done_ext)
+	{
+		struct timeval tv_now, tv_delta;
+		timer_set_now(&tv_now);
+		timersub(&tv_now, &conn->tv_hashes_done, &tv_delta);
+		conn->tv_hashes_done = tv_now;
+		hashes_done(thr, 0x100000000, &tv_delta, NULL);
+	}
+}
+
+static
+void stratumsrv_mining_hashes_done(struct bufferevent * const bev, json_t * const params, const char * const idstr, struct stratumsrv_conn * const conn)
+{
+	double f;
+	struct timeval tv_delta;
+	struct cgpu_info *cgpu;
+	struct thr_info *thr;
+	struct proxy_client * const client = stratumsrv_find_or_create_client(__json_array_string(params, 0));
+	json_t *jduration = json_array_get(params, 1);
+	json_t *jhashcount = json_array_get(params, 2);
+	
+	if (!(json_is_number(jduration) && json_is_number(jhashcount)))
+		return_stratumsrv_failure(20, "mining.hashes_done(String username, Number duration-in-seconds, Number hashcount)");
+	
+	cgpu = client->cgpu;
+	thr = cgpu->thr[0];
+	
+	f = json_number_value(jduration);
+	tv_delta.tv_sec = f;
+	tv_delta.tv_usec = (f - tv_delta.tv_sec) * 1e6;
+	
+	f = json_number_value(jhashcount);
+	hashes_done(thr, f, &tv_delta, NULL);
+	
+	conn->hashes_done_ext = true;
+}
+
+static
+bool stratumsrv_process_line(struct bufferevent * const bev, const char * const ln, void * const p)
+{
+	struct stratumsrv_conn *conn = p;
+	json_error_t jerr;
+	json_t *json, *params, *j2;
+	const char *method;
+	char *idstr;
+	
+	json = JSON_LOADS(ln, &jerr);
+	if (!json)
+	{
+		applog(LOG_ERR, "SSM: JSON parse error: %s", ln);
+		return false;
+	}
+	
+	method = bfg_json_obj_string(json, "method", NULL);
+	if (!method)
+	{
+		applog(LOG_ERR, "SSM: JSON missing method: %s", ln);
+		return false;
+	}
+	
+	params = json_object_get(json, "params");
+	if (!params)
+	{
+		applog(LOG_ERR, "SSM: JSON missing params: %s", ln);
+		return false;
+	}
+	
+	applog(LOG_DEBUG, "SSM: RECV: %s", ln);
+	
+	j2 = json_object_get(json, "id");
+	idstr = (j2 && !json_is_null(j2)) ? json_dumps_ANY(j2, 0) : NULL;
+	
+	if (!strcasecmp(method, "mining.submit"))
+		stratumsrv_mining_submit(bev, params, idstr, conn);
+	else
+	if (!strcasecmp(method, "mining.hashes_done"))
+		stratumsrv_mining_hashes_done(bev, params, idstr, conn);
+	else
+	if (!strcasecmp(method, "mining.authorize"))
+		stratumsrv_mining_authorize(bev, params, idstr, &conn->xnonce1_le);
+	else
+	if (!strcasecmp(method, "mining.subscribe"))
+		stratumsrv_mining_subscribe(bev, params, idstr, &conn->xnonce1_le);
+	else
+		_stratumsrv_failure(bev, idstr, -3, "Method not supported");
+	
+	free(idstr);
+	return true;
+}
+
+static
+void stratumsrv_client_close(struct stratumsrv_conn * const conn)
+{
+	struct bufferevent * const bev = conn->bev;
+	
+	bufferevent_free(bev);
+	LL_DELETE(_ssm_connections, conn);
+	free(conn);
+}
+
+static
+void stratumsrv_read(struct bufferevent *bev, void *p)
+{
+	struct evbuffer *input = bufferevent_get_input(bev);
+	char *ln;
+	bool rv;
+	
+	while ( (ln = evbuffer_readln(input, NULL, EVBUFFER_EOL_ANY)) )
+	{
+		rv = stratumsrv_process_line(bev, ln, p);
+		free(ln);
+		if (unlikely(!rv))
+		{
+			stratumsrv_client_close(p);
+			break;
+		}
+	}
+}
+
+static
+void stratumsrv_event(struct bufferevent *bev, short events, void *p)
+{
+	if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR))
+	{
+		if (events & BEV_EVENT_ERROR)
+			applog(LOG_ERR, "Error from bufferevent");
+		if (events & BEV_EVENT_EOF)
+			applog(LOG_DEBUG, "EOF from bufferevent");
+		stratumsrv_client_close(p);
+	}
+}
+
+static
+void stratumlistener(struct evconnlistener *listener, evutil_socket_t sock, struct sockaddr *addr, int len, void *p)
+{
+	struct stratumsrv_conn *conn;
+	struct event_base *evbase = evconnlistener_get_base(listener);
+	struct bufferevent *bev = bufferevent_socket_new(evbase, sock, BEV_OPT_CLOSE_ON_FREE);
+	conn = malloc(sizeof(*conn));
+	*conn = (struct stratumsrv_conn){
+		.bev = bev,
+	};
+	LL_PREPEND(_ssm_connections, conn);
+	bufferevent_setcb(bev, stratumsrv_read, NULL, stratumsrv_event, conn);
+	bufferevent_enable(bev, EV_READ | EV_WRITE);
+}
+
+void stratumsrv_start();
+
+void stratumsrv_change_port()
+{
+	struct event_base * const evbase = _smm_evbase;
+	
+	if (_smm_listener)
+		evconnlistener_free(_smm_listener);
+	
+	if (!_smm_running)
+	{
+		stratumsrv_start();
+		return;
+	}
+	
+	struct sockaddr_in sin = {
+		.sin_family = AF_INET,
+		.sin_addr.s_addr = INADDR_ANY,
+		.sin_port = htons(stratumsrv_port),
+	};
+	_smm_listener = evconnlistener_new_bind(evbase, stratumlistener, NULL, (
+		LEV_OPT_CLOSE_ON_FREE | LEV_OPT_CLOSE_ON_EXEC | LEV_OPT_REUSEABLE
+	), 0x10, (void*)&sin, sizeof(sin));
+}
+
+static
+void *stratumsrv_thread(__maybe_unused void *p)
+{
+	pthread_detach(pthread_self());
+	RenameThread("stratumsrv");
+	
+	for (uint64_t n = MAX_CLIENTS; n; n >>= 8)
+		++_ssm_client_octets;
+	_ssm_client_xnonce2sz = 2;
+	
+	struct event_base *evbase = event_base_new();
+	_smm_evbase = evbase;
+	{
+		ev_notify = evtimer_new(evbase, _stratumsrv_update_notify, NULL);
+		_stratumsrv_update_notify(-1, 0, NULL);
+	}
+	{
+		notifier_init(_ssm_update_notifier);
+		struct event *ev_update_notifier = event_new(evbase, _ssm_update_notifier[0], EV_READ | EV_PERSIST, _stratumsrv_update_notify, NULL);
+		event_add(ev_update_notifier, NULL);
+	}
+	stratumsrv_change_port();
+	event_base_dispatch(evbase);
+	
+	return NULL;
+}
+
+void stratumsrv_start()
+{
+	_smm_running = true;
+	pthread_t pth;
+	if (unlikely(pthread_create(&pth, NULL, stratumsrv_thread, NULL)))
+		quit(1, "stratumsrv thread create failed");
+}

miner.c

@@ -163,6 +163,9 @@ static bool opt_nogpu;
 #include "httpsrv.h"
 int httpsrv_port = -1;
 #endif
+#ifdef USE_LIBEVENT
+int stratumsrv_port = -1;
+#endif
 
 struct string_elist *scan_devices;
 bool opt_force_dev_init;
@@ -1792,6 +1795,11 @@ static struct opt_table opt_config_table[] = {
 	OPT_WITH_ARG("--socks-proxy",
 		     opt_set_charp, NULL, &opt_socks_proxy,
 		     "Set socks4 proxy (host:port)"),
+#ifdef USE_LIBEVENT
+	OPT_WITH_ARG("--stratum-port",
+	             opt_set_intval, opt_show_intval, &stratumsrv_port,
+	             "Port number to listen on for stratum miners (-1 means disabled)"),
+#endif
 	OPT_WITHOUT_ARG("--submit-stale",
 			opt_set_bool, &opt_submit_stale,
 	                opt_hidden),
@@ -7918,23 +7926,31 @@ void set_target(unsigned char *dest_target, double diff)
 	}
 }
 
+void stratum_work_cpy(struct stratum_work * const dst, const struct stratum_work * const src)
+{
+	*dst = *src;
+	dst->job_id = strdup(src->job_id);
+	bytes_cpy(&dst->coinbase, &src->coinbase);
+	bytes_cpy(&dst->merkle_bin, &src->merkle_bin);
+}
+
+void stratum_work_clean(struct stratum_work * const swork)
+{
+	free(swork->job_id);
+	bytes_free(&swork->coinbase);
+	bytes_free(&swork->merkle_bin);
+}
+
 /* Generates stratum based work based on the most recent notify information
  * from the pool. This will keep generating work while a pool is down so we use
  * other means to detect when the pool has died in stratum_thread */
 static void gen_stratum_work(struct pool *pool, struct work *work)
 {
-	unsigned char *coinbase, merkle_root[32], merkle_sha[64];
-	uint8_t *merkle_bin;
-	uint32_t *data32, *swap32;
-	int i;
-	
-	coinbase = bytes_buf(&pool->swork.coinbase);
-
 	clean_work(work);
-
+	
 	cg_wlock(&pool->data_lock);
-
-	/* Generate coinbase */
+	pool->swork.data_lock_p = &pool->data_lock;
+	
 	bytes_resize(&work->nonce2, pool->n2size);
 #ifndef __OPTIMIZE__
 	if (pool->nonce2sz < pool->n2size)
@@ -7948,17 +7964,35 @@ static void gen_stratum_work(struct pool *pool, struct work *work)
 	       &pool->nonce2,
 #endif
 	       pool->nonce2sz);
-	memcpy(&coinbase[pool->swork.nonce2_offset], bytes_buf(&work->nonce2), pool->n2size);
 	pool->nonce2++;
+	
+	work->pool = pool;
+	work->work_restart_id = work->pool->work_restart_id;
+	gen_stratum_work2(work, &pool->swork, pool->nonce1);
+	
+	cgtime(&work->tv_staged);
+}
 
-	/* Downgrade to a read lock to read off the pool variables */
-	cg_dwlock(&pool->data_lock);
+void gen_stratum_work2(struct work *work, struct stratum_work *swork, const char *nonce1)
+{
+	unsigned char *coinbase, merkle_root[32], merkle_sha[64];
+	uint8_t *merkle_bin;
+	uint32_t *data32, *swap32;
+	int i;
+
+	/* Generate coinbase */
+	coinbase = bytes_buf(&swork->coinbase);
+	memcpy(&coinbase[swork->nonce2_offset], bytes_buf(&work->nonce2), bytes_len(&work->nonce2));
+
+	/* Downgrade to a read lock to read off the variables */
+	if (swork->data_lock_p)
+		cg_dwlock(swork->data_lock_p);
 
 	/* Generate merkle root */
-	gen_hash(coinbase, merkle_root, bytes_len(&pool->swork.coinbase));
+	gen_hash(coinbase, merkle_root, bytes_len(&swork->coinbase));
 	memcpy(merkle_sha, merkle_root, 32);
-	merkle_bin = bytes_buf(&pool->swork.merkle_bin);
-	for (i = 0; i < pool->swork.merkles; ++i, merkle_bin += 32) {
+	merkle_bin = bytes_buf(&swork->merkle_bin);
+	for (i = 0; i < swork->merkles; ++i, merkle_bin += 32) {
 		memcpy(merkle_sha + 32, merkle_bin, 32);
 		gen_hash(merkle_sha, merkle_root, 64);
 		memcpy(merkle_sha, merkle_root, 32);
@@ -7967,21 +8001,22 @@ static void gen_stratum_work(struct pool *pool, struct work *work)
 	swap32 = (uint32_t *)merkle_root;
 	flip32(swap32, data32);
 	
-	memcpy(&work->data[0], pool->swork.header1, 36);
+	memcpy(&work->data[0], swork->header1, 36);
 	memcpy(&work->data[36], merkle_root, 32);
-	*((uint32_t*)&work->data[68]) = htobe32(pool->swork.ntime + timer_elapsed(&pool->swork.tv_received, NULL));
-	memcpy(&work->data[72], pool->swork.diffbits, 4);
+	*((uint32_t*)&work->data[68]) = htobe32(swork->ntime + timer_elapsed(&swork->tv_received, NULL));
+	memcpy(&work->data[72], swork->diffbits, 4);
 	memset(&work->data[76], 0, 4);  // nonce
 	memcpy(&work->data[80], workpadding_bin, 48);
 
 	/* Store the stratum work diff to check it still matches the pool's
 	 * stratum diff when submitting shares */
-	work->sdiff = pool->swork.diff;
+	work->sdiff = swork->diff;
 
 	/* Copy parameters required for share submission */
-	work->job_id = strdup(pool->swork.job_id);
-	work->nonce1 = strdup(pool->nonce1);
-	cg_runlock(&pool->data_lock);
+	work->job_id = strdup(swork->job_id);
+	work->nonce1 = strdup(nonce1);
+	if (swork->data_lock_p)
+		cg_runlock(swork->data_lock_p);
 
 	if (opt_debug)
 	{
@@ -7998,16 +8033,12 @@ static void gen_stratum_work(struct pool *pool, struct work *work)
 	set_target(work->target, work->sdiff);
 
 	local_work++;
-	work->pool = pool;
 	work->stratum = true;
 	work->blk.nonce = 0;
 	work->id = total_work++;
 	work->longpoll = false;
 	work->getwork_mode = GETWORK_MODE_STRATUM;
-	work->work_restart_id = work->pool->work_restart_id;
 	calc_diff(work, 0);
-
-	cgtime(&work->tv_staged);
 }
 
 void request_work(struct thr_info *thr)
@@ -9882,6 +9913,7 @@ static void raise_fd_limits(void)
 }
 
 extern void bfg_init_threadlocal();
+extern void stratumsrv_start();
 
 int main(int argc, char *argv[])
 {
@@ -10169,17 +10201,22 @@ int main(int argc, char *argv[])
 	}
 
 	if (!total_devices) {
-#ifndef USE_LIBMICROHTTPD
-		const int httpsrv_port = -1;
+		const bool netdev_support =
+#ifdef USE_LIBMICROHTTPD
+			(httpsrv_port != -1) ||
+#endif
+#ifdef USE_LIBEVENT
+			(stratumsrv_port != -1) ||
 #endif
-		if (httpsrv_port == -1 && (!use_curses) && !opt_api_listen)
+			false;
+		if ((!netdev_support) && (!use_curses) && !opt_api_listen)
 			quit(1, "All devices disabled, cannot mine!");
 		applog(LOG_WARNING, "No devices detected!");
 		if (use_curses)
 			applog(LOG_WARNING, "Waiting for devices; press 'M+' to add, or 'Q' to quit");
 		else
 			applog(LOG_WARNING, "Waiting for %s or press Ctrl-C to quit",
-		       (httpsrv_port == -1) ? "RPC commands" : "network devices");
+		       netdev_support ? "network devices" : "RPC commands");
 	}
 
 	load_temp_config();
@@ -10409,6 +10446,11 @@ begin_bench:
 		httpsrv_start(httpsrv_port);
 #endif
 
+#ifdef USE_LIBEVENT
+	if (stratumsrv_port != -1)
+		stratumsrv_start();
+#endif
+
 #ifdef HAVE_CURSES
 	/* Create curses input thread for keyboard input. Create this last so
 	 * that we know all threads are created since this can call kill_work

miner.h

@@ -891,6 +891,7 @@ extern bool opt_autofan;
 extern bool opt_autoengine;
 extern bool use_curses;
 extern int httpsrv_port;
+extern int stratumsrv_port;
 extern char *opt_api_allow;
 extern bool opt_api_mcast;
 extern char *opt_api_mcast_addr;
@@ -1151,6 +1152,8 @@ struct stratum_work {
 	bool transparency_probed;
 	struct timeval tv_transparency;
 	bool opaque;
+	
+	cglock_t *data_lock_p;
 };
 
 #define RBUFSIZE 8192
@@ -1329,6 +1332,9 @@ struct work {
 
 extern void get_datestamp(char *, size_t, time_t);
 #define get_now_datestamp(buf, bufsz)  get_datestamp(buf, bufsz, INVALID_TIMESTAMP)
+extern void stratum_work_cpy(struct stratum_work *dst, const struct stratum_work *src);
+extern void stratum_work_clean(struct stratum_work *);
+extern void gen_stratum_work2(struct work *, struct stratum_work *, const char *nonce1);
 extern void inc_hw_errors2(struct thr_info *thr, const struct work *work, const uint32_t *bad_nonce_p);
 extern void inc_hw_errors(struct thr_info *, const struct work *, const uint32_t bad_nonce);
 #define inc_hw_errors_only(thr)  inc_hw_errors(thr, NULL, 0)

util.c

@@ -1790,8 +1790,7 @@ static bool parse_notify(struct pool *pool, json_t *val)
 	for (i = 0; i < merkles; i++)
 		hex2bin(&bytes_buf(&pool->swork.merkle_bin)[i * 32], json_string_value(json_array_get(arr, i)), 32);
 	pool->swork.merkles = merkles;
-	if (clean)
-		pool->nonce2 = 0;
+	pool->nonce2 = 0;
 	cg_wunlock(&pool->data_lock);
 
 	applog(LOG_DEBUG, "Received stratum notify from pool %u with job_id=%s",