Speedup SideChain::get_outputs_blob()

src/side_chain.cpp

@@ -682,7 +682,7 @@ bool SideChain::get_block_blob(const hash& id, std::vector<uint8_t>& blob) const
 	return true;
 }
 
-bool SideChain::get_outputs_blob(PoolBlock* block, uint64_t total_reward, std::vector<uint8_t>& blob) const
+bool SideChain::get_outputs_blob(PoolBlock* block, uint64_t total_reward, std::vector<uint8_t>& blob, uv_loop_t* loop) const
 {
 	blob.clear();
 
@@ -719,6 +719,60 @@ bool SideChain::get_outputs_blob(PoolBlock* block, uint64_t total_reward, std::v
 
 	const size_t n = tmpShares.size();
 
+	// Helper jobs call get_eph_public_key with indices in descending order
+	// Current thread will process indices in ascending order so when they meet, everything will be cached
+
+	std::atomic<int> counter{ 0 };
+	std::atomic<int> num_helper_jobs_finished{ 0 };
+	int num_helper_jobs_started = 0;
+
+	if (loop) {
+		constexpr size_t HELPER_JOBS_COUNT = 4;
+
+		struct Work
+		{
+			uv_work_t req;
+			const std::vector<MinerShare>& tmpShares;
+			const hash& txkeySec;
+			std::atomic<int>& counter;
+			std::atomic<int>& num_helper_jobs_finished;
+
+			// Fix MSVC warnings
+			Work() = delete;
+			Work& operator=(Work&&) = delete;
+		};
+
+		counter = static_cast<int>(n) - 1;
+		num_helper_jobs_started = HELPER_JOBS_COUNT;
+
+		for (size_t i = 0; i < HELPER_JOBS_COUNT; ++i) {
+			Work* w = new Work{ {}, tmpShares, block->m_txkeySec, counter, num_helper_jobs_finished };
+			w->req.data = w;
+
+			const int err = uv_queue_work(loop, &w->req,
+				[](uv_work_t* req)
+				{
+					Work* work = reinterpret_cast<Work*>(req->data);
+					hash eph_public_key;
+
+					int index;
+					while ((index = work->counter.fetch_sub(1)) >= 0) {
+						uint8_t view_tag;
+						work->tmpShares[index].m_wallet->get_eph_public_key(work->txkeySec, static_cast<size_t>(index), eph_public_key, view_tag);
+					}
+
+					++work->num_helper_jobs_finished;
+					delete work;
+				}, nullptr);
+
+			if (err) {
+				LOGERR(1, "get_outputs_blob: uv_queue_work failed, error " << uv_err_name(err));
+				--num_helper_jobs_started;
+				delete w;
+			}
+		}
+	}
+
 	blob.reserve(n * 39 + 64);
 
 	writeVarint(n, blob);
@@ -730,6 +784,13 @@ bool SideChain::get_outputs_blob(PoolBlock* block, uint64_t total_reward, std::v
 
 	hash eph_public_key;
 	for (size_t i = 0; i < n; ++i) {
+		// stop helper jobs when they meet with current thread
+		const int c = counter.load();
+		if ((c >= 0) && (static_cast<int>(i) >= c)) {
+			// this will cause all helper jobs to finish immediately
+			counter = -1;
+		}
+
 		writeVarint(tmpRewards[i], blob);
 
 		blob.emplace_back(tx_type);
@@ -747,6 +808,15 @@ bool SideChain::get_outputs_blob(PoolBlock* block, uint64_t total_reward, std::v
 		block->m_outputs.emplace_back(tmpRewards[i], eph_public_key, tx_type, view_tag);
 	}
 
+	if (loop) {
+		// this will cause all helper jobs to finish immediately
+		counter = -1;
+
+		while (num_helper_jobs_finished < num_helper_jobs_started) {
+			std::this_thread::yield();
+		}
+	}
+
 	return true;
 }