diff --git a/src/bitcoin.c b/src/bitcoin.c index 487f0aa5..987f241c 100644 --- a/src/bitcoin.c +++ b/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); +} diff --git a/src/bitcoin.h b/src/bitcoin.h index 9ba4534d..49e0ba82 100644 --- a/src/bitcoin.h +++ b/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 */ diff --git a/src/ckpool.h b/src/ckpool.h index 520c6eb3..f8270560 100644 --- a/src/ckpool.h +++ b/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; diff --git a/src/generator.c b/src/generator.c index 2908421d..dc735b23 100644 --- a/src/generator.c +++ b/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)) { - started = true; + if (unlikely(!ckp->generator_ready)) { + 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)) { - started = true; + if (unlikely(!ckp->generator_ready)) { + 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")) { diff --git a/src/stratifier.c b/src/stratifier.c index ff137832..ab22bd9e 100644 --- a/src/stratifier.c +++ b/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; }