Use async send proc in the stratifier

src/stratifier.c

@@ -828,7 +828,7 @@ static char *send_recv_generator(ckpool_t *ckp, const char *msg, const int prio)
 	return buf;
 }
 
-static void send_generator(const ckpool_t *ckp, const char *msg, const int prio)
+static void send_generator(ckpool_t *ckp, const char *msg, const int prio)
 {
 	sdata_t *sdata = ckp->data;
 	bool set;
@@ -838,7 +838,7 @@ static void send_generator(const ckpool_t *ckp, const char *msg, const int prio)
 		set = true;
 	} else
 		set = false;
-	send_proc(ckp->generator, msg);
+	async_send_proc(ckp, ckp->generator, msg);
 	if (set)
 		sdata->gen_priority = 0;
 }
@@ -963,7 +963,7 @@ static void connector_drop_client(ckpool_t *ckp, const int64_t id)
 
 	LOGDEBUG("Stratifier requesting connector drop client %"PRId64, id);
 	snprintf(buf, 255, "dropclient=%"PRId64, id);
-	send_proc(ckp->connector, buf);
+	async_send_proc(ckp, ckp->connector, buf);
 }
 
 static void drop_allclients(ckpool_t *ckp)
@@ -1009,12 +1009,10 @@ static sdata_t *duplicate_sdata(const sdata_t *sdata)
 	memcpy(dsdata->donkeytxnbin, sdata->donkeytxnbin, 25);
 
 	/* Use the same work queues for all subproxies */
+	dsdata->ckwqs = sdata->ckwqs;
 	dsdata->ssends = sdata->ssends;
-	dsdata->srecvs = sdata->srecvs;
 	dsdata->ckdbq = sdata->ckdbq;
-	dsdata->sshareq = sdata->sshareq;
 	dsdata->sauthq = sdata->sauthq;
-	dsdata->stxnq = sdata->stxnq;
 
 	/* Give the sbuproxy its own workbase list and lock */
 	cklock_init(&dsdata->workbase_lock);
@@ -1138,7 +1136,7 @@ out_unlock:
 
 static void reconnect_client(sdata_t *sdata, stratum_instance_t *client);
 
-static void generator_recruit(const ckpool_t *ckp)
+static void generator_recruit(ckpool_t *ckp)
 {
 	LOGINFO("Stratifer requesting more proxies from generator");
 	send_generator(ckp, "recruit", GEN_PRIORITY);
@@ -2487,7 +2485,7 @@ static void stratum_send_message(sdata_t *sdata, const stratum_instance_t *clien
  * in proxy mode where we find a subproxy based on the current proxy with room
  * for more clients. Signal the generator to recruit more subproxies if we are
  * running out of room. */
-static sdata_t *select_sdata(const ckpool_t *ckp, sdata_t *ckp_sdata)
+static sdata_t *select_sdata(ckpool_t *ckp, sdata_t *ckp_sdata)
 {
 	proxy_t *current, *proxy, *subproxy, *best = NULL, *tmp, *tmpsub;
 	int best_id, best_subid = 0;
@@ -3895,6 +3893,7 @@ static void send_transactions(ckpool_t *ckp, json_params_t *jp);
 static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64_t client_id,
 			 json_t *id_val, json_t *method_val, json_t *params_val, const char *address)
 {
+	ckpool_t *ckp = client->ckp;
 	const char *method;
 
 	/* Random broken clients send something not an integer as the id so we
@@ -3941,14 +3940,14 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
 		 * to it since it's unauthorised. Set the flag just in case. */
 		client->authorised = false;
 		snprintf(buf, 255, "passthrough=%"PRId64, client_id);
-		send_proc(client->ckp->connector, buf);
+		async_send_proc(ckp, ckp->connector, buf);
 		return;
 	}
 
 	/* We should only accept subscribed requests from here on */
 	if (!client->subscribed) {
 		LOGINFO("Dropping unsubscribed client %"PRId64, client_id);
-		connector_drop_client(client->ckp, client_id);
+		connector_drop_client(ckp, client_id);
 		return;
 	}
 
@@ -3970,7 +3969,7 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
 		 * stratifier process to restart since it will have lost all
 		 * the stratum instance data. Clients will just reconnect. */
 		LOGINFO("Dropping unauthorised client %"PRId64, client_id);
-		connector_drop_client(client->ckp, client_id);
+		connector_drop_client(ckp, client_id);
 		return;
 	}
 
@@ -4144,7 +4143,7 @@ static void ssend_process(ckpool_t *ckp, smsg_t *msg)
 	 * connector process to be delivered */
 	json_object_set_new_nocheck(msg->json_msg, "client_id", json_integer(msg->client_id));
 	s = json_dumps(msg->json_msg, 0);
-	send_proc(ckp->connector, s);
+	async_send_proc(ckp, ckp->connector, s);
 	free(s);
 	free_smsg(msg);
 }