Add Stratum support

configure.ac

@@ -102,8 +102,8 @@ else
 	JANSSON_LIBS=-ljansson
 fi
 
-LIBCURL_CHECK_CONFIG(, 7.10.1, ,
-  [AC_MSG_ERROR([Missing required libcurl >= 7.10.1])])
+LIBCURL_CHECK_CONFIG(, 7.15.2, ,
+  [AC_MSG_ERROR([Missing required libcurl >= 7.15.2])])
 
 AC_SUBST(JANSSON_LIBS)
 AC_SUBST(PTHREAD_FLAGS)

cpu-miner.c

@@ -1,5 +1,6 @@
 /*
- * Copyright 2010 Jeff Garzik, 2012 pooler
+ * Copyright 2010 Jeff Garzik
+ * Copyright 2012-2013 pooler
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the Free
@@ -116,6 +117,8 @@ bool opt_protocol = false;
 static bool opt_benchmark = false;
 bool want_longpoll = true;
 bool have_longpoll = false;
+bool want_stratum = true;
+bool have_stratum = false;
 static bool submit_old = false;
 bool use_syslog = false;
 static bool opt_background = false;
@@ -136,8 +139,11 @@ char *opt_proxy;
 long opt_proxy_type;
 struct thr_info *thr_info;
 static int work_thr_id;
-int longpoll_thr_id;
+int longpoll_thr_id = -1;
+int stratum_thr_id = -1;
 struct work_restart *work_restart = NULL;
+static struct stratum_ctx stratum;
+
 pthread_mutex_t applog_lock;
 pthread_mutex_t stats_lock;
 
@@ -175,6 +181,7 @@ Options:\n\
   -s, --scantime=N      upper bound on time spent scanning current work when\n\
                           long polling is unavailable, in seconds (default: 5)\n\
       --no-longpoll     disable X-Long-Polling support\n\
+      --no-stratum      disable X-Stratum support\n\
   -q, --quiet           disable per-thread hashmeter output\n\
   -D, --debug           enable debug output\n\
   -P, --protocol-dump   verbose dump of protocol-level activities\n"
@@ -212,6 +219,7 @@ static struct option const options[] = {
 	{ "debug", 0, NULL, 'D' },
 	{ "help", 0, NULL, 'h' },
 	{ "no-longpoll", 0, NULL, 1003 },
+	{ "no-stratum", 0, NULL, 1007 },
 	{ "pass", 1, NULL, 'p' },
 	{ "protocol-dump", 0, NULL, 'P' },
 	{ "proxy", 1, NULL, 'x' },
@@ -234,6 +242,10 @@ static struct option const options[] = {
 struct work {
 	uint32_t data[32];
 	uint32_t target[8];
+
+	char job_id[128];
+	size_t xnonce2_len;
+	unsigned char xnonce2[32];
 };
 
 static struct work g_work;
@@ -286,13 +298,37 @@ err_out:
 	return false;
 }
 
-static bool submit_upstream_work(CURL *curl, struct work *work)
+static void share_result(int result, const char *reason)
 {
-	char *hexstr = NULL;
-	json_t *val, *res;
 	char s[345];
 	double hashrate;
 	int i;
+
+	hashrate = 0.;
+	pthread_mutex_lock(&stats_lock);
+	for (i = 0; i < opt_n_threads; i++)
+		hashrate += thr_hashrates[i];
+	result ? accepted_count++ : rejected_count++;
+	pthread_mutex_unlock(&stats_lock);
+	
+	sprintf(s, hashrate >= 1e6 ? "%.0f" : "%.2f", 1e-3 * hashrate);
+	applog(LOG_INFO, "accepted: %lu/%lu (%.2f%%), %s khash/s %s",
+		   accepted_count,
+		   accepted_count + rejected_count,
+		   100. * accepted_count / (accepted_count + rejected_count),
+		   s,
+		   result ? "(yay!!!)" : "(booooo)");
+
+	if (opt_debug && reason)
+		applog(LOG_DEBUG, "DEBUG: reject reason: %s", reason);
+}
+
+static bool submit_upstream_work(CURL *curl, struct work *work)
+{
+	char *str = NULL;
+	json_t *val, *res, *reason;
+	char s[345];
+	int i;
 	bool rc = false;
 
 	/* pass if the previous hash is not the current previous hash */
@@ -302,11 +338,34 @@ static bool submit_upstream_work(CURL *curl, struct work *work)
 		return true;
 	}
 
+	if (have_stratum) {
+		uint32_t ntime, nonce;
+		char *ntimestr, *noncestr, *xnonce2str;
+
+		if (!work->job_id)
+			return true;
+		le32enc(&ntime, work->data[17]);
+		le32enc(&nonce, work->data[19]);
+		ntimestr = bin2hex((const unsigned char *)(&ntime), 4);
+		noncestr = bin2hex((const unsigned char *)(&nonce), 4);
+		xnonce2str = bin2hex(work->xnonce2, work->xnonce2_len);
+		sprintf(s,
+			"{\"method\": \"mining.submit\", \"params\": [\"%s\", \"%s\", \"%s\", \"%s\", \"%s\"], \"id\":4}",
+			rpc_user, work->job_id, xnonce2str, ntimestr, noncestr);
+		free(ntimestr);
+		free(noncestr);
+		free(xnonce2str);
+
+		if (unlikely(!stratum_send_line(&stratum, s))) {
+			applog(LOG_ERR, "submit_upstream_work stratum_send_line failed");
+			goto out;
+		}
+	} else {
 		/* build hex string */
 		for (i = 0; i < ARRAY_SIZE(work->data); i++)
 			le32enc(work->data + i, work->data[i]);
-	hexstr = bin2hex((unsigned char *)work->data, sizeof(work->data));
-	if (unlikely(!hexstr)) {
+		str = bin2hex((unsigned char *)work->data, sizeof(work->data));
+		if (unlikely(!str)) {
 			applog(LOG_ERR, "submit_upstream_work OOM");
 			goto out;
 		}
@@ -314,7 +373,7 @@ static bool submit_upstream_work(CURL *curl, struct work *work)
 		/* build JSON-RPC request */
 		sprintf(s,
 			"{\"method\": \"getwork\", \"params\": [ \"%s\" ], \"id\":1}\r\n",
-		hexstr);
+			str);
 
 		/* issue JSON-RPC request */
 		val = json_rpc_call(curl, rpc_url, rpc_userpass, s, false, false, NULL);
@@ -324,36 +383,16 @@ static bool submit_upstream_work(CURL *curl, struct work *work)
 		}
 
 		res = json_object_get(val, "result");
-	
-	hashrate = 0.;
-	pthread_mutex_lock(&stats_lock);
-	for (i = 0; i < opt_n_threads; i++)
-		hashrate += thr_hashrates[i];
-	json_is_true(res) ? accepted_count++ : rejected_count++;
-	pthread_mutex_unlock(&stats_lock);
-	
-	sprintf(s, hashrate >= 1e6 ? "%.0f" : "%.2f", 1e-3 * hashrate);
-	applog(LOG_INFO, "accepted: %lu/%lu (%.2f%%), %s khash/s %s",
-	       accepted_count,
-	       accepted_count + rejected_count,
-	       100. * accepted_count / (accepted_count + rejected_count),
-	       s,
-	       json_is_true(res) ? "(yay!!!)" : "(booooo)");
-
-	if (opt_debug) {
-		json_t *tmp;
-		const char *reason;
-		tmp = json_object_get(val, "reject-reason");
-		if (tmp && (reason = json_string_value(tmp)))
-			applog(LOG_DEBUG, "DEBUG: reject reason: %s", reason);
-	}
+		reason = json_object_get(val, "reject-reason");
+		share_result(json_is_true(res), reason ? json_string_value(reason) : NULL);
 
 		json_decref(val);
+	}
 
 	rc = true;
 
 out:
-	free(hexstr);
+	free(str);
 	return rc;
 }
 
@@ -540,9 +579,6 @@ static bool submit_work(struct thr_info *thr, const struct work *work_in)
 {
 	struct workio_cmd *wc;
 	
-	if (opt_benchmark)
-		return true;
-	
 	/* fill out work request message */
 	wc = calloc(1, sizeof(*wc));
 	if (!wc)
@@ -567,6 +603,54 @@ err_out:
 	return false;
 }
 
+static void stratum_gen_work(struct stratum_ctx *sctx, struct work *work)
+{
+	unsigned char merkle_root[64];
+	int i;
+
+	pthread_mutex_lock(&sctx->work_lock);
+
+	strcpy(work->job_id, sctx->job.job_id);
+	work->xnonce2_len = sctx->xnonce2_size;
+	memcpy(work->xnonce2, sctx->job.xnonce2, sctx->xnonce2_size);
+
+	/* Generate merkle root */
+	sha256d(merkle_root, sctx->job.coinbase, sctx->job.coinbase_size);
+	for (i = 0; i < sctx->job.merkle_count; i++) {
+		memcpy(merkle_root + 32, sctx->job.merkle[i], 32);
+		sha256d(merkle_root, merkle_root, 64);
+	}
+	
+	/* Increment extranonce2 */
+	for (i = 0; i < sctx->xnonce2_size && !++sctx->job.xnonce2[i]; i++);
+
+	/* Assemble block header */
+	memset(work->data, 0, 128);
+	work->data[0] = le32dec(sctx->job.version);
+	for (i = 0; i < 8; i++)
+		work->data[1 + i] = le32dec((uint32_t *)sctx->job.prevhash + i);
+	for (i = 0; i < 8; i++)
+		work->data[9 + i] = be32dec((uint32_t *)merkle_root + i);
+	work->data[17] = le32dec(sctx->job.ntime);
+	work->data[18] = le32dec(sctx->job.nbits);
+	work->data[20] = 0x80000000;
+	work->data[31] = 0x00000280;
+
+	pthread_mutex_unlock(&sctx->work_lock);
+
+	if (opt_debug) {
+		char *xnonce2str = bin2hex(work->xnonce2, sctx->xnonce2_size);
+		applog(LOG_DEBUG, "DEBUG: job_id='%s' extranonce2=%s ntime=%08x",
+		       work->job_id, xnonce2str, swab32(work->data[17]));
+		free(xnonce2str);
+	}
+
+	if (opt_algo == ALGO_SCRYPT)
+		diff_to_target(work->target, sctx->job.diff / 65536.0);
+	else
+		diff_to_target(work->target, sctx->job.diff);
+}
+
 static void *miner_thread(void *userdata)
 {
 	struct thr_info *mythr = userdata;
@@ -602,10 +686,18 @@ static void *miner_thread(void *userdata)
 		int64_t max64;
 		int rc;
 
+		if (have_stratum) {
+			while (!*g_work.job_id || time(NULL) >= g_work_time + 120)
+				sleep(1);
+			pthread_mutex_lock(&g_work_lock);
+			if (work.data[19] >= end_nonce)
+				stratum_gen_work(&stratum, &g_work);
+		} else {
 			/* obtain new work from internal workio thread */
 			pthread_mutex_lock(&g_work_lock);
-		if (!have_longpoll || time(NULL) >= g_work_time + LP_SCANTIME*3/4
-			|| work.data[19] >= end_nonce) {
+			if (!(have_longpoll || have_stratum) ||
+					time(NULL) >= g_work_time + LP_SCANTIME*3/4 ||
+					work.data[19] >= end_nonce) {
 				if (unlikely(!get_work(mythr, &g_work))) {
 					applog(LOG_ERR, "work retrieval failed, exiting "
 						"mining thread %d", mythr->id);
@@ -614,6 +706,11 @@ static void *miner_thread(void *userdata)
 				}
 				time(&g_work_time);
 			}
+			if (have_stratum) {
+				pthread_mutex_unlock(&g_work_lock);
+				continue;
+			}
+		}
 		if (memcmp(work.data, g_work.data, 76)) {
 			memcpy(&work, &g_work, sizeof(struct work));
 			work.data[19] = 0xffffffffU / opt_n_threads * thr_id;
@@ -623,6 +720,9 @@ static void *miner_thread(void *userdata)
 		work_restart[thr_id].restart = 0;
 		
 		/* adjust max_nonce to meet target scan time */
+		if (have_stratum)
+			max64 = LP_SCANTIME;
+		else
 			max64 = g_work_time + (have_longpoll ? LP_SCANTIME : opt_scantime)
 			      - time(NULL);
 		max64 *= thr_hashrates[thr_id];
@@ -679,7 +779,7 @@ static void *miner_thread(void *userdata)
 		}
 
 		/* if nonce found, submit work */
-		if (rc && !submit_work(mythr, &work))
+		if (rc && !opt_benchmark && !submit_work(mythr, &work))
 			break;
 	}
 
@@ -742,6 +842,11 @@ start:
 
 		val = json_rpc_call(curl, lp_url, rpc_userpass, rpc_req,
 				    false, true, &err);
+		if (have_stratum) {
+			if (val)
+				json_decref(val);
+			goto out;
+		}
 		if (likely(val)) {
 			applog(LOG_INFO, "LONGPOLL detected new block");
 			soval = json_object_get(json_object_get(val, "result"), "submitold");
@@ -783,6 +888,99 @@ out:
 	return NULL;
 }
 
+static bool stratum_handle_response(char *buf)
+{
+	json_t *val, *err_val, *res_val, *id_val;
+	json_error_t err;
+	bool ret = false;
+
+	val = JSON_LOADS(buf, &err);
+	if (!val) {
+		applog(LOG_INFO, "JSON decode failed(%d): %s", err.line, err.text);
+		goto out;
+	}
+
+	res_val = json_object_get(val, "result");
+	err_val = json_object_get(val, "error");
+	id_val = json_object_get(val, "id");
+
+	if (!id_val || json_is_null(id_val) || !res_val)
+		goto out;
+
+	share_result(json_is_true(res_val),
+		err_val ? json_string_value(json_array_get(err_val, 1)) : NULL);
+
+	ret = true;
+out:
+	if (val)
+		json_decref(val);
+
+	return ret;
+}
+
+static void *stratum_thread(void *userdata)
+{
+	struct thr_info *mythr = userdata;
+	char *s;
+
+	stratum.url = tq_pop(mythr->q, NULL);
+	if (!stratum.url)
+		goto out;
+	applog(LOG_INFO, "Starting Stratum on %s", stratum.url);
+
+	while (1) {
+		int failures = 0;
+
+		while (!stratum.curl) {
+			pthread_mutex_lock(&g_work_lock);
+			g_work_time = 0;
+			pthread_mutex_unlock(&g_work_lock);
+			restart_threads();
+
+			if (!stratum_connect(&stratum, stratum.url) ||
+			    !stratum_subscribe(&stratum) ||
+			    !stratum_authorize(&stratum, rpc_user, rpc_pass)) {
+				stratum_disconnect(&stratum);
+				if (opt_retries >= 0 && ++failures > opt_retries) {
+					applog(LOG_ERR, "...terminating workio thread");
+					tq_push(thr_info[work_thr_id].q, NULL);
+					goto out;
+				}
+				applog(LOG_ERR, "...retry after %d seconds", opt_fail_pause);
+				sleep(opt_fail_pause);
+			}
+		}
+
+		if (strcmp(stratum.job.job_id, g_work.job_id) || !g_work_time) {
+			pthread_mutex_lock(&g_work_lock);
+			stratum_gen_work(&stratum, &g_work);
+			time(&g_work_time);
+			pthread_mutex_unlock(&g_work_lock);
+			if (stratum.job.clean) {
+				applog(LOG_INFO, "Stratum detected new block");
+				restart_threads();
+			}
+		}
+		
+		if (!stratum_socket_full(&stratum, 120)) {
+			applog(LOG_ERR, "Stratum connection timed out");
+			s = NULL;
+		} else
+			s = stratum_recv_line(&stratum);
+		if (!s) {
+			stratum_disconnect(&stratum);
+			applog(LOG_ERR, "Stratum connection interrupted");
+			continue;
+		}
+		if (!stratum_handle_method(&stratum, s))
+			stratum_handle_response(s);
+		free(s);
+	}
+
+out:
+	return NULL;
+}
+
 static void show_version_and_exit(void)
 {
 	printf("%s\n%s\n", PACKAGE_STRING, curl_version());
@@ -883,7 +1081,8 @@ static void parse_arg (int key, char *arg)
 	case 'o':			/* --url */
 		p = strstr(arg, "://");
 		if (p) {
-			if (strncmp(arg, "http://", 7) && strncmp(arg, "https://", 8))
+			if (strncasecmp(arg, "http://", 7) && strncasecmp(arg, "https://", 8) &&
+					strncasecmp(arg, "stratum+tcp://", 14))
 				show_usage_and_exit(1);
 			free(rpc_url);
 			rpc_url = strdup(arg);
@@ -891,38 +1090,52 @@ static void parse_arg (int key, char *arg)
 			if (!strlen(arg) || *arg == '/')
 				show_usage_and_exit(1);
 			free(rpc_url);
-			rpc_url = malloc((strlen(arg) + 8) * sizeof(char));
+			rpc_url = malloc(strlen(arg) + 8);
 			sprintf(rpc_url, "http://%s", arg);
 		}
 		p = strrchr(rpc_url, '@');
 		if (p) {
-			char *ap = strstr(rpc_url, "://") + 3;
+			char *sp, *ap;
 			*p = '\0';
-			if (strchr(ap, ':')) {
+			ap = strstr(rpc_url, "://") + 3;
+			sp = strchr(ap, ':');
+			if (sp) {
 				free(rpc_userpass);
 				rpc_userpass = strdup(ap);
+				free(rpc_user);
+				rpc_user = calloc(sp - ap + 1, 1);
+				strncpy(rpc_user, ap, sp - ap);
+				free(rpc_pass);
+				rpc_pass = strdup(sp + 1);
 			} else {
 				free(rpc_user);
 				rpc_user = strdup(ap);
 			}
-			memmove(ap, p + 1, (strlen(p + 1) + 1) * sizeof(char));
+			memmove(ap, p + 1, strlen(p + 1) + 1);
 		}
+		have_stratum = !strncasecmp(rpc_url, "stratum", 7);
 		break;
 	case 'O':			/* --userpass */
-		if (!strchr(arg, ':'))
+		p = strchr(arg, ':');
+		if (!p)
 			show_usage_and_exit(1);
 		free(rpc_userpass);
 		rpc_userpass = strdup(arg);
+		free(rpc_user);
+		rpc_user = calloc(p - arg + 1, 1);
+		strncpy(rpc_user, arg, p - arg);
+		free(rpc_pass);
+		rpc_pass = strdup(p + 1);
 		break;
 	case 'x':			/* --proxy */
-		if (!strncmp(arg, "socks4://", 9))
+		if (!strncasecmp(arg, "socks4://", 9))
 			opt_proxy_type = CURLPROXY_SOCKS4;
-		else if (!strncmp(arg, "socks5://", 9))
+		else if (!strncasecmp(arg, "socks5://", 9))
 			opt_proxy_type = CURLPROXY_SOCKS5;
 #if LIBCURL_VERSION_NUM >= 0x071200
-		else if (!strncmp(arg, "socks4a://", 10))
+		else if (!strncasecmp(arg, "socks4a://", 10))
 			opt_proxy_type = CURLPROXY_SOCKS4A;
-		else if (!strncmp(arg, "socks5h://", 10))
+		else if (!strncasecmp(arg, "socks5h://", 10))
 			opt_proxy_type = CURLPROXY_SOCKS5_HOSTNAME;
 #endif
 		else
@@ -932,9 +1145,15 @@ static void parse_arg (int key, char *arg)
 		break;
 	case 1005:
 		opt_benchmark = true;
+		want_longpoll = false;
+		want_stratum = false;
+		break;
 	case 1003:
 		want_longpoll = false;
 		break;
+	case 1007:
+		want_stratum = false;
+		break;
 	case 'S':
 		use_syslog = true;
 		break;
@@ -1028,6 +1247,8 @@ int main(int argc, char *argv[])
 	int i;
 
 	rpc_url = strdup(DEF_RPC_URL);
+	rpc_user = strdup("");
+	rpc_pass = strdup("");
 
 	/* parse command line */
 	parse_cmdline(argc, argv);
@@ -1052,6 +1273,8 @@ int main(int argc, char *argv[])
 	pthread_mutex_init(&applog_lock, NULL);
 	pthread_mutex_init(&stats_lock, NULL);
 	pthread_mutex_init(&g_work_lock, NULL);
+	pthread_mutex_init(&stratum.sock_lock, NULL);
+	pthread_mutex_init(&stratum.work_lock, NULL);
 
 #if defined(WIN32)
 	SYSTEM_INFO sysinfo;
@@ -1071,11 +1294,7 @@ int main(int argc, char *argv[])
 	if (!opt_n_threads)
 		opt_n_threads = num_processors;
 
-	if (!rpc_userpass && (rpc_user || rpc_pass)) {
-		if (!rpc_user)
-			rpc_user = strdup("");
-		if (!rpc_pass)
-			rpc_pass = strdup("");
+	if (!rpc_userpass) {
 		rpc_userpass = malloc(strlen(rpc_user) + strlen(rpc_pass) + 2);
 		if (!rpc_userpass)
 			return 1;
@@ -1091,7 +1310,7 @@ int main(int argc, char *argv[])
 	if (!work_restart)
 		return 1;
 
-	thr_info = calloc(opt_n_threads + 2, sizeof(*thr));
+	thr_info = calloc(opt_n_threads + 3, sizeof(*thr));
 	if (!thr_info)
 		return 1;
 	
@@ -1113,8 +1332,8 @@ int main(int argc, char *argv[])
 		return 1;
 	}
 
+	if (want_longpoll && !have_stratum) {
 		/* init longpoll thread info */
-	if (want_longpoll) {
 		longpoll_thr_id = opt_n_threads + 1;
 		thr = &thr_info[longpoll_thr_id];
 		thr->id = longpoll_thr_id;
@@ -1127,8 +1346,25 @@ int main(int argc, char *argv[])
 			applog(LOG_ERR, "longpoll thread create failed");
 			return 1;
 		}
-	} else
-		longpoll_thr_id = -1;
+	}
+	if (want_stratum) {
+		/* init stratum thread info */
+		stratum_thr_id = opt_n_threads + 2;
+		thr = &thr_info[stratum_thr_id];
+		thr->id = stratum_thr_id;
+		thr->q = tq_new();
+		if (!thr->q)
+			return 1;
+
+		/* start stratum thread */
+		if (unlikely(pthread_create(&thr->pth, NULL, stratum_thread, thr))) {
+			applog(LOG_ERR, "stratum thread create failed");
+			return 1;
+		}
+
+		if (have_stratum)
+			tq_push(thr_info[stratum_thr_id].q, strdup(rpc_url));
+	}
 
 	/* start mining threads */
 	for (i = 0; i < opt_n_threads; i++) {

miner.h

@@ -42,6 +42,7 @@ void *alloca (size_t);
 enum {
 	LOG_ERR,
 	LOG_WARNING,
+	LOG_NOTICE,
 	LOG_INFO,
 	LOG_DEBUG,
 };
@@ -121,8 +122,17 @@ static inline void le32enc(void *pp, uint32_t x)
 }
 #endif
 
+#if JANSSON_MAJOR_VERSION >= 2
+#define JSON_LOADS(str, err_ptr) json_loads((str), 0, (err_ptr))
+#else
+#define JSON_LOADS(str, err_ptr) json_loads((str), (err_ptr))
+#endif
+
+#define USER_AGENT PACKAGE_NAME "/" PACKAGE_VERSION
+
 void sha256_init(uint32_t *state);
 void sha256_transform(uint32_t *state, const uint32_t *block, int swap);
+void sha256d(unsigned char *hash, const unsigned char *data, int len);
 
 #if defined(__ARM_NEON__) || defined(__i386__) || defined(__x86_64__)
 #define HAVE_SHA256_4WAY 1
@@ -155,12 +165,15 @@ extern bool opt_protocol;
 extern int opt_timeout;
 extern bool want_longpoll;
 extern bool have_longpoll;
+extern bool want_stratum;
+extern bool have_stratum;
 extern char *opt_proxy;
 extern long opt_proxy_type;
 extern bool use_syslog;
 extern pthread_mutex_t applog_lock;
 extern struct thr_info *thr_info;
 extern int longpoll_thr_id;
+extern int stratum_thr_id;
 extern struct work_restart *work_restart;
 
 extern void applog(int prio, const char *fmt, ...);
@@ -171,6 +184,52 @@ extern bool hex2bin(unsigned char *p, const char *hexstr, size_t len);
 extern int timeval_subtract(struct timeval *result, struct timeval *x,
 	struct timeval *y);
 extern bool fulltest(const uint32_t *hash, const uint32_t *target);
+extern void diff_to_target(uint32_t *target, double diff);
+
+struct stratum_job {
+	char *job_id;
+	unsigned char prevhash[32];
+	size_t coinbase_size;
+	unsigned char *coinbase;
+	unsigned char *xnonce2;
+	int merkle_count;
+	unsigned char **merkle;
+	unsigned char version[4];
+	unsigned char nbits[4];
+	unsigned char ntime[4];
+	bool clean;
+	double diff;
+};
+
+struct stratum_ctx {
+	char *url;
+
+	CURL *curl;
+	char *curl_url;
+	char curl_err_str[CURL_ERROR_SIZE];
+	curl_socket_t sock;
+	size_t sockbuf_size;
+	char *sockbuf;
+	pthread_mutex_t sock_lock;
+
+	double next_diff;
+
+	char *session_id;
+	size_t xnonce1_size;
+	unsigned char *xnonce1;
+	size_t xnonce2_size;
+	struct stratum_job job;
+	pthread_mutex_t work_lock;
+};
+
+bool stratum_socket_full(struct stratum_ctx *sctx, int timeout);
+bool stratum_send_line(struct stratum_ctx *sctx, char *s);
+char *stratum_recv_line(struct stratum_ctx *sctx);
+bool stratum_connect(struct stratum_ctx *sctx, const char *url);
+void stratum_disconnect(struct stratum_ctx *sctx);
+bool stratum_subscribe(struct stratum_ctx *sctx);
+bool stratum_authorize(struct stratum_ctx *sctx, const char *user, const char *pass);
+bool stratum_handle_method(struct stratum_ctx *sctx, const char *s);
 
 struct thread_q;
 

sha2.c

@@ -1,5 +1,6 @@
 /*
- * Copyright 2011 ArtForz, 2011-2012 pooler
+ * Copyright 2011 ArtForz
+ * Copyright 2011-2013 pooler
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the Free
@@ -180,7 +181,7 @@ static const uint32_t sha256d_hash1[16] = {
 	0x00000000, 0x00000000, 0x00000000, 0x00000100
 };
 
-static void sha256d(uint32_t *hash, uint32_t *data)
+static void sha256d_80(uint32_t *hash, const uint32_t *data)
 {
 	uint32_t S[16];
 
@@ -192,6 +193,31 @@ static void sha256d(uint32_t *hash, uint32_t *data)
 	sha256_transform(hash, S, 0);
 }
 
+void sha256d(unsigned char *hash, const unsigned char *data, int len)
+{
+	uint32_t S[16], T[16];
+	int i, r;
+
+	sha256_init(S);
+	for (r = len; r > -9; r -= 64) {
+		if (r < 64)
+			memset(T, 0, 64);
+		memcpy(T, data + len - r, r > 64 ? 64 : (r < 0 ? 0 : r));
+		if (r < 64)
+			((unsigned char *)T)[r] = 0x80;
+		for (i = 0; i < 16; i++)
+			T[i] = be32dec(T + i);
+		if (r < 56)
+			T[15] = 8 * len;
+		sha256_transform(S, T, 0);
+	}
+	memcpy(S + 8, sha256d_hash1 + 8, 32);
+	sha256_init(T);
+	sha256_transform(T, S, 0);
+	for (i = 0; i < 8; i++)
+		be32enc((uint32_t *)hash + i, T[i]);
+}
+
 static inline void sha256d_preextend(uint32_t *W)
 {
 	W[16] = s1(W[14]) + W[ 9] + s0(W[ 1]) + W[ 0];
@@ -477,7 +503,7 @@ static inline int scanhash_sha256d_4way(int thr_id, uint32_t *pdata,
 		for (i = 0; i < 4; i++) {
 			if (hash[4 * 7 + i] <= Htarg) {
 				pdata[19] = data[4 * 3 + i];
-				sha256d(hash, pdata);
+				sha256d_80(hash, pdata);
 				if (fulltest(hash, ptarget)) {
 					*hashes_done = n - first_nonce + 1;
 					return 1;
@@ -523,7 +549,7 @@ int scanhash_sha256d(int thr_id, uint32_t *pdata, const uint32_t *ptarget,
 		sha256d_ms(hash, data, midstate, prehash);
 		if (hash[7] <= Htarg) {
 			pdata[19] = data[3];
-			sha256d(hash, pdata);
+			sha256d_80(hash, pdata);
 			if (fulltest(hash, ptarget)) {
 				*hashes_done = n - first_nonce + 1;
 				return 1;

util.c

@@ -1,5 +1,6 @@
 /*
- * Copyright 2010 Jeff Garzik, 2012 pooler
+ * Copyright 2010 Jeff Garzik
+ * Copyright 2012-2013 pooler
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the Free
@@ -25,6 +26,7 @@
 #include <winsock2.h>
 #include <mstcpip.h>
 #else
+#include <errno.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <netinet/tcp.h>
@@ -47,6 +49,7 @@ struct upload_buffer {
 struct header_info {
 	char		*lp_path;
 	char		*reason;
+	char		*stratum_url;
 };
 
 struct tq_ent {
@@ -235,6 +238,11 @@ static size_t resp_hdr_cb(void *ptr, size_t size, size_t nmemb, void *user_data)
 		val = NULL;
 	}
 
+	if (!strcasecmp("X-Stratum", key)) {
+		hi->stratum_url = val;	/* steal memory reference */
+		val = NULL;
+	}
+
 out:
 	free(key);
 	free(val);
@@ -242,7 +250,7 @@ out:
 }
 
 #if LIBCURL_VERSION_NUM >= 0x070f06
-static int json_rpc_call_lp_cb(void *userdata, curl_socket_t fd,
+static int sockopt_keepalive_cb(void *userdata, curl_socket_t fd,
 	curlsocktype purpose)
 {
 	int keepalive = 1;
@@ -333,7 +341,7 @@ json_t *json_rpc_call(CURL *curl, const char *url,
 	}
 #if LIBCURL_VERSION_NUM >= 0x070f06
 	if (longpoll)
-		curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, json_rpc_call_lp_cb);
+		curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, sockopt_keepalive_cb);
 #endif
 	curl_easy_setopt(curl, CURLOPT_POST, 1);
 
@@ -346,13 +354,10 @@ json_t *json_rpc_call(CURL *curl, const char *url,
 	sprintf(len_hdr, "Content-Length: %lu",
 		(unsigned long) upload_data.len);
 
-	headers = curl_slist_append(headers,
-		"Content-Type: application/json");
+	headers = curl_slist_append(headers, "Content-Type: application/json");
 	headers = curl_slist_append(headers, len_hdr);
-	headers = curl_slist_append(headers,
-		"User-Agent: " PACKAGE_NAME "/" PACKAGE_VERSION);
-	headers = curl_slist_append(headers,
-		"X-Mining-Extensions: midstate");
+	headers = curl_slist_append(headers, "User-Agent: " USER_AGENT);
+	headers = curl_slist_append(headers, "X-Mining-Extensions: midstate");
 	headers = curl_slist_append(headers, "Accept:"); /* disable Accept hdr*/
 	headers = curl_slist_append(headers, "Expect:"); /* disable Expect hdr*/
 
@@ -367,24 +372,27 @@ json_t *json_rpc_call(CURL *curl, const char *url,
 		goto err_out;
 	}
 
+	/* If X-Stratum was found, activate Stratum */
+	if (want_stratum && hi.stratum_url &&
+	    !strncasecmp(hi.stratum_url, "stratum+tcp://", 14)) {
+		have_stratum = true;
+		tq_push(thr_info[stratum_thr_id].q, hi.stratum_url);
+		hi.stratum_url = NULL;
+	}
+
 	/* If X-Long-Polling was found, activate long polling */
-	if (lp_scanning && hi.lp_path) {
+	if (lp_scanning && hi.lp_path && !have_stratum) {
 		have_longpoll = true;
 		tq_push(thr_info[longpoll_thr_id].q, hi.lp_path);
-	} else
-		free(hi.lp_path);
 		hi.lp_path = NULL;
+	}
 
 	if (!all_data.buf) {
 		applog(LOG_ERR, "Empty data received in json_rpc_call.");
 		goto err_out;
 	}
 
-#if JANSSON_VERSION_HEX >= 0x020000
-	val = json_loads(all_data.buf, 0, &err);
-#else
-	val = json_loads(all_data.buf, &err);
-#endif
+	val = JSON_LOADS(all_data.buf, &err);
 	if (!val) {
 		applog(LOG_ERR, "JSON decode failed(%d): %s", err.line, err.text);
 		goto err_out;
@@ -426,6 +434,9 @@ json_t *json_rpc_call(CURL *curl, const char *url,
 	return val;
 
 err_out:
+	free(hi.lp_path);
+	free(hi.reason);
+	free(hi.stratum_url);
 	databuf_free(&all_data);
 	curl_slist_free_all(headers);
 	curl_easy_reset(curl);
@@ -542,6 +553,650 @@ bool fulltest(const uint32_t *hash, const uint32_t *target)
 	return rc;
 }
 
+void diff_to_target(uint32_t *target, double diff)
+{
+	uint64_t m;
+	int k;
+	
+	for (k = 6; k > 0 && diff > 1.0; k--)
+		diff /= 4294967296.0;
+	m = 4294901760.0 / diff;
+	if (m == 0 && k == 6)
+		memset(target, 0xff, 32);
+	else {
+		memset(target, 0, 32);
+		target[k] = (uint32_t)m;
+		target[k + 1] = (uint32_t)(m >> 32);
+	}
+}
+
+#ifdef WIN32
+#define socket_blocks() (WSAGetLastError() == WSAEWOULDBLOCK)
+#else
+#define socket_blocks() (errno == EAGAIN || errno == EWOULDBLOCK)
+#endif
+
+static bool send_line(curl_socket_t sock, char *s)
+{
+	ssize_t len, sent = 0;
+	
+	len = strlen(s);
+	s[len++] = '\n';
+
+	while (len > 0) {
+		struct timeval timeout = {0, 0};
+		ssize_t n;
+		fd_set wd;
+
+		FD_ZERO(&wd);
+		FD_SET(sock, &wd);
+		if (select(sock + 1, NULL, &wd, NULL, &timeout) < 1)
+			return false;
+		n = send(sock, s + sent, len, 0);
+		if (n < 0) {
+			if (!socket_blocks())
+				return false;
+			n = 0;
+		}
+		sent += n;
+		len -= n;
+	}
+
+	return true;
+}
+
+bool stratum_send_line(struct stratum_ctx *sctx, char *s)
+{
+	bool ret = false;
+
+	if (opt_protocol)
+		applog(LOG_DEBUG, "> %s", s);
+
+	pthread_mutex_lock(&sctx->sock_lock);
+	ret = send_line(sctx->sock, s);
+	pthread_mutex_unlock(&sctx->sock_lock);
+
+	return ret;
+}
+
+static bool socket_full(curl_socket_t sock, int timeout)
+{
+	struct timeval tv;
+	fd_set rd;
+
+	FD_ZERO(&rd);
+	FD_SET(sock, &rd);
+	tv.tv_sec = timeout;
+	tv.tv_usec = 0;
+	if (select(sock + 1, &rd, NULL, NULL, &tv) > 0)
+		return true;
+	return false;
+}
+
+bool stratum_socket_full(struct stratum_ctx *sctx, int timeout)
+{
+	return strlen(sctx->sockbuf) || socket_full(sctx->sock, timeout);
+}
+
+#define RBUFSIZE 2048
+#define RECVSIZE (RBUFSIZE - 4)
+
+static void stratum_buffer_append(struct stratum_ctx *sctx, const char *s)
+{
+	size_t old, new;
+
+	old = strlen(sctx->sockbuf);
+	new = old + strlen(s) + 1;
+	if (new >= sctx->sockbuf_size) {
+		sctx->sockbuf_size = new + (RBUFSIZE - (new % RBUFSIZE));
+		sctx->sockbuf = realloc(sctx->sockbuf, sctx->sockbuf_size);
+	}
+	strcpy(sctx->sockbuf + old, s);
+}
+
+char *stratum_recv_line(struct stratum_ctx *sctx)
+{
+	ssize_t len, buflen;
+	char *tok, *sret = NULL;
+
+	if (!strstr(sctx->sockbuf, "\n")) {
+		bool ret = true;
+		time_t rstart;
+
+		time(&rstart);
+		if (!socket_full(sctx->sock, 60)) {
+			applog(LOG_ERR, "stratum_recv_line timed out");
+			goto out;
+		}
+		do {
+			char s[RBUFSIZE];
+			ssize_t n;
+
+			memset(s, 0, RBUFSIZE);
+			n = recv(sctx->sock, s, RECVSIZE, 0);
+			if (!n) {
+				ret = false;
+				break;
+			}
+			if (n < 0) {
+				if (!socket_blocks() || !socket_full(sctx->sock, 1)) {
+					ret = false;
+					break;
+				}
+			} else
+				stratum_buffer_append(sctx, s);
+		} while (time(NULL) - rstart < 60 && !strstr(sctx->sockbuf, "\n"));
+
+		if (!ret) {
+			applog(LOG_ERR, "stratum_recv_line failed");
+			goto out;
+		}
+	}
+
+	buflen = strlen(sctx->sockbuf);
+	tok = strtok(sctx->sockbuf, "\n");
+	if (!tok) {
+		applog(LOG_ERR, "stratum_recv_line failed to parse a newline-terminated string");
+		goto out;
+	}
+	sret = strdup(tok);
+	len = strlen(sret);
+
+	if (buflen > len + 1)
+		memmove(sctx->sockbuf, sctx->sockbuf + len + 1, buflen - len + 1);
+	else
+		sctx->sockbuf[0] = '\0';
+
+out:
+	if (sret && opt_protocol)
+		applog(LOG_DEBUG, "< %s", sret);
+	return sret;
+}
+
+#if LIBCURL_VERSION_NUM >= 0x071101
+static curl_socket_t opensocket_grab_cb(void *clientp, curlsocktype purpose,
+	struct curl_sockaddr *addr)
+{
+	curl_socket_t *sock = clientp;
+	*sock = socket(addr->family, addr->socktype, addr->protocol);
+	return *sock;
+}
+#endif
+
+bool stratum_connect(struct stratum_ctx *sctx, const char *url)
+{
+	CURL *curl;
+	int rc;
+
+	pthread_mutex_lock(&sctx->sock_lock);
+	if (sctx->curl)
+		curl_easy_cleanup(sctx->curl);
+	sctx->curl = curl_easy_init();
+	if (!sctx->curl) {
+		applog(LOG_ERR, "CURL initialization failed");
+		pthread_mutex_unlock(&sctx->sock_lock);
+		return false;
+	}
+	curl = sctx->curl;
+	if (!sctx->sockbuf) {
+		sctx->sockbuf = calloc(RBUFSIZE, 1);
+		sctx->sockbuf_size = RBUFSIZE;
+	}
+	sctx->sockbuf[0] = '\0';
+	pthread_mutex_unlock(&sctx->sock_lock);
+
+	if (url != sctx->url) {
+		free(sctx->url);
+		sctx->url = strdup(url);
+	}
+	free(sctx->curl_url);
+	sctx->curl_url = malloc(strlen(url));
+	sprintf(sctx->curl_url, "http%s", strstr(url, "://"));
+
+	if (opt_protocol)
+		curl_easy_setopt(curl, CURLOPT_VERBOSE, 1);
+	curl_easy_setopt(curl, CURLOPT_URL, sctx->curl_url);
+	curl_easy_setopt(curl, CURLOPT_FRESH_CONNECT, 1);
+	curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 30);
+	curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, sctx->curl_err_str);
+	curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
+	curl_easy_setopt(curl, CURLOPT_TCP_NODELAY, 1);
+	if (opt_proxy && opt_proxy_type != CURLPROXY_HTTP) {
+		curl_easy_setopt(curl, CURLOPT_PROXY, opt_proxy);
+		curl_easy_setopt(curl, CURLOPT_PROXYTYPE, opt_proxy_type);
+	} else if (getenv("http_proxy")) {
+		if (getenv("all_proxy"))
+			curl_easy_setopt(curl, CURLOPT_PROXY, getenv("all_proxy"));
+		else if (getenv("ALL_PROXY"))
+			curl_easy_setopt(curl, CURLOPT_PROXY, getenv("ALL_PROXY"));
+		else
+			curl_easy_setopt(curl, CURLOPT_PROXY, "");
+	}
+#if LIBCURL_VERSION_NUM >= 0x070f06
+	curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, sockopt_keepalive_cb);
+#endif
+#if LIBCURL_VERSION_NUM >= 0x071101
+	curl_easy_setopt(curl, CURLOPT_OPENSOCKETFUNCTION, opensocket_grab_cb);
+	curl_easy_setopt(curl, CURLOPT_OPENSOCKETDATA, &sctx->sock);
+#endif
+	curl_easy_setopt(curl, CURLOPT_CONNECT_ONLY, 1);
+
+	rc = curl_easy_perform(curl);
+	if (rc) {
+		applog(LOG_ERR, "Stratum connection failed: %s", sctx->curl_err_str);
+		curl_easy_cleanup(curl);
+		sctx->curl = NULL;
+		return false;
+	}
+
+#if LIBCURL_VERSION_NUM < 0x071101
+	/* CURLINFO_LASTSOCKET is broken on Win64; only use it as a last resort */
+	curl_easy_getinfo(curl, CURLINFO_LASTSOCKET, (long *)&sctx->sock);
+#endif
+
+	return true;
+}
+
+void stratum_disconnect(struct stratum_ctx *sctx)
+{
+	pthread_mutex_lock(&sctx->sock_lock);
+	if (sctx->curl) {
+		curl_easy_cleanup(sctx->curl);
+		sctx->curl = NULL;
+		sctx->sockbuf[0] = '\0';
+	}
+	pthread_mutex_unlock(&sctx->sock_lock);
+}
+
+static const char *get_stratum_session_id(json_t *val)
+{
+	json_t *arr_val;
+	int i, n;
+
+	arr_val = json_array_get(val, 0);
+	if (!arr_val || !json_is_array(arr_val))
+		return NULL;
+	n = json_array_size(arr_val);
+	for (i = 0; i < n; i++) {
+		const char *notify;
+		json_t *arr = json_array_get(arr_val, i);
+
+		if (!arr || !json_is_array(arr))
+			break;
+		notify = json_string_value(json_array_get(arr, 0));
+		if (!notify)
+			continue;
+		if (!strcasecmp(notify, "mining.notify"))
+			return json_string_value(json_array_get(arr, 1));
+	}
+	return NULL;
+}
+
+bool stratum_subscribe(struct stratum_ctx *sctx)
+{
+	char *s, *sret = NULL;
+	const char *sid, *xnonce1;
+	int xn2_size;
+	json_t *val = NULL, *res_val, *err_val;
+	json_error_t err;
+	bool ret = false, retry = false;
+
+start:
+	s = malloc(128 + (sctx->session_id ? strlen(sctx->session_id) : 0));
+	if (retry)
+		sprintf(s, "{\"id\": 1, \"method\": \"mining.subscribe\", \"params\": []}");
+	else if (sctx->session_id)
+		sprintf(s, "{\"id\": 1, \"method\": \"mining.subscribe\", \"params\": [\"" USER_AGENT "\", \"%s\"]}", sctx->session_id);
+	else
+		sprintf(s, "{\"id\": 1, \"method\": \"mining.subscribe\", \"params\": [\"" USER_AGENT "\"]}");
+
+	if (!stratum_send_line(sctx, s))
+		goto out;
+
+	if (!socket_full(sctx->sock, 30)) {
+		applog(LOG_ERR, "stratum_subscribe timed out");
+		goto out;
+	}
+
+	sret = stratum_recv_line(sctx);
+	if (!sret)
+		goto out;
+
+	val = JSON_LOADS(sret, &err);
+	free(sret);
+	if (!val) {
+		applog(LOG_ERR, "JSON decode failed(%d): %s", err.line, err.text);
+		goto out;
+	}
+
+	res_val = json_object_get(val, "result");
+	err_val = json_object_get(val, "error");
+
+	if (!res_val || json_is_null(res_val) ||
+	    (err_val && !json_is_null(err_val))) {
+		if (opt_debug || retry) {
+			free(s);
+			if (err_val)
+				s = json_dumps(err_val, JSON_INDENT(3));
+			else
+				s = strdup("(unknown reason)");
+			applog(LOG_ERR, "JSON-RPC call failed: %s", s);
+		}
+		goto out;
+	}
+
+	sid = get_stratum_session_id(res_val);
+	if (opt_debug && !sid)
+		applog(LOG_DEBUG, "Failed to get Stratum session id");
+	xnonce1 = json_string_value(json_array_get(res_val, 1));
+	if (!xnonce1) {
+		applog(LOG_ERR, "Failed to get extranonce1");
+		goto out;
+	}
+	xn2_size = json_integer_value(json_array_get(res_val, 2));
+	if (!xn2_size) {
+		applog(LOG_ERR, "Failed to get extranonce2_size");
+		goto out;
+	}
+
+	pthread_mutex_lock(&sctx->work_lock);
+	free(sctx->session_id);
+	free(sctx->xnonce1);
+	sctx->session_id = sid ? strdup(sid) : NULL;
+	sctx->xnonce1_size = strlen(xnonce1) / 2;
+	sctx->xnonce1 = malloc(sctx->xnonce1_size);
+	hex2bin(sctx->xnonce1, xnonce1, sctx->xnonce1_size);
+	sctx->xnonce2_size = xn2_size;
+	sctx->next_diff = 1.0;
+	pthread_mutex_unlock(&sctx->work_lock);
+
+	if (opt_debug && sid)
+		applog(LOG_DEBUG, "Stratum session id: %s", sctx->session_id);
+
+	ret = true;
+
+out:
+	free(s);
+	if (val)
+		json_decref(val);
+
+	if (!ret) {
+		if (sret && !retry) {
+			retry = true;
+			goto start;
+		}
+	}
+
+	return ret;
+}
+
+bool stratum_authorize(struct stratum_ctx *sctx, const char *user, const char *pass)
+{
+	json_t *val = NULL, *res_val, *err_val;
+	char *s, *sret;
+	json_error_t err;
+	bool ret = false;
+
+	s = malloc(80 + strlen(user) + strlen(pass));
+	sprintf(s, "{\"id\": 2, \"method\": \"mining.authorize\", \"params\": [\"%s\", \"%s\"]}",
+	        user, pass);
+
+	if (!stratum_send_line(sctx, s))
+		goto out;
+
+	while (1) {
+		sret = stratum_recv_line(sctx);
+		if (!sret)
+			goto out;
+		if (!stratum_handle_method(sctx, sret))
+			break;
+		free(sret);
+	}
+
+	val = JSON_LOADS(sret, &err);
+	free(sret);
+	if (!val) {
+		applog(LOG_ERR, "JSON decode failed(%d): %s", err.line, err.text);
+		goto out;
+	}
+
+	res_val = json_object_get(val, "result");
+	err_val = json_object_get(val, "error");
+
+	if (!res_val || !json_is_true(res_val) ||
+	    (err_val && !json_is_null(err_val)))  {
+		applog(LOG_ERR, "Stratum authentication failed");
+		goto out;
+	}
+
+	ret = true;
+
+out:
+	free(s);
+	if (val)
+		json_decref(val);
+
+	return ret;
+}
+
+static bool stratum_notify(struct stratum_ctx *sctx, json_t *params)
+{
+	const char *job_id, *prevhash, *coinb1, *coinb2, *version, *nbits, *ntime;
+	size_t coinb1_size, coinb2_size;
+	bool clean, ret = false;
+	int merkle_count, i;
+	json_t *merkle_arr;
+	unsigned char **merkle;
+
+	job_id = json_string_value(json_array_get(params, 0));
+	prevhash = json_string_value(json_array_get(params, 1));
+	coinb1 = json_string_value(json_array_get(params, 2));
+	coinb2 = json_string_value(json_array_get(params, 3));
+	merkle_arr = json_array_get(params, 4);
+	if (!merkle_arr || !json_is_array(merkle_arr))
+		goto out;
+	merkle_count = json_array_size(merkle_arr);
+	version = json_string_value(json_array_get(params, 5));
+	nbits = json_string_value(json_array_get(params, 6));
+	ntime = json_string_value(json_array_get(params, 7));
+	clean = json_is_true(json_array_get(params, 8));
+
+	if (!job_id || !prevhash || !coinb1 || !coinb2 || !version || !nbits || !ntime ||
+	    strlen(prevhash) != 64 || strlen(version) != 8 ||
+	    strlen(nbits) != 8 || strlen(ntime) != 8) {
+		applog(LOG_ERR, "Stratum notify: invalid parameters");
+		goto out;
+	}
+	merkle = malloc(merkle_count * sizeof(char *));
+	for (i = 0; i < merkle_count; i++) {
+		const char *s = json_string_value(json_array_get(merkle_arr, i));
+		if (!s || strlen(s) != 64) {
+			while (i--)
+				free(merkle[i]);
+			free(merkle);
+			applog(LOG_ERR, "Stratum notify: invalid Merkle branch");
+			goto out;
+		}
+		merkle[i] = malloc(32);
+		hex2bin(merkle[i], s, 32);
+	}
+
+	pthread_mutex_lock(&sctx->work_lock);
+
+	coinb1_size = strlen(coinb1) / 2;
+	coinb2_size = strlen(coinb2) / 2;
+	sctx->job.coinbase_size = coinb1_size + sctx->xnonce1_size +
+	                          sctx->xnonce2_size + coinb2_size;
+	sctx->job.coinbase = realloc(sctx->job.coinbase, sctx->job.coinbase_size);
+	sctx->job.xnonce2 = sctx->job.coinbase + coinb1_size + sctx->xnonce1_size;
+	hex2bin(sctx->job.coinbase, coinb1, coinb1_size);
+	memcpy(sctx->job.coinbase + coinb1_size, sctx->xnonce1, sctx->xnonce1_size);
+	if (!sctx->job.job_id || strcmp(sctx->job.job_id, job_id))
+		memset(sctx->job.xnonce2, 0, sctx->xnonce2_size);
+	hex2bin(sctx->job.xnonce2 + sctx->xnonce2_size, coinb2, coinb2_size);
+
+	free(sctx->job.job_id);
+	sctx->job.job_id = strdup(job_id);
+	hex2bin(sctx->job.prevhash, prevhash, 32);
+
+	for (i = 0; i < sctx->job.merkle_count; i++)
+		free(sctx->job.merkle[i]);
+	free(sctx->job.merkle);
+	sctx->job.merkle = merkle;
+	sctx->job.merkle_count = merkle_count;
+
+	hex2bin(sctx->job.version, version, 4);
+	hex2bin(sctx->job.nbits, nbits, 4);
+	hex2bin(sctx->job.ntime, ntime, 4);
+	sctx->job.clean = clean;
+
+	sctx->job.diff = sctx->next_diff;
+
+	pthread_mutex_unlock(&sctx->work_lock);
+
+	ret = true;
+
+out:
+	return ret;
+}
+
+static bool stratum_set_difficulty(struct stratum_ctx *sctx, json_t *params)
+{
+	double diff;
+
+	diff = json_number_value(json_array_get(params, 0));
+	if (diff == 0)
+		return false;
+
+	pthread_mutex_lock(&sctx->work_lock);
+	sctx->next_diff = diff;
+	pthread_mutex_unlock(&sctx->work_lock);
+
+	if (opt_debug)
+		applog(LOG_DEBUG, "Stratum difficulty set to %g", diff);
+
+	return true;
+}
+
+static bool stratum_reconnect(struct stratum_ctx *sctx, json_t *params)
+{
+	json_t *port_val;
+	const char *host;
+	int port;
+
+	host = json_string_value(json_array_get(params, 0));
+	port_val = json_array_get(params, 1);
+	if (json_is_string(port_val))
+		port = atoi(json_string_value(port_val));
+	else
+		port = json_integer_value(port_val);
+	if (!host || !port)
+		return false;
+	
+	free(sctx->url);
+	sctx->url = malloc(32 + strlen(host));
+	sprintf(sctx->url, "stratum+tcp://%s:%d", host, port);
+
+	applog(LOG_NOTICE, "Server requested reconnection to %s", sctx->url);
+
+	stratum_disconnect(sctx);
+
+	return true;
+}
+
+static bool stratum_get_version(struct stratum_ctx *sctx, json_t *id)
+{
+	char *s;
+	json_t *val;
+	bool ret;
+	
+	if (!id || json_is_null(id))
+		return false;
+
+	val = json_object();
+	json_object_set(val, "id", id);
+	json_object_set_new(val, "error", json_null());
+	json_object_set_new(val, "result", json_string(USER_AGENT));
+	s = json_dumps(val, 0);
+	ret = stratum_send_line(sctx, s);
+	json_decref(val);
+	free(s);
+
+	return ret;
+}
+
+static bool stratum_show_message(struct stratum_ctx *sctx, json_t *id, json_t *params)
+{
+	char *s;
+	json_t *val;
+	bool ret;
+
+	val = json_array_get(params, 0);
+	if (val)
+		applog(LOG_NOTICE, "MESSAGE FROM SERVER: %s", json_string_value(val));
+	
+	if (!id || json_is_null(id))
+		return true;
+
+	val = json_object();
+	json_object_set(val, "id", id);
+	json_object_set_new(val, "error", json_null());
+	json_object_set_new(val, "result", json_true());
+	s = json_dumps(val, 0);
+	ret = stratum_send_line(sctx, s);
+	json_decref(val);
+	free(s);
+
+	return ret;
+}
+
+bool stratum_handle_method(struct stratum_ctx *sctx, const char *s)
+{
+	json_t *val, *id, *params;
+	json_error_t err;
+	const char *method;
+	bool ret = false;
+
+	val = JSON_LOADS(s, &err);
+	if (!val) {
+		applog(LOG_ERR, "JSON decode failed(%d): %s", err.line, err.text);
+		goto out;
+	}
+
+	method = json_string_value(json_object_get(val, "method"));
+	if (!method)
+		goto out;
+	id = json_object_get(val, "id");
+	params = json_object_get(val, "params");
+
+	if (!strcasecmp(method, "mining.notify")) {
+		ret = stratum_notify(sctx, params);
+		goto out;
+	}
+	if (!strcasecmp(method, "mining.set_difficulty")) {
+		ret = stratum_set_difficulty(sctx, params);
+		goto out;
+	}
+	if (!strcasecmp(method, "client.reconnect")) {
+		ret = stratum_reconnect(sctx, params);
+		goto out;
+	}
+	if (!strcasecmp(method, "client.get_version")) {
+		ret = stratum_get_version(sctx, id);
+		goto out;
+	}
+	if (!strcasecmp(method, "client.show_message")) {
+		ret = stratum_show_message(sctx, id, params);
+		goto out;
+	}
+
+out:
+	if (val)
+		json_decref(val);
+
+	return ret;
+}
+
 struct thread_q *tq_new(void)
 {
 	struct thread_q *tq;