[PATCH ncns v1 1/4] chnet: ring: Refactor SQE handling

[Ammar Faizi <ammarfaizi2@gnuweeb.org>]

Make everything about SQE handling simpler and rename them to represent
the function purpose better.

Signed-off-by: Ammar Faizi <ammarfaizi2@gnuweeb.org>
---
 chnet/chnet_ring.cc | 146 +++++++++++++++++++++++++-------------------
 chnet/chnet_ring.h  |  11 ++--
 2 files changed, 88 insertions(+), 69 deletions(-)

diff --git a/chnet/chnet_ring.cc b/chnet/chnet_ring.cc
index 13a7009..966e60e 100644
--- a/chnet/chnet_ring.cc
+++ b/chnet/chnet_ring.cc
@@ -113,7 +113,6 @@ bool CNRingCtx::TryPostCQE(CNRingSQE *sqe, int64_t res)
 
 	state_->cqe_lock_.lock();
 	cqe = GetCQENoTailIncrement();
-
 	if (unlikely(!cqe)) {
 		NotifyCQEWaiter();
 		state_->cqe_lock_.unlock();
@@ -129,45 +128,94 @@ bool CNRingCtx::TryPostCQE(CNRingSQE *sqe, int64_t res)
 	return true;
 }
 
-void CNRingCtx::PostCQENoFail(CNRingSQE *sqe, int64_t res)
+void CNRingCtx::PostCQE(CNRingSQE *sqe, int64_t res)
 {
 	while (1) {
 		if (TryPostCQE(sqe, res))
-			return;
+			break;
+
 		sleep(1);
 	}
 }
 
-void CNRingCtx::PostCQE(CNRingSQE *sqe, int64_t res)
+/*
+ * Use this when the caller is not allowed to block.
+ */
+void CNRingCtx::CallPostCQE(CNRingSQE *sqe, int64_t res)
 {
-	CNRingSQE *sqe_l;
+	CNRingSQE *tmp;
 
-	if (TryPostCQE(sqe, res))
+	if (likely(TryPostCQE(sqe, res)))
 		return;
 
-	sqe_l = new CNRingSQE;
-	*sqe_l = *sqe;
-	state_->wq_.schedule_work([=](struct wq_job_data *data){
-		CNRingSQE *sqe_tmp = (CNRingSQE *)data->data;
-		this->PostCQENoFail(sqe_tmp, res);
-		delete sqe_tmp;
-	}, sqe_l);
+	/*
+	 * The CQE slot is full, but the caller is not allowed to
+	 * block, let's schedule the post in the workqueue thread.
+	 */
+	tmp = new CNRingSQE;
+	*tmp = *sqe;
+	state_->wq_.schedule_work([this, res](struct wq_job_data *data){
+		CNRingSQE *sqe = (CNRingSQE *)data->data;
+
+		this->PostCQE(sqe, res);
+		delete sqe;
+	}, tmp);
+}
+
+void CNRingCtx::IssueNopSQE(CNRingSQE *sqe)
+{
+	CallPostCQE(sqe, 0);
+}
+
+static void issue_start_sqe(void *udata, net::URLRequest *url_req, int net_err)
+{
+	CNRingOpStart *sop;
+	CNRingSQE *sqe;
+
+	sqe = (CNRingSQE *)udata;
+	sop = (CNRingOpStart *)sqe->sq_data;
+	sop->ring->PostCQE(sqe, net_err);
+	delete sop;
+	delete sqe;
+}
+
+void CNRingCtx::IssueStartSQE(CNRingSQE *sqe)
+{
+	struct net::CHNCallback *cb;
+	CNRingOpStart *sop;
+	CNRingSQE *tmp;
+	int ret;
+
+	tmp = new CNRingSQE;
+	*tmp = *sqe;
+
+	sop = (CNRingOpStart *)tmp->sq_data;
+	sop->ring = this;
+
+	cb = &sop->chnet->ch()->cb;
+	cb->response_started_ = issue_start_sqe;
+	cb->response_started_data_ = tmp;
+	ret = sop->chnet->Start();
+	if (unlikely(ret)) {
+		CallPostCQE(sqe, ret);
+		delete tmp;
+	}
 }
 
-static void handle_read_sqe(void *udata, net::URLRequest *url_req, int bytes_read)
+static void issue_read_sqe(void *udata, net::URLRequest *url_req, int read_ret)
 {
 	CNRingOpStart *sop;
 	CNRingSQE *sqe;
 
 	sqe = (CNRingSQE *)udata;
 	sop = (CNRingOpStart *)sqe->sq_data;
-	sop->ring->PostCQENoFail(sqe, bytes_read);
+	sop->ring->PostCQE(sqe, read_ret);
 	delete sop;
 	delete sqe;
 }
 
-static void handle_read_sqe_need_start(void *udata, net::URLRequest *url_req,
-				       int net_err)
+static void issue_read_sqe_need_start(void *udata, net::URLRequest *url_req,
+				      int net_err)
 {
 	struct net::CHNCallback *cb;
 	CNRingOpRead *sop;
@@ -177,22 +225,27 @@ static void handle_read_sqe_need_start(void *udata, net::URLRequest *url_req,
 	sop = (CNRingOpRead *)sqe->sq_data;
 
 	if (likely(net_err == net::OK)) {
+		/*
+		 * The start operation succeeds, now we can do
+		 * the read operation directly without issuing
+		 * extra SQE.
+		 */
 		cb = &sop->chnet->ch()->cb;
-		cb->read_completed_ = handle_read_sqe;
+		cb->read_completed_ = issue_read_sqe;
 		cb->read_completed_data_ = sqe;
 		sop->chnet->ch()->_Read(nullptr, sop->read_size);
 		return;
 	}
 
 	/*
-	 * Operation fails!
+	 * The start operation fails, just post the CQE directly.
 	 */
-	sop->ring->PostCQENoFail(sqe, net_err);
+	sop->ring->PostCQE(sqe, net_err);
 	delete sop;
 	delete sqe;
 }
 
-void CNRingCtx::HandleReadSQE(CNRingSQE *sqe)
+void CNRingCtx::IssueReadSQE(CNRingSQE *sqe)
 {
 	struct net::CHNCallback *cb;
 	net::CHNetDelegate *ch;
@@ -214,14 +267,14 @@ void CNRingCtx::HandleReadSQE(CNRingSQE *sqe)
 		 * a read operation at this point, do the start
 		 * operation first!
 		 */
-		cb->response_started_ = handle_read_sqe_need_start;
+		cb->response_started_ = issue_read_sqe_need_start;
 		cb->response_started_data_ = tmp;
 		ret = sop->chnet->Start();
 	} else {
 		/*
 		 * Normal read operation here.
 		 */
-		cb->read_completed_ = handle_read_sqe;
+		cb->read_completed_ = issue_read_sqe;
 		cb->read_completed_data_ = tmp;
 		ret = sop->chnet->Read(sop->read_size);
 	}
@@ -232,56 +285,21 @@ void CNRingCtx::HandleReadSQE(CNRingSQE *sqe)
 	/*
 	 * Aiee, the operation fails!
 	 */
-	PostCQE(sqe, ret);
+	CallPostCQE(sqe, ret);
 	delete tmp;
 }
 
-static void handle_start_sqe(void *udata, net::URLRequest *url_req, int net_err)
-{
-	CNRingOpStart *sop;
-	CNRingSQE *sqe;
-
-	sqe = (CNRingSQE *)udata;
-	sop = (CNRingOpStart *)sqe->sq_data;
-	sop->ring->PostCQENoFail(sqe, net_err);
-	delete sop;
-	delete sqe;
-}
-
-void CNRingCtx::HandleStartSQE(CNRingSQE *sqe)
-{
-	struct net::CHNCallback *cb;
-	CNRingOpStart *sop;
-	CNRingSQE *tmp;
-	int ret;
-
-	tmp = new CNRingSQE;
-	*tmp = *sqe;
-
-	sop = (CNRingOpStart *)tmp->sq_data;
-	sop->ring = this;
-
-	cb = &sop->chnet->ch()->cb;
-	cb->response_started_ = handle_start_sqe;
-	cb->response_started_data_ = tmp;
-	ret = sop->chnet->Start();
-	if (unlikely(ret)) {
-		PostCQE(sqe, ret);
-		delete tmp;
-	}
-}
-
-void CNRingCtx::ProcessSQE(CNRingSQE *sqe)
+void CNRingCtx::IssueSQE(CNRingSQE *sqe)
 {
 	switch (sqe->op) {
 	case CNRING_OP_NOP:
-		PostCQE(sqe, 0);
+		IssueNopSQE(sqe);
 		break;
 	case CNRING_OP_START:
-		HandleStartSQE(sqe);
+		IssueStartSQE(sqe);
 		break;
 	case CNRING_OP_READ:
-		HandleReadSQE(sqe);
+		IssueReadSQE(sqe);
 		break;
 	}
 }
@@ -296,7 +314,7 @@ uint32_t CNRingCtx::SubmitSQE(uint32_t to_submit)
 		if (unlikely(head == tail))
 			break;
 
-		ProcessSQE(&sqes_[head++ & sq_mask_]);
+		IssueSQE(&sqes_[head++ & sq_mask_]);
 		ret++;
 	}
 
diff --git a/chnet/chnet_ring.h b/chnet/chnet_ring.h
index e4574df..b0cfc50 100644
--- a/chnet/chnet_ring.h
+++ b/chnet/chnet_ring.h
@@ -123,7 +123,7 @@ public:
 	uint32_t cqe_size(void);
 
 	void PostCQE(CNRingSQE *sqe, int64_t res);
-	void PostCQENoFail(CNRingSQE *sqe, int64_t res);
+	void CallPostCQE(CNRingSQE *sqe, int64_t res);
 
 	inline void CQAdvance(uint32_t n)
 	{
@@ -155,13 +155,14 @@ public:
 private:
 	void HandleSQE(CNRingSQE *sqe);
 	CNRingSQE *ConsumeSQE(void);
-	void ProcessSQEWQ(CNRingSQE *sqe);
-	void ProcessSQE(CNRingSQE *sqe);
 	bool TryPostCQE(CNRingSQE *sqe, int64_t res);
 	CNRingCQE *GetCQENoTailIncrement(void);
 	void NotifyCQEWaiter(void);
-	void HandleReadSQE(CNRingSQE *sqe);
-	void HandleStartSQE(CNRingSQE *sqe);
+
+	void IssueSQE(CNRingSQE *sqe);
+	void IssueNopSQE(CNRingSQE *sqe);
+	void IssueStartSQE(CNRingSQE *sqe);
+	void IssueReadSQE(CNRingSQE *sqe);
 };
 
 #define chnring_for_each_cqe(ring, head, cqe)				\
-- 
Ammar Faizi