[PATCH ncns v1 2/4] chnet: ring: Improve `PostCQE()` waiting mechanism

[Ammar Faizi <ammarfaizi2@gnuweeb.org>]

Don't use sleep(1) in a loop to wait for the CQE slot be free, instead,
use a conditional variable to reduce the loop cycle and also to reduce
the latency because the wake up doesn't need to wait for a full second.
It just waits until it's signaled whenever the CQE slot becomes
available again.

Signed-off-by: Ammar Faizi <ammarfaizi2@gnuweeb.org>
---
 chnet/chnet_ring.cc | 42 +++++++++++++++++++++++++++++++++++++++++-
 chnet/chnet_ring.h  | 12 +++++++-----
 2 files changed, 48 insertions(+), 6 deletions(-)

diff --git a/chnet/chnet_ring.cc b/chnet/chnet_ring.cc
index 966e60e..c9a3da5 100644
--- a/chnet/chnet_ring.cc
+++ b/chnet/chnet_ring.cc
@@ -34,6 +34,7 @@ CNRingCtx::CNRingCtx(uint32_t entry)
 	state_ = new CNRingState(sq_max, cq_max, 2);
 
 	state_->cqe_to_wait_.store(0, std::memory_order_relaxed);
+	state_->nr_post_cqe_wait_.store(0, std::memory_order_relaxed);
 	sq_head_.store(0, std::memory_order_relaxed);
 	sq_tail_.store(0, std::memory_order_relaxed);
 	cq_head_.store(0, std::memory_order_relaxed);
@@ -128,16 +129,54 @@ bool CNRingCtx::TryPostCQE(CNRingSQE *sqe, int64_t res)
 	return true;
 }
 
+void CNRingCtx::WaitForCQEFreeSlot(void)
+{
+	std::unique_lock<std::mutex> lock(state_->post_cqe_lock_);
+
+	state_->nr_post_cqe_wait_++;
+	state_->post_cqe_cond_.wait(lock, [this]{
+		bool ret;
+
+		state_->cqe_lock_.lock();
+		ret = !!GetCQENoTailIncrement();
+		state_->cqe_lock_.unlock();
+		return ret;
+	});
+	state_->nr_post_cqe_wait_--;
+}
+
 void CNRingCtx::PostCQE(CNRingSQE *sqe, int64_t res)
 {
 	while (1) {
 		if (TryPostCQE(sqe, res))
 			break;
 
-		sleep(1);
+		WaitForCQEFreeSlot();
+	}
+}
+
+void CNRingCtx::NotifyWaitCQEFreeSlot(void)
+{
+	uint32_t nr_wait = state_->nr_post_cqe_wait_.load();
+
+	if (unlikely(nr_wait)) {
+		state_->post_cqe_lock_.lock();
+		if (nr_wait == 1)
+			state_->post_cqe_cond_.notify_one();
+		else
+			state_->post_cqe_cond_.notify_all();
+		state_->post_cqe_lock_.unlock();
 	}
 }
 
+void CNRingCtx::CQAdvance(uint32_t n)
+{
+	state_->cqe_lock_.lock();
+	cq_head_ += n;
+	state_->cqe_lock_.unlock();
+	NotifyWaitCQEFreeSlot();
+}
+
 /*
  * Use this when the caller is not allowed to block.
  */
@@ -335,6 +374,7 @@ void CNRingCtx::WaitCQE(uint32_t to_wait)
 	if (to_wait > max_to_wait)
 		to_wait = max_to_wait;
 
+	NotifyWaitCQEFreeSlot();
 	if (to_wait <= cqe_size())
 		return;
 
diff --git a/chnet/chnet_ring.h b/chnet/chnet_ring.h
index b0cfc50..f41d75b 100644
--- a/chnet/chnet_ring.h
+++ b/chnet/chnet_ring.h
@@ -109,6 +109,10 @@ public:
 	std::mutex	cqe_lock_;
 	std::condition_variable cqe_cond_;
 	std::atomic<uint32_t> cqe_to_wait_;
+
+	std::mutex	post_cqe_lock_;
+	std::condition_variable post_cqe_cond_;
+	std::atomic<uint32_t> nr_post_cqe_wait_;
 };
 #endif
 
@@ -124,11 +128,7 @@ public:
 
 	void PostCQE(CNRingSQE *sqe, int64_t res);
 	void CallPostCQE(CNRingSQE *sqe, int64_t res);
-
-	inline void CQAdvance(uint32_t n)
-	{
-		cq_head_ += n;
-	}
+	void CQAdvance(uint32_t n);
 
 	inline CNRingCQE *HeadCQE(void)
 	{
@@ -158,6 +158,8 @@ private:
 	bool TryPostCQE(CNRingSQE *sqe, int64_t res);
 	CNRingCQE *GetCQENoTailIncrement(void);
 	void NotifyCQEWaiter(void);
+	void NotifyWaitCQEFreeSlot(void);
+	void WaitForCQEFreeSlot(void);
 
 	void IssueSQE(CNRingSQE *sqe);
 	void IssueNopSQE(CNRingSQE *sqe);
-- 
Ammar Faizi