Make nodes submit upstream transactions to the local btcd to speed up blocksolve processing

src/bitcoin.c

@@ -354,3 +354,21 @@ out:
 	json_decref(val);
 	return ret;
 }
+
+void submit_txn(connsock_t *cs, char *params)
+{
+	char *rpc_req;
+	json_t *val;
+	int len;
+
+	if (unlikely(!cs->alive))
+		return;
+
+	len = strlen(params) + 64;
+	rpc_req = ckalloc(len);
+	sprintf(rpc_req, "{\"method\": \"sendrawtransaction\", \"params\": [\"%s\"]}\n", params);
+	val = json_rpc_call(cs, rpc_req);
+	dealloc(rpc_req);
+	/* We don't really care about the result */
+	json_decref(val);
+}

src/bitcoin.h

@@ -39,5 +39,6 @@ int get_blockcount(connsock_t *cs);
 bool get_blockhash(connsock_t *cs, int height, char *hash);
 bool get_bestblockhash(connsock_t *cs, char *hash);
 bool submit_block(connsock_t *cs, char *params);
+void submit_txn(connsock_t *cs, char *params);
 
 #endif /* BITCOIN_H */

src/ckpool.h

@@ -88,6 +88,8 @@ struct connsock {
 	ckpool_t *ckp;
 	/* Semaphore used to serialise request/responses */
 	sem_t sem;
+
+	bool alive;
 };
 
 typedef struct connsock connsock_t;
@@ -181,6 +183,10 @@ struct ckpool_instance {
 	proc_instance_t stratifier;
 	proc_instance_t connector;
 
+	bool generator_ready;
+	bool stratifier_ready;
+	bool connector_ready;
+
 	/* Threads of main process */
 	pthread_t pth_listener;
 	pthread_t pth_watchdog;

src/generator.c

@@ -217,7 +217,7 @@ static bool server_alive(ckpool_t *ckp, server_instance_t *si, bool pinging)
 		LOGWARNING("Invalid btcaddress: %s !", ckp->btcaddress);
 		goto out;
 	}
-	si->alive = ret = true;
+	si->alive = cs->alive = ret = true;
 	LOGNOTICE("Server alive: %s:%s", cs->url, cs->port);
 out:
 	/* Close the file handle */
@@ -297,7 +297,6 @@ static void gen_loop(proc_instance_t *pi)
 	server_instance_t *si = NULL, *old_si;
 	unix_msg_t *umsg = NULL;
 	ckpool_t *ckp = pi->ckp;
-	bool started = false;
 	char *buf = NULL;
 	connsock_t *cs;
 	gbtbase_t *gbt;
@@ -309,8 +308,8 @@ reconnect:
 	si = live_server(ckp);
 	if (!si)
 		goto out;
-	if (unlikely(!started)) {
+	if (unlikely(!ckp->generator_ready)) {
-		started = true;
+		ckp->generator_ready = true;
 		LOGWARNING("%s generator ready", ckp->name);
 	}
 
@@ -339,7 +338,7 @@ retry:
 		if (!gen_gbtbase(cs, gbt)) {
 			LOGWARNING("Failed to get block template from %s:%s",
 				   cs->url, cs->port);
-			si->alive = false;
+			si->alive = cs->alive = false;
 			send_unix_msg(umsg->sockd, "Failed");
 			goto reconnect;
 		} else {
@@ -355,7 +354,7 @@ retry:
 		else if (!get_bestblockhash(cs, hash)) {
 			LOGINFO("No best block hash support from %s:%s",
 				cs->url, cs->port);
-			si->alive = false;
+			si->alive = cs->alive = false;
 			send_unix_msg(umsg->sockd, "failed");
 		} else {
 			send_unix_msg(umsg->sockd, hash);
@@ -366,13 +365,13 @@ retry:
 		if (si->notify)
 			send_unix_msg(umsg->sockd, "notify");
 		else if ((height = get_blockcount(cs)) == -1) {
-			si->alive = false;
+			si->alive = cs->alive = false;
 			send_unix_msg(umsg->sockd,  "failed");
 			goto reconnect;
 		} else {
 			LOGDEBUG("Height: %d", height);
 			if (!get_blockhash(cs, height, hash)) {
-				si->alive = false;
+				si->alive = cs->alive = false;
 				send_unix_msg(umsg->sockd, "failed");
 				goto reconnect;
 			} else {
@@ -2708,7 +2707,6 @@ static void proxy_loop(proc_instance_t *pi)
 	gdata_t *gdata = ckp->gdata;
 	unix_msg_t *umsg = NULL;
 	connsock_t *cs = NULL;
-	bool started = false;
 	char *buf = NULL;
 
 reconnect:
@@ -2737,8 +2735,8 @@ reconnect:
 			   proxi->id, proxi->url, ckp->passthrough ? " in passthrough mode" : "");
 	}
 
-	if (unlikely(!started)) {
+	if (unlikely(!ckp->generator_ready)) {
-		started = true;
+		ckp->generator_ready = true;
 		LOGWARNING("%s generator ready", ckp->name);
 	}
 retry:
@@ -2778,6 +2776,12 @@ retry:
 		memset(buf + 12 + 64, 0, 1);
 		sprintf(blockmsg, "%sblock:%s", ret ? "" : "no", buf + 12);
 		send_proc(ckp->stratifier, blockmsg);
+	} else if (cmdmatch(buf, "submittxn:")) {
+		if (unlikely(strlen(buf) < 11)) {
+			LOGWARNING("Got zero length submittxn");
+			goto retry;
+		}
+		submit_txn(cs, buf + 10);
 	} else if (cmdmatch(buf, "loglevel")) {
 		sscanf(buf, "loglevel=%d", &ckp->loglevel);
 	} else if (cmdmatch(buf, "ping")) {

src/stratifier.c

@@ -6466,7 +6466,20 @@ out:
 	free(buf);
 }
 
-static void add_node_txns(sdata_t *sdata, const json_t *val)
+/* Submit the transactions in node mode so the local btcd has all the
+ * transactions that will go into the next blocksolve. */
+static void submit_transaction(ckpool_t *ckp, const char *hash)
+{
+	char *buf;
+
+	if (unlikely(!ckp->generator_ready))
+		return;
+	ASPRINTF(&buf, "submittxn:%s", hash);
+	send_generator(ckp, buf, GEN_LAX);
+	free(buf);
+}
+
+static void add_node_txns(ckpool_t *ckp, sdata_t *sdata, const json_t *val)
 {
 	json_t *txn_array, *txn_val, *data_val, *hash_val;
 	txntable_t *txn;
@@ -6494,6 +6507,7 @@ static void add_node_txns(sdata_t *sdata, const json_t *val)
 			txn->refcount = 100;
 			continue;
 		}
+		submit_transaction(ckp, data);
 		txn = ckzalloc(sizeof(txntable_t));
 		memcpy(txn->hash, hash, 65);
 		txn->data = strdup(data);
@@ -6579,7 +6593,7 @@ static void parse_node_msg(ckpool_t *ckp, sdata_t *sdata, json_t *val)
 	LOGDEBUG("Got node method %d:%s", msg_type, stratum_msgs[msg_type]);
 	switch (msg_type) {
 		case SM_TRANSACTIONS:
-			add_node_txns(sdata, val);
+			add_node_txns(ckp, sdata, val);
 			break;
 		case SM_WORKINFO:
 			add_node_base(ckp, val);
@@ -6660,11 +6674,12 @@ static void srecv_process(ckpool_t *ckp, json_t *val)
 	msg->json_msg = val;
 	val = json_object_get(msg->json_msg, "client_id");
 	if (unlikely(!val)) {
-		buf = json_dumps(val, JSON_COMPACT);
 		if (ckp->node)
 			parse_node_msg(ckp, sdata, msg->json_msg);
-		else
+		else {
+			buf = json_dumps(val, JSON_COMPACT);
 			LOGWARNING("Failed to extract client_id from connector json smsg %s", buf);
+		}
 		goto out;
 	}