[PATCH 16/16] NTB: ntb_transport: Add optional polling for direct-DMA RX

[Koichiro Den <den@valinux.co.jp>]

Direct RX normally relies on peer MSI or doorbell notifications.
Polling the completion state once per jiffy lets RX buffers be recycled
sooner, keeps the DMA engine saturated, and raises the throughput
ceiling.

Add an opt-in direct_dma_poll parameter. Notification-driven operation
remains the default, so continuous polling is enabled only when
requested.

The poll continues even after all currently visible completions are
consumed. The posted MWr transactions carrying data and completion can
theoretically be overtaken by the CPU MMIO write used for peer
notification. The notification may go through another PCI function and
ordering domain. Without polling, the last completion could remain
pending if no later notification arrives. This has not been observed.

Reading back each completion destination would close this window, but
the non-posted PCIe round trip significantly lowers throughput.

Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
 drivers/ntb/ntb_transport.c | 35 ++++++++++++++++++++++++++++++++++-
 1 file changed, 34 insertions(+), 1 deletion(-)

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 2f69328d5201..d3eb4e2106b0 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -104,6 +104,11 @@ module_param(use_direct_dma, bool, 0644);
 MODULE_PARM_DESC(use_direct_dma,
 		 "Use PCI endpoint DMA to transfer directly to peer RX buffers");
 
+static bool direct_dma_poll;
+module_param(direct_dma_poll, bool, 0444);
+MODULE_PARM_DESC(direct_dma_poll,
+		 "Poll direct-DMA RX completion state once per jiffy");
+
 static unsigned int direct_dma_func;
 module_param(direct_dma_func, uint, 0644);
 MODULE_PARM_DESC(direct_dma_func,
@@ -234,6 +239,7 @@ struct ntb_transport_qp {
 	dma_cookie_t last_cookie;
 	struct work_struct rxc_db_work;
 	struct delayed_work direct_rx_retry;
+	struct delayed_work rxc_poll;
 
 	void (*event_handler)(void *data, int status);
 	struct delayed_work link_work;
@@ -835,6 +841,8 @@ EXPORT_SYMBOL_GPL(ntb_transport_rx_queue_size);
 
 static void ntb_transport_rxc_db(struct work_struct *work);
 static void ntb_direct_rx_retry_work(struct work_struct *work);
+static void ntb_transport_rxc_poll(struct work_struct *work);
+static bool ntb_direct_rx_can_complete(struct ntb_transport_qp *qp);
 static void ntb_direct_rx_reclaim(struct ntb_transport_qp *qp);
 static const struct ntb_ctx_ops ntb_transport_ops;
 static struct ntb_client ntb_transport_client;
@@ -1523,6 +1531,7 @@ static void ntb_qp_link_down_reset(struct ntb_transport_qp *qp)
 	if (ntb_direct_link_capable(qp)) {
 		qp->active = false;
 		cancel_delayed_work_sync(&qp->direct_rx_retry);
+		cancel_delayed_work_sync(&qp->rxc_poll);
 		cancel_work_sync(&qp->rxc_db_work);
 		/* Catch a retry armed while draining RX work. */
 		cancel_delayed_work_sync(&qp->direct_rx_retry);
@@ -1808,8 +1817,11 @@ static void ntb_qp_link_work(struct work_struct *work)
 		if (qp->event_handler)
 			qp->event_handler(qp->cb_data, qp->link_is_up);
 
-		if (qp->active)
+		if (qp->active) {
+			if (direct_dma_poll && ntb_direct_rx_can_complete(qp))
+				queue_delayed_work(system_dfl_wq, &qp->rxc_poll, 1);
 			queue_work(system_dfl_wq, &qp->rxc_db_work);
+		}
 	} else if (nt->link_is_up)
 		schedule_delayed_work(&qp->link_work,
 				      msecs_to_jiffies(NTB_LINK_DOWN_TIMEOUT));
@@ -1900,6 +1912,7 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 
 	INIT_WORK(&qp->rxc_db_work, ntb_transport_rxc_db);
 	INIT_DELAYED_WORK(&qp->direct_rx_retry, ntb_direct_rx_retry_work);
+	INIT_DELAYED_WORK(&qp->rxc_poll, ntb_transport_rxc_poll);
 
 	return 0;
 }
@@ -2629,6 +2642,25 @@ static void ntb_transport_rxc_db(struct work_struct *work)
 	}
 }
 
+static void ntb_transport_rxc_poll(struct work_struct *work)
+{
+	struct ntb_transport_qp *qp =
+		container_of(work, struct ntb_transport_qp, rxc_poll.work);
+	bool completion, control;
+
+	if (!qp->active || !ntb_direct_rx_can_complete(qp))
+		return;
+
+	control = ntb_direct_control_pending(qp);
+	completion = ntb_direct_rx_completion_word(qp);
+
+	if (completion || control)
+		queue_work(system_dfl_wq, &qp->rxc_db_work);
+
+	/* keep checking completion and control state */
+	queue_delayed_work(system_dfl_wq, &qp->rxc_poll, 1);
+}
+
 static void ntb_tx_copy_callback(void *data,
 				 const struct dmaengine_result *res)
 {
@@ -3517,6 +3549,7 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 
 	ntb_db_set_mask(qp->ndev, qp_bit);
 	cancel_delayed_work_sync(&qp->direct_rx_retry);
+	cancel_delayed_work_sync(&qp->rxc_poll);
 	cancel_work_sync(&qp->rxc_db_work);
 	/* Catch a retry armed while draining RX work. */
 	cancel_delayed_work_sync(&qp->direct_rx_retry);
-- 
2.51.0