diff --git a/src/ckpool.c b/src/ckpool.c
index cf79794f..a1c4a514 100644
--- a/src/ckpool.c
+++ b/src/ckpool.c
@@ -134,39 +134,6 @@ static void *ckmsg_queue(void *arg)
 	return NULL;
 }
 
-/* Generic workqueue function and message receiving and parsing thread */
-static void *ckwq_queue(void *arg)
-{
-	ckwq_t *ckwq = (ckwq_t *)arg;
-	ckpool_t *ckp = ckwq->ckp;
-
-	pthread_detach(pthread_self());
-	rename_proc(ckwq->name);
-
-	while (42) {
-		ckwqmsg_t *wqmsg;
-		tv_t now;
-		ts_t abs;
-
-		mutex_lock(ckwq->lock);
-		tv_time(&now);
-		tv_to_ts(&abs, &now);
-		abs.tv_sec++;
-		if (!ckwq->wqmsgs)
-			cond_timedwait(ckwq->cond, ckwq->lock, &abs);
-		wqmsg = ckwq->wqmsgs;
-		if (wqmsg)
-			DL_DELETE(ckwq->wqmsgs, wqmsg);
-		mutex_unlock(ckwq->lock);
-
-		if (!wqmsg)
-			continue;
-		wqmsg->func(ckp, wqmsg->data);
-		free(wqmsg);
-	}
-	return NULL;
-}
-
 ckmsgq_t *create_ckmsgq(ckpool_t *ckp, const char *name, const void *func)
 {
 	ckmsgq_t *ckmsgq = ckzalloc(sizeof(ckmsgq_t));
@@ -207,29 +174,6 @@ ckmsgq_t *create_ckmsgqs(ckpool_t *ckp, const char *name, const void *func, cons
 	return ckmsgq;
 }
 
-ckwq_t *create_ckwqs(ckpool_t *ckp, const char *name, const int count)
-{
-	ckwq_t *ckwq = ckzalloc(sizeof(ckwq_t) * count);
-	mutex_t *lock;
-	pthread_cond_t *cond;
-	int i;
-
-	lock = ckalloc(sizeof(mutex_t));
-	cond = ckalloc(sizeof(pthread_cond_t));
-	mutex_init(lock);
-	cond_init(cond);
-
-	for (i = 0; i < count; i++) {
-		snprintf(ckwq[i].name, 15, "%.6swq%d", name, i);
-		ckwq[i].ckp = ckp;
-		ckwq[i].lock = lock;
-		ckwq[i].cond = cond;
-		create_pthread(&ckwq[i].pth, ckwq_queue, &ckwq[i]);
-	}
-
-	return ckwq;
-}
-
 /* Generic function for adding messages to a ckmsgq linked list and signal the
  * ckmsgq parsing thread to wake up and process it. */
 void ckmsgq_add(ckmsgq_t *ckmsgq, void *data)
@@ -241,24 +185,10 @@ void ckmsgq_add(ckmsgq_t *ckmsgq, void *data)
 	mutex_lock(ckmsgq->lock);
 	ckmsgq->messages++;
 	DL_APPEND(ckmsgq->msgs, msg);
-	pthread_cond_broadcast(ckmsgq->cond);
+	pthread_cond_signal(ckmsgq->cond);
 	mutex_unlock(ckmsgq->lock);
 }
 
-void ckwq_add(ckwq_t *ckwq, const void *func, void *data)
-{
-	ckwqmsg_t *wqmsg = ckalloc(sizeof(ckwqmsg_t));
-
-	wqmsg->func = func;
-	wqmsg->data = data;
-
-	mutex_lock(ckwq->lock);
-	ckwq->messages++;
-	DL_APPEND(ckwq->wqmsgs, wqmsg);
-	pthread_cond_broadcast(ckwq->cond);
-	mutex_unlock(ckwq->lock);
-}
-
 /* Return whether there are any messages queued in the ckmsgq linked list. */
 bool ckmsgq_empty(ckmsgq_t *ckmsgq)
 {
@@ -568,14 +498,32 @@ out:
 
 static void childsighandler(const int sig);
 
-/* Send a single message to a process instance when there will be no response,
- * closing the socket immediately. */
-void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line)
+struct proc_message {
+	proc_instance_t *pi;
+	char *msg;
+	const char *file;
+	const char *func;
+	int line;
+};
+
+/* Send all one way messages asynchronously so we can wait till the receiving
+ * end closes the socket to ensure all messages are received but no deadlocks
+ * can occur with 2 processes waiting for each other's socket closure. */
+void *async_send_proc(void *arg)
 {
+	struct proc_message *pm = (struct proc_message *)arg;
+	proc_instance_t *pi = pm->pi;
+	char *msg = pm->msg;
+	const char *file = pm->file;
+	const char *func = pm->func;
+	int line = pm->line;
+
 	char *path = pi->us.path;
 	bool ret = false;
 	int sockd;
 
+	pthread_detach(pthread_self());
+
 	if (unlikely(!path || !strlen(path))) {
 		LOGERR("Attempted to send message %s to null path in send_proc", msg ? msg : "");
 		goto out;
@@ -586,14 +534,16 @@ void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const ch
 	}
 	/* At startup the pid fields are not set up before some processes are
 	 * forked so they never inherit them. */
-	if (unlikely(!pi->pid))
+	if (unlikely(!pi->pid)) {
 		pi->pid = get_proc_pid(pi);
-	if (!pi->pid) {
-		LOGALERT("Attempting to send message %s to non existent process %s", msg, pi->processname);
-		return;
+		if (!pi->pid) {
+			LOGALERT("Attempting to send message %s to non existent process %s", msg, pi->processname);
+			goto out_nofail;
+		}
 	}
 	if (unlikely(kill_pid(pi->pid, 0))) {
-		LOGALERT("Attempting to send message %s to non existent process %s", msg, pi->processname);
+		LOGALERT("Attempting to send message %s to non existent process %s pid %d",
+			 msg, pi->processname, pi->pid);
 		goto out;
 	}
 	sockd = open_unix_client(path);
@@ -613,41 +563,25 @@ out:
 		LOGERR("Failure in send_proc from %s %s:%d", file, func, line);
 		childsighandler(15);
 	}
-}
-
-struct proc_message {
-	proc_instance_t *pi;
-	char *msg;
-	const char *file;
-	const char *func;
-	int line;
-};
-
-static void asp_send(ckpool_t __maybe_unused *ckp, struct proc_message *pm)
-{
-	_send_proc(pm->pi, pm->msg, pm->file, pm->func, pm->line);
-	free(pm->msg);
+out_nofail:
+	free(msg);
 	free(pm);
+	return NULL;
 }
 
-/* Fore sending asynchronous messages to another process, the sending process
- * must have ckwqs of its own, referenced in the ckpool structure */
-void _async_send_proc(ckpool_t *ckp, proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line)
+/* Send a single message to a process instance when there will be no response,
+ * closing the socket immediately. */
+void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line)
 {
-	struct proc_message *pm;
+	struct proc_message *pm = ckalloc(sizeof(struct proc_message));
+	pthread_t pth;
 
-	if (unlikely(!ckp->ckwqs)) {
-		LOGALERT("Workqueues not set up in async_send_proc!");
-		_send_proc(pi, msg, file, func, line);
-		return;
-	}
-	pm = ckzalloc(sizeof(struct proc_message));
 	pm->pi = pi;
 	pm->msg = strdup(msg);
 	pm->file = file;
 	pm->func = func;
 	pm->line = line;
-	ckwq_add(ckp->ckwqs, &asp_send, pm);
+	create_pthread(&pth, async_send_proc, pm);
 }
 
 /* Send a single message to a process instance and retrieve the response, then
diff --git a/src/ckpool.h b/src/ckpool.h
index 74dc5e3e..771ac692 100644
--- a/src/ckpool.h
+++ b/src/ckpool.h
@@ -22,22 +22,13 @@
 struct ckpool_instance;
 typedef struct ckpool_instance ckpool_t;
 
-typedef struct ckmsg ckmsg_t;
-
 struct ckmsg {
-	ckmsg_t *next;
-	ckmsg_t *prev;
+	struct ckmsg *next;
+	struct ckmsg *prev;
 	void *data;
 };
 
-typedef struct ckwqmsg ckwqmsg_t;
-
-struct ckwqmsg {
-	ckwqmsg_t *next;
-	ckwqmsg_t *prev;
-	void *data;
-	void (*func)(ckpool_t *, void *);
-};
+typedef struct ckmsg ckmsg_t;
 
 struct ckmsgq {
 	ckpool_t *ckp;
@@ -52,18 +43,6 @@ struct ckmsgq {
 
 typedef struct ckmsgq ckmsgq_t;
 
-struct ckwq {
-	ckpool_t *ckp;
-	char name[16];
-	pthread_t pth;
-	mutex_t *lock;
-	pthread_cond_t *cond;
-	ckwqmsg_t *wqmsgs;
-	int64_t messages;
-};
-
-typedef struct ckwq ckwq_t;
-
 struct proc_instance {
 	ckpool_t *ckp;
 	unixsock_t us;
@@ -213,8 +192,6 @@ struct ckpool_instance {
 
 	/* Private data for each process */
 	void *data;
-	/* Private generic workqueues if this process has them */
-	ckwq_t *ckwqs;
 };
 
 #ifdef USE_CKDB
@@ -227,9 +204,7 @@ struct ckpool_instance {
 
 ckmsgq_t *create_ckmsgq(ckpool_t *ckp, const char *name, const void *func);
 ckmsgq_t *create_ckmsgqs(ckpool_t *ckp, const char *name, const void *func, const int count);
-ckwq_t *create_ckwqs(ckpool_t *ckp, const char *name, const int count);
 void ckmsgq_add(ckmsgq_t *ckmsgq, void *data);
-void ckwq_add(ckwq_t *ckwq, const void *func, void *data);
 bool ckmsgq_empty(ckmsgq_t *ckmsgq);
 
 ckpool_t *global_ckp;
@@ -239,8 +214,6 @@ void empty_buffer(connsock_t *cs);
 int read_socket_line(connsock_t *cs, const int timeout);
 void _send_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line);
 #define send_proc(pi, msg) _send_proc(pi, msg, __FILE__, __func__, __LINE__)
-void _async_send_proc(ckpool_t *ckp, proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line);
-#define async_send_proc(ckp, pi, msg) _async_send_proc(ckp, pi, msg, __FILE__, __func__, __LINE__)
 char *_send_recv_proc(proc_instance_t *pi, const char *msg, const char *file, const char *func, const int line);
 #define send_recv_proc(pi, msg) _send_recv_proc(pi, msg, __FILE__, __func__, __LINE__)
 char *_send_recv_ckdb(const ckpool_t *ckp, const char *msg, const char *file, const char *func, const int line);
diff --git a/src/connector.c b/src/connector.c
index ece13099..f6d9f1ab 100644
--- a/src/connector.c
+++ b/src/connector.c
@@ -97,8 +97,6 @@ struct connector_data {
 	/* For protecting the pending sends list */
 	mutex_t sender_lock;
 	pthread_cond_t sender_cond;
-
-	ckwq_t *ckwqs;
 };
 
 typedef struct connector_data cdata_t;
@@ -244,7 +242,7 @@ static void stratifier_drop_client(ckpool_t *ckp, int64_t id)
 	char buf[256];
 
 	sprintf(buf, "dropclient=%"PRId64, id);
-	async_send_proc(ckp, ckp->stratifier, buf);
+	send_proc(ckp->stratifier, buf);
 }
 
 /* Invalidate this instance. Remove them from the hashtables we look up
@@ -363,9 +361,9 @@ reparse:
 		 * filtered by the stratifier. */
 		if (likely(client->fd != -1)) {
 			if (ckp->passthrough)
-				async_send_proc(ckp, ckp->generator, s);
+				send_proc(ckp->generator, s);
 			else
-				async_send_proc(ckp, ckp->stratifier, s);
+				send_proc(ckp->stratifier, s);
 		}
 
 		free(s);
@@ -863,8 +861,6 @@ int connector(proc_instance_t *pi)
 	LOGWARNING("%s connector starting", ckp->name);
 	ckp->data = cdata;
 	cdata->ckp = ckp;
-	/* Connector only requires one work queue */
-	ckp->ckwqs = cdata->ckwqs = create_ckwqs(ckp, "conn", 1);
 
 	if (!ckp->serverurls)
 		cdata->serverfd = ckalloc(sizeof(int *));
diff --git a/src/generator.c b/src/generator.c
index 233f25e6..90f054f4 100644
--- a/src/generator.c
+++ b/src/generator.c
@@ -132,7 +132,6 @@ struct generator_data {
 	proxy_instance_t *proxy_list; /* Linked list of all active proxies */
 	int proxy_notify_id;	// Globally increasing notify id
 	ckmsgq_t *srvchk;	// Server check message queue
-	ckwq_t *ckwqs;
 };
 
 typedef struct generator_data gdata_t;
@@ -222,7 +221,7 @@ retry:
 	cs = &alive->cs;
 	LOGINFO("Connected to live server %s:%s", cs->url, cs->port);
 out:
-	async_send_proc(ckp, ckp->connector, alive ? "accept" : "reject");
+	send_proc(ckp->connector, alive ? "accept" : "reject");
 	return alive;
 }
 
@@ -368,7 +367,7 @@ retry:
 		ret = submit_block(cs, buf + 12 + 64 + 1);
 		memset(buf + 12 + 64, 0, 1);
 		sprintf(blockmsg, "%sblock:%s", ret ? "" : "no", buf + 12);
-		async_send_proc(ckp, ckp->stratifier, blockmsg);
+		send_proc(ckp->stratifier, blockmsg);
 	} else if (cmdmatch(buf, "checkaddr:")) {
 		if (validate_address(cs, buf + 10))
 			send_unix_msg(sockd, "true");
@@ -1298,22 +1297,22 @@ static void *proxy_recv(void *arg)
 		if (ret < 1) {
 			/* Send ourselves a reconnect message */
 			LOGWARNING("Failed to read_socket_line in proxy_recv, attempting reconnect");
-			async_send_proc(ckp, ckp->generator, "reconnect");
+			send_proc(ckp->generator, "reconnect");
 			break;
 		}
 		if (parse_method(proxi, cs->buf)) {
 			if (proxi->notified) {
-				async_send_proc(ckp, ckp->stratifier, "notify");
+				send_proc(ckp->stratifier, "notify");
 				proxi->notified = false;
 			}
 			if (proxi->diffed) {
-				async_send_proc(ckp, ckp->stratifier, "diff");
+				send_proc(ckp->stratifier, "diff");
 				proxi->diffed = false;
 			}
 			if (proxi->reconnect) {
 				proxi->reconnect = false;
 				LOGWARNING("Reconnect issue, dropping existing connection");
-				async_send_proc(ckp, ckp->generator, "reconnect");
+				send_proc(ckp->generator, "reconnect");
 				break;
 			}
 			continue;
@@ -1403,13 +1402,13 @@ static void *passthrough_recv(void *arg)
 		if (ret < 1) {
 			/* Send ourselves a reconnect message */
 			LOGWARNING("Failed to read_socket_line in proxy_recv, attempting reconnect");
-			async_send_proc(ckp, ckp->generator, "reconnect");
+			send_proc(ckp->generator, "reconnect");
 			break;
 		}
 		/* Simply forward the message on, as is, to the connector to
 		 * process. Possibly parse parameters sent by upstream pool
 		 * here */
-		async_send_proc(ckp, ckp->connector, cs->buf);
+		send_proc(ckp->connector, cs->buf);
 	}
 	return NULL;
 }
@@ -1524,8 +1523,8 @@ retry:
 		}
 	}
 	if (!alive) {
-		async_send_proc(ckp, ckp->connector, "reject");
-		async_send_proc(ckp, ckp->stratifier, "dropall");
+		send_proc(ckp->connector, "reject");
+		send_proc(ckp->stratifier, "dropall");
 		LOGWARNING("Failed to connect to any servers as proxy, retrying in 5s!");
 		sleep(5);
 		goto retry;
@@ -1544,7 +1543,7 @@ retry:
 		create_pthread(&alive->pth_psend, proxy_send, alive);
 	}
 out:
-	async_send_proc(ckp, ckp->connector, alive ? "accept" : "reject");
+	send_proc(ckp->connector, alive ? "accept" : "reject");
 	return alive;
 }
 
@@ -1553,8 +1552,8 @@ static void kill_proxy(ckpool_t *ckp, proxy_instance_t *proxi)
 	notify_instance_t *ni, *tmp;
 	connsock_t *cs;
 
-	async_send_proc(ckp, ckp->stratifier, "reconnect");
-	async_send_proc(ckp, ckp->connector, "reject");
+	send_proc(ckp->stratifier, "reconnect");
+	send_proc(ckp->connector, "reject");
 
 	if (!proxi) // This shouldn't happen
 		return;
@@ -1604,8 +1603,8 @@ reconnect:
 	/* We've just subscribed and authorised so tell the stratifier to
 	 * retrieve the first subscription. */
 	if (!ckp->passthrough) {
-		async_send_proc(ckp, ckp->stratifier, "subscribe");
-		async_send_proc(ckp, ckp->stratifier, "notify");
+		send_proc(ckp->stratifier, "subscribe");
+		send_proc(ckp->stratifier, "notify");
 		proxi->notified = false;
 	}
 
@@ -1793,7 +1792,7 @@ static void server_watchdog(ckpool_t *ckp, server_instance_t *cursi)
 			break;
 	}
 	if (alive)
-		async_send_proc(ckp, ckp->generator, "reconnect");
+		send_proc(ckp->generator, "reconnect");
 }
 
 static void proxy_watchdog(ckpool_t *ckp, server_instance_t *cursi)
@@ -1839,7 +1838,7 @@ static void proxy_watchdog(ckpool_t *ckp, server_instance_t *cursi)
 			break;
 	}
 	if (alive)
-		async_send_proc(ckp, ckp->generator, "reconnect");
+		send_proc(ckp->generator, "reconnect");
 }
 
 
@@ -1852,8 +1851,6 @@ int generator(proc_instance_t *pi)
 	LOGWARNING("%s generator starting", ckp->name);
 	gdata = ckzalloc(sizeof(gdata_t));
 	ckp->data = gdata;
-	/* Generator only requires one work queue */
-	ckp->ckwqs = gdata->ckwqs = create_ckwqs(ckp, "gen", 1);
 	if (ckp->proxy) {
 		gdata->srvchk = create_ckmsgq(ckp, "prxchk", &proxy_watchdog);
 		ret = proxy_mode(ckp, pi);
diff --git a/src/stratifier.c b/src/stratifier.c
index b46bafc0..859edc6e 100644
--- a/src/stratifier.c
+++ b/src/stratifier.c
@@ -313,10 +313,12 @@ struct stratifier_data {
 	char lasthash[68];
 	char lastswaphash[68];
 
-	ckwq_t *ckwqs;		// Generic workqueues
 	ckmsgq_t *ssends;	// Stratum sends
+	ckmsgq_t *srecvs;	// Stratum receives
 	ckmsgq_t *ckdbq;	// ckdb
+	ckmsgq_t *sshareq;	// Stratum share sends
 	ckmsgq_t *sauthq;	// Stratum authorisations
+	ckmsgq_t *stxnq;	// Transaction requests
 
 	int64_t user_instance_id;
 
@@ -795,26 +797,38 @@ static void send_generator(ckpool_t *ckp, const char *msg, const int prio)
 		set = true;
 	} else
 		set = false;
-	async_send_proc(ckp, ckp->generator, msg);
+	send_proc(ckp->generator, msg);
 	if (set)
 		sdata->gen_priority = 0;
 }
 
+struct update_req {
+	pthread_t *pth;
+	ckpool_t *ckp;
+	int prio;
+};
+
 static void broadcast_ping(sdata_t *sdata);
 
 /* This function assumes it will only receive a valid json gbt base template
  * since checking should have been done earlier, and creates the base template
  * for generating work templates. */
-static void do_update(ckpool_t *ckp, int *prio)
+static void *do_update(void *arg)
 {
+	struct update_req *ur = (struct update_req *)arg;
+	ckpool_t *ckp = ur->ckp;
 	sdata_t *sdata = ckp->data;
 	bool new_block = false;
+	int prio = ur->prio;
 	bool ret = false;
 	workbase_t *wb;
 	json_t *val;
 	char *buf;
 
-	buf = send_recv_generator(ckp, "getbase", *prio);
+	pthread_detach(pthread_self());
+	rename_proc("updater");
+
+	buf = send_recv_generator(ckp, "getbase", prio);
 	if (unlikely(!buf)) {
 		LOGNOTICE("Get base in update_base delayed due to higher priority request");
 		goto out;
@@ -874,17 +888,21 @@ out:
 		LOGINFO("Broadcast ping due to failed stratum base update");
 		broadcast_ping(sdata);
 	}
-	free(buf);
-	free(prio);
+	dealloc(buf);
+	free(ur->pth);
+	free(ur);
+	return NULL;
 }
 
 static void update_base(ckpool_t *ckp, const int prio)
 {
-	int *pprio = ckalloc(sizeof(int));
-	sdata_t *sdata = ckp->data;
+	struct update_req *ur = ckalloc(sizeof(struct update_req));
+	pthread_t *pth = ckalloc(sizeof(pthread_t));
 
-	*pprio = prio;
-	ckwq_add(sdata->ckwqs, &do_update, pprio);
+	ur->pth = pth;
+	ur->ckp = ckp;
+	ur->prio = prio;
+	create_pthread(pth, do_update, ur);
 }
 
 static void __kill_instance(stratum_instance_t *client)
@@ -928,7 +946,7 @@ static void drop_allclients(ckpool_t *ckp)
 			client->dropped = true;
 		kills++;
 		sprintf(buf, "dropclient=%"PRId64, client_id);
-		async_send_proc(ckp, ckp->connector, buf);
+		send_proc(ckp->connector, buf);
 	}
 	HASH_ITER(hh, sdata->disconnected_instances, client, tmp) {
 		disconnects++;
@@ -1633,21 +1651,6 @@ static void ckmsgq_stats(ckmsgq_t *ckmsgq, const int size, json_t **val)
 	JSON_CPACK(*val, "{si,si,si}", "count", objects, "memory", memsize, "generated", generated);
 }
 
-static void ckwq_stats(ckwq_t *ckwq, const int size, json_t **val)
-{
-	int objects, generated;
-	int64_t memsize;
-	ckwqmsg_t *wqmsg;
-
-	mutex_lock(ckwq->lock);
-	DL_COUNT(ckwq->wqmsgs, wqmsg, objects);
-	generated = ckwq->messages;
-	mutex_unlock(ckwq->lock);
-
-	memsize = (sizeof(ckwqmsg_t) + size) * objects;
-	JSON_CPACK(*val, "{si,si,si}", "count", objects, "memory", memsize, "generated", generated);
-}
-
 static char *stratifier_stats(ckpool_t *ckp, sdata_t *sdata)
 {
 	json_t *val = json_object(), *subval;
@@ -1694,14 +1697,15 @@ static char *stratifier_stats(ckpool_t *ckp, sdata_t *sdata)
 
 	ckmsgq_stats(sdata->ssends, sizeof(smsg_t), &subval);
 	json_set_object(val, "ssends", subval);
-
 	/* Don't know exactly how big the string is so just count the pointer for now */
-	ckwq_stats(sdata->ckwqs, sizeof(char *) + sizeof(void *), &subval);
-	json_set_object(val, "ckwqs", subval);
+	ckmsgq_stats(sdata->srecvs, sizeof(char *), &subval);
+	json_set_object(val, "srecvs", subval);
 	if (!CKP_STANDALONE(ckp)) {
 		ckmsgq_stats(sdata->ckdbq, sizeof(char *), &subval);
 		json_set_object(val, "ckdbq", subval);
 	}
+	ckmsgq_stats(sdata->stxnq, sizeof(json_params_t), &subval);
+	json_set_object(val, "stxnq", subval);
 
 	buf = json_dumps(val, JSON_NO_UTF8 | JSON_PRESERVE_ORDER);
 	json_decref(val);
@@ -1709,8 +1713,6 @@ static char *stratifier_stats(ckpool_t *ckp, sdata_t *sdata)
 	return buf;
 }
 
-static void srecv_process(ckpool_t *ckp, char *buf);
-
 static int stratum_loop(ckpool_t *ckp, proc_instance_t *pi)
 {
 	int sockd, ret = 0, selret = 0;
@@ -1765,7 +1767,7 @@ retry:
 		/* The bulk of the messages will be received json from the
 		 * connector so look for this first. The srecv_process frees
 		 * the buf heap ram */
-		ckwq_add(sdata->ckwqs, &srecv_process, buf);
+		ckmsgq_add(sdata->srecvs, buf);
 		Close(sockd);
 		buf = NULL;
 		goto retry;
@@ -3287,14 +3289,10 @@ static void suggest_diff(stratum_instance_t *client, const char *method, const j
 	stratum_send_diff(sdata, client);
 }
 
-static void sshare_process(ckpool_t *ckp, json_params_t *jp);
-static void send_transactions(ckpool_t *ckp, json_params_t *jp);
-
 /* Enter with client holding ref count */
 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;
 	char buf[256];
 
@@ -3305,7 +3303,7 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
 	if (likely(cmdmatch(method, "mining.submit") && client->authorised)) {
 		json_params_t *jp = create_json_params(client_id, method_val, params_val, id_val, address);
 
-		ckwq_add(sdata->ckwqs, &sshare_process, jp);
+		ckmsgq_add(sdata->sshareq, jp);
 		return;
 	}
 
@@ -3340,7 +3338,7 @@ 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);
-		async_send_proc(ckp, client->ckp->connector, buf);
+		send_proc(client->ckp->connector, buf);
 		return;
 	}
 
@@ -3348,7 +3346,7 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
 	if (!client->subscribed) {
 		LOGINFO("Dropping unsubscribed client %"PRId64, client_id);
 		snprintf(buf, 255, "dropclient=%"PRId64, client_id);
-		async_send_proc(ckp, client->ckp->connector, buf);
+		send_proc(client->ckp->connector, buf);
 		return;
 	}
 
@@ -3371,7 +3369,7 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
 		 * the stratum instance data. Clients will just reconnect. */
 		LOGINFO("Dropping unauthorised client %"PRId64, client_id);
 		snprintf(buf, 255, "dropclient=%"PRId64, client_id);
-		async_send_proc(ckp, client->ckp->connector, buf);
+		send_proc(client->ckp->connector, buf);
 		return;
 	}
 
@@ -3384,7 +3382,7 @@ static void parse_method(sdata_t *sdata, stratum_instance_t *client, const int64
 	if (cmdmatch(method, "mining.get")) {
 		json_params_t *jp = create_json_params(client_id, method_val, params_val, id_val, address);
 
-		ckwq_add(sdata->ckwqs, &send_transactions, jp);
+		ckmsgq_add(sdata->stxnq, jp);
 		return;
 	}
 	/* Unhandled message here */
@@ -3403,13 +3401,12 @@ static void parse_instance_msg(sdata_t *sdata, smsg_t *msg, stratum_instance_t *
 {
 	json_t *val = msg->json_msg, *id_val, *method, *params;
 	int64_t client_id = msg->client_id;
-	ckpool_t *ckp = client->ckp;
 	char buf[256];
 
 	if (unlikely(client->reject == 2)) {
 		LOGINFO("Dropping client %"PRId64" tagged for lazy invalidation", client_id);
 		snprintf(buf, 255, "dropclient=%"PRId64, client_id);
-		async_send_proc(ckp, ckp->connector, buf);
+		send_proc(client->ckp->connector, buf);
 		goto out;
 	}
 
@@ -3543,7 +3540,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);
-	async_send_proc(ckp, ckp->connector, s);
+	send_proc(ckp->connector, s);
 	free(s);
 	free_smsg(msg);
 }
@@ -4401,10 +4398,14 @@ int stratifier(proc_instance_t *pi)
 
 	mutex_init(&sdata->ckdb_lock);
 	sdata->ssends = create_ckmsgq(ckp, "ssender", &ssend_process);
-	/* Create as many generic workqueue threads as there are CPUs */
-	threads = sysconf(_SC_NPROCESSORS_ONLN);
-	ckp->ckwqs = sdata->ckwqs = create_ckwqs(ckp, "strat", threads);
+	/* Create half as many share processing threads as there are CPUs */
+	threads = sysconf(_SC_NPROCESSORS_ONLN) / 2 ? : 1;
+	sdata->sshareq = create_ckmsgqs(ckp, "sprocessor", &sshare_process, threads);
+	/* Create 1/4 as many stratum processing threads as there are CPUs */
+	threads = threads / 2 ? : 1;
+	sdata->srecvs = create_ckmsgqs(ckp, "sreceiver", &srecv_process, threads);
 	sdata->sauthq = create_ckmsgq(ckp, "authoriser", &sauth_process);
+	sdata->stxnq = create_ckmsgq(ckp, "stxnq", &send_transactions);
 	if (!CKP_STANDALONE(ckp)) {
 		sdata->ckdbq = create_ckmsgq(ckp, "ckdbqueue", &ckdbq_process);
 		create_pthread(&pth_heartbeat, ckdb_heartbeat, ckp);