Stratum: check found P2Pool shares one at a time

Parallel checks could result in a deadlock.

src/stratum_server.cpp

@@ -79,6 +79,8 @@ StratumServer::StratumServer(p2pool* pool)
 		m_submittedSharesPool[i] = share;
 	}
 
+	m_pendingShareChecks.reserve(10);
+
 	uv_async_init_checked(&m_loop, &m_blobsAsync, on_blobs_ready);
 	m_blobsAsync.data = this;
 	m_blobsQueue.reserve(2);
@@ -499,13 +501,25 @@ bool StratumServer::on_submit(StratumClient* client, uint32_t id, const char* jo
 		}
 
 		// Else switch to a worker thread to check PoW which can take a long time
-		const int err = uv_queue_work(&m_loop, &share->m_req, on_share_found, on_after_share_found);
-		if (err) {
-			LOGERR(1, "uv_queue_work failed, error " << uv_err_name(err));
+		if (m_pendingShareChecks.empty()) {
+			const int err = uv_queue_work(&m_loop, &share->m_req, on_share_found, on_after_share_found);
+			if (err) {
+				LOGERR(1, "uv_queue_work failed, error " << uv_err_name(err));
 
-			// If uv_queue_work failed, process this share here anyway
-			on_share_found(&share->m_req);
-			on_after_share_found(&share->m_req, 0);
+				// If uv_queue_work failed, process this share here anyway
+				on_share_found(&share->m_req);
+				on_after_share_found(&share->m_req, 0);
+			}
+			else {
+				m_pendingShareChecks.push_back(share);
+				LOGINFO(5, "Pending share checks count = " << m_pendingShareChecks.size());
+			}
+		}
+		else {
+			// If there is a check running already, it will pick up this share in on_after_share_found
+			// This ensures that only one share at a time is checked
+			m_pendingShareChecks.push_back(share);
+			LOGINFO(5, "Pending share checks count = " << m_pendingShareChecks.size());
 		}
 
 		return true;
@@ -1017,6 +1031,37 @@ void StratumServer::on_share_found(uv_work_t* req)
 void StratumServer::on_after_share_found(uv_work_t* req, int /*status*/)
 {
 	SubmittedShare* share = reinterpret_cast<SubmittedShare*>(req->data);
+	StratumServer* server = share->m_server;
+
+	server->check_event_loop_thread(__func__);
+
+	ON_SCOPE_LEAVE([share, server]()
+		{
+			auto it = std::find(server->m_pendingShareChecks.begin(), server->m_pendingShareChecks.end(), share);
+			if (it != server->m_pendingShareChecks.end()) {
+				server->m_pendingShareChecks.erase(it);
+			}
+			else {
+				LOGERR(1, "Background share check tracking is broken, fix the code!");
+			}
+			LOGINFO(5, "Pending share checks count = " << server->m_pendingShareChecks.size());
+
+			ASAN_POISON_MEMORY_REGION(share, sizeof(SubmittedShare));
+			server->m_submittedSharesPool.push_back(share);
+
+			if (!server->m_pendingShareChecks.empty()) {
+				SubmittedShare* share2 = server->m_pendingShareChecks.front();
+
+				const int err = uv_queue_work(&server->m_loop, &share2->m_req, on_share_found, on_after_share_found);
+				if (err) {
+					LOGERR(1, "uv_queue_work failed, error " << uv_err_name(err));
+
+					// If uv_queue_work failed, process this share here anyway
+					server->on_share_found(&share2->m_req);
+					server->on_after_share_found(&share2->m_req, 0);
+				}
+			}
+		});
 
 	if (share->m_highEnoughDifficulty) {
 		const char* s = share->m_clientCustomUser;
@@ -1040,14 +1085,6 @@ void StratumServer::on_after_share_found(uv_work_t* req, int /*status*/)
 		BACKGROUND_JOB_STOP(StratumServer::on_share_found);
 	}
 
-	StratumServer* server = share->m_server;
-
-	ON_SCOPE_LEAVE([share, server]()
-		{
-			ASAN_POISON_MEMORY_REGION(share, sizeof(SubmittedShare));
-			server->m_submittedSharesPool.push_back(share);
-		});
-
 	const bool bad_share = (share->m_result == SubmittedShare::Result::LOW_DIFF) || (share->m_result == SubmittedShare::Result::INVALID_POW);
 
 	StratumClient* client = share->m_client;

src/stratum_server.h

@@ -204,6 +204,8 @@ private:
 
 	std::atomic<uint64_t> m_apiLastUpdateTime;
 
+	std::vector<SubmittedShare*> m_pendingShareChecks;
+
 	void update_hashrate_data(uint64_t hashes, uint64_t timestamp);
 	void api_update_local_stats(uint64_t timestamp);