[PATCH 11/16] NTB: ntb_transport: Implement direct-DMA RX buffer publication

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

A direct-DMA sender needs the peer's RX buffer addresses, but reading a
peer-owned ring would add a non-posted PCIe round trip to the TX path.

Have the receiver map posted buffers and write each address and length
into the sender's local publication ring. Keep each mapping until its
completion is consumed, then unmap it before invoking the client
callback.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index ca08cb690311..152400feb6d6 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -206,6 +206,11 @@ struct ntb_transport_qp {
 	unsigned int direct_ring_entries;
 	/* Serialize direct session, TX ring, and RX publication state. */
 	spinlock_t direct_lock;
+	/*
+	 * rxc_db_work owns the direct RX queue and counters while active; teardown
+	 * accesses them only after cancel_work_sync().
+	 */
+	struct list_head direct_rx_q;
 	u32 *direct_rx_cpl;
 	dma_addr_t direct_rx_cpl_dma;
 	u32 *direct_tx_cpl;
@@ -224,6 +229,7 @@ struct ntb_transport_qp {
 	unsigned int rx_alloc_entry;
 	dma_cookie_t last_cookie;
 	struct work_struct rxc_db_work;
+	struct delayed_work direct_rx_retry;
 
 	void (*event_handler)(void *data, int status);
 	struct delayed_work link_work;
@@ -524,6 +530,7 @@ static bool ntb_direct_layout(struct ntb_transport_ctx *nt)
 #define NTB_QP_DEF_NUM_ENTRIES	100
 #define NTB_LINK_DOWN_TIMEOUT	10
 #define NTB_DIRECT_TEARDOWN_RETRY_INTERVAL_MS	10
+#define NTB_DIRECT_RX_RETRY_INTERVAL_MS	10
 
 static bool ntb_direct_rx_mode(struct ntb_transport_qp *qp)
 {
@@ -764,6 +771,8 @@ unsigned int ntb_transport_rx_queue_size(struct ntb_transport_qp *qp)
 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_direct_rx_reclaim(struct ntb_transport_qp *qp);
 static const struct ntb_ctx_ops ntb_transport_ops;
 static struct ntb_client ntb_transport_client;
 static int ntb_async_tx_submit(struct ntb_transport_qp *qp,
@@ -1447,6 +1456,15 @@ static void ntb_qp_link_context_reset(struct ntb_transport_qp *qp)
 
 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_work_sync(&qp->rxc_db_work);
+		/* Catch a retry armed while draining RX work. */
+		cancel_delayed_work_sync(&qp->direct_rx_retry);
+		ntb_direct_rx_reclaim(qp);
+	}
+
 	ntb_qp_link_context_reset(qp);
 	if (qp->remote_rx_info)
 		qp->remote_rx_info->entry = qp->rx_max_entry - 1;
@@ -1811,8 +1829,10 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 	INIT_LIST_HEAD(&qp->rx_free_q);
 	INIT_LIST_HEAD(&qp->tx_free_q);
 	INIT_LIST_HEAD(&qp->tx_offl_q);
+	INIT_LIST_HEAD(&qp->direct_rx_q);
 
 	INIT_WORK(&qp->rxc_db_work, ntb_transport_rxc_db);
+	INIT_DELAYED_WORK(&qp->direct_rx_retry, ntb_direct_rx_retry_work);
 
 	return 0;
 }
@@ -2219,6 +2239,195 @@ static void ntb_async_rx(struct ntb_queue_entry *entry, void *offset)
 	qp->rx_memcpy++;
 }
 
+static bool ntb_direct_rx_enabled(struct ntb_transport_qp *qp)
+{
+	return READ_ONCE(qp->direct_state) == NTB_DIRECT_ACTIVE &&
+	       ntb_direct_rx_mode(qp);
+}
+
+static bool ntb_direct_rx_can_complete(struct ntb_transport_qp *qp)
+{
+	enum ntb_direct_state state = READ_ONCE(qp->direct_state);
+
+	return (state == NTB_DIRECT_ACTIVE || state == NTB_DIRECT_QUIESCING) &&
+	       ntb_direct_rx_mode(qp);
+}
+
+static int ntb_direct_rx_publish(struct ntb_transport_qp *qp,
+				 struct ntb_queue_entry *entry)
+{
+	struct ntb_direct_pub __iomem *pub;
+	struct device *dma_dev;
+	dma_addr_t dma_addr;
+	u32 head, idx;
+	int rc = 0;
+
+	dma_dev = qp->transport->direct_dma_dev;
+	dma_addr = dma_map_single(dma_dev, entry->buf, entry->len,
+				  DMA_FROM_DEVICE);
+	if (dma_mapping_error(dma_dev, dma_addr))
+		return -EIO;
+
+	scoped_guard(spinlock_bh, &qp->direct_lock) {
+		if (!ntb_direct_rx_enabled(qp)) {
+			rc = -EOPNOTSUPP;
+			goto unmap;
+		}
+
+		head = qp->direct_rx_prod;
+		if (ntb_direct_ring_used(head, qp->direct_rx_cons) >=
+		    qp->direct_ring_entries - 1) {
+			rc = -ENOSPC;
+			goto unmap;
+		}
+
+		idx = ntb_direct_ring_idx(qp, head);
+		entry->direct_dma_addr = dma_addr;
+		list_add_tail(&entry->entry, &qp->direct_rx_q);
+
+		WRITE_ONCE(qp->direct_rx_cpl[idx], 0);
+		/* Publish the slot only after its completion target is clear. */
+		dma_wmb();
+		pub = &qp->peer_direct_shared->pub[idx];
+		iowrite32(lower_32_bits(dma_addr), &pub->addr_lo);
+		iowrite32(upper_32_bits(dma_addr), &pub->addr_hi);
+		iowrite32(entry->len, &pub->len);
+		iowrite32(head + 1, &qp->peer_direct_shared->pub_head);
+		WRITE_ONCE(qp->direct_rx_prod, head + 1);
+	}
+
+	return 0;
+
+unmap:
+	dma_unmap_single(dma_dev, dma_addr, entry->len, DMA_FROM_DEVICE);
+
+	return rc;
+}
+
+static void ntb_direct_rx_retry_work(struct work_struct *work)
+{
+	struct ntb_transport_qp *qp =
+		container_of(to_delayed_work(work), struct ntb_transport_qp,
+			     direct_rx_retry);
+
+	queue_work(system_dfl_wq, &qp->rxc_db_work);
+}
+
+static void ntb_direct_rx_replenish(struct ntb_transport_qp *qp)
+{
+	struct ntb_queue_entry *entry;
+	int rc;
+
+	while ((entry = ntb_list_rm(&qp->ntb_rx_q_lock, &qp->rx_pend_q))) {
+		rc = ntb_direct_rx_publish(qp, entry);
+		if (!rc)
+			continue;
+
+		ntb_list_add(&qp->ntb_rx_q_lock, &entry->entry,
+			     &qp->rx_pend_q);
+		if (rc == -EIO) {
+			dev_warn_ratelimited(&qp->ndev->pdev->dev,
+					     "qp %d: failed to map direct RX buffer; retrying\n",
+					     qp->qp_num);
+			/* Avoid hot-looping rxc_db_work on a persistent failure. */
+			mod_delayed_work(system_dfl_wq, &qp->direct_rx_retry,
+					 msecs_to_jiffies(NTB_DIRECT_RX_RETRY_INTERVAL_MS));
+		}
+		break;
+	}
+}
+
+static u32 ntb_direct_rx_completion_word(struct ntb_transport_qp *qp)
+{
+	u32 cons, prod, idx;
+
+	cons = READ_ONCE(qp->direct_rx_cons);
+	prod = READ_ONCE(qp->direct_rx_prod);
+	if (!ntb_direct_ring_used(prod, cons))
+		return 0;
+
+	idx = ntb_direct_ring_idx(qp, cons);
+	return READ_ONCE(qp->direct_rx_cpl[idx]);
+}
+
+static bool ntb_direct_rx_complete_one(struct ntb_transport_qp *qp)
+{
+	struct ntb_queue_entry *entry;
+	struct device *dma_dev;
+	void *cb_data;
+	bool notify;
+	int cb_len;
+	u32 word;
+
+	word = ntb_direct_rx_completion_word(qp);
+	if (!word)
+		return false;
+
+	entry = list_first_entry_or_null(&qp->direct_rx_q,
+					 struct ntb_queue_entry, entry);
+	if (WARN_ON_ONCE(!entry))
+		return false;
+
+	/* The completion DMA follows the payload on the same channel. */
+	dma_rmb();
+	list_del(&entry->entry);
+	WRITE_ONCE(qp->direct_rx_cons, READ_ONCE(qp->direct_rx_cons) + 1);
+
+	dma_dev = qp->transport->direct_dma_dev;
+	dma_unmap_single(dma_dev, entry->direct_dma_addr, entry->len,
+			 DMA_FROM_DEVICE);
+	cb_data = entry->cb_data;
+
+	if (word == NTB_DIRECT_CPL_ERROR) {
+		cb_len = -EIO;
+	} else if (word > INT_MAX || word > entry->len) {
+		qp->rx_err_oflow++;
+		cb_len = -EIO;
+	} else {
+		qp->rx_bytes += word;
+		qp->rx_pkts++;
+		cb_len = word;
+	}
+
+	notify = qp->client_ready && qp->rx_handler;
+	if (notify)
+		ntb_list_add(&qp->ntb_rx_q_lock, &entry->entry, &qp->rx_free_q);
+	else
+		ntb_list_add(&qp->ntb_rx_q_lock, &entry->entry, &qp->rx_pend_q);
+
+	if (notify)
+		qp->rx_handler(qp, qp->cb_data, cb_data, cb_len);
+
+	return true;
+}
+
+static void ntb_direct_rx_complete(struct ntb_transport_qp *qp)
+{
+	unsigned int count;
+
+	for (count = 0; count < qp->direct_ring_entries; count++)
+		if (!ntb_direct_rx_complete_one(qp))
+			break;
+}
+
+static void ntb_direct_rx_reclaim(struct ntb_transport_qp *qp)
+{
+	struct ntb_queue_entry *entry;
+	struct device *dma_dev;
+	LIST_HEAD(reclaim);
+
+	dma_dev = qp->transport->direct_dma_dev;
+	list_splice_init(&qp->direct_rx_q, &reclaim);
+
+	list_for_each_entry(entry, &reclaim, entry) {
+		dma_unmap_single(dma_dev, entry->direct_dma_addr, entry->len,
+				 DMA_FROM_DEVICE);
+	}
+
+	guard(spinlock_irqsave)(&qp->ntb_rx_q_lock);
+	list_splice_tail(&reclaim, &qp->rx_pend_q);
+}
+
 static int ntb_process_rxc(struct ntb_transport_qp *qp)
 {
 	struct ntb_payload_header *hdr;
@@ -2304,6 +2513,12 @@ static void ntb_transport_rxc_db(struct work_struct *work)
 		ntb_direct_control_progress(qp);
 	if (!qp->active)
 		goto clear_db;
+	if (ntb_direct_rx_can_complete(qp)) {
+		ntb_direct_rx_complete(qp);
+		if (ntb_direct_rx_enabled(qp))
+			ntb_direct_rx_replenish(qp);
+		goto clear_db;
+	}
 
 	/* Limit the number of packets processed in a single interrupt to
 	 * provide fairness to others
@@ -2886,11 +3101,15 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 	qp_bit = BIT_ULL(qp->qp_num);
 
 	ntb_db_set_mask(qp->ndev, qp_bit);
+	cancel_delayed_work_sync(&qp->direct_rx_retry);
 	cancel_work_sync(&qp->rxc_db_work);
+	/* Catch a retry armed while draining RX work. */
+	cancel_delayed_work_sync(&qp->direct_rx_retry);
 
 	/* Catch cleanup queued while draining RX processing. */
 	cancel_work_sync(&qp->link_cleanup);
 	cancel_delayed_work_sync(&qp->link_work);
+	ntb_direct_rx_reclaim(qp);
 	ntb_qp_link_context_reset(qp);
 
 	qp->cb_data = NULL;
-- 
2.51.0