[PATCH 00/16] NTB: Add direct TX/RX using PCI endpoint DMA

[PATCH 00/16] NTB: Add direct TX/RX using PCI endpoint DMA

[Koichiro Den]

Hi,

This series lets ntb_transport transfer packets directly from client TX
buffers to peer RX buffers using PCI endpoint DMA, avoiding the
memory-window copy path. It raises throughput significantly.

My previous attempt at the same goal was:

  [RFC PATCH v4 00/38] NTB transport backed by PCI EP embedded DMA
  https://lore.kernel.org/r/20260118135440.1958279-1-den@valinux.co.jp/

Since that RFC, prerequisite work has entered mainline in a different
form. Unlike that RFC, this series extends ntb_transport itself and can
fall back to the existing memory-window copy path at runtime.

Some work needed to run this series on my R-Car S4 testbed is still under
review; see Dependencies below.

The series contains:

  * Patches 1-5   : small fixes and prerequisites
  * Patches 6-14  : the main implementation for direct TX/RX
  * Patches 15-16 : optional features


Summary
=======

ntb_transport currently copies each packet through a fixed slot in a
memory window. This series adds an optional path where the receiver
publishes client-buffer DMA addresses and the sender's DMA writes
packets directly to them. The existing copy path remains available, and
direct RX and TX are negotiated independently.

Bidirectional direct TX/RX requires a multi-function PCI endpoint. vNTB
uses one function, while the PCI DMA EPF on another exposes the endpoint
DMA controller to the host. Each sender needs an ordered channel that can
transfer from mapped system memory to peer-published DMA addresses. The
tested setup used dw-edma with the PCI DMA EPF on PF0 and vNTB on PF1.


Design
======

Session
-------

ntb_transport automatically reconnects a logical QP while the physical
NTB link and its shared MW remain up. One peer may therefore start the
next direct-DMA "session" while the other is still finishing the
previous teardown. A fresh session ID distinguishes the two instances,
and direct DMA remains disabled until both peers have acknowledged it.
Otherwise, stale RX addresses from the previous instance could be reused
after their mappings have been released.

Resetting the physical NTB link on every QP close would avoid this
overlap, but would also tear down all QPs for ordinary operations such
as ntb_netdev close/open, MTU changes, or queue reconfiguration. This
series keeps the existing automatic QP reconnect behavior instead.

Quiesce
-------

A published RX address must stay mapped until the sender can no longer
write to it. Closing the local QP alone does not prove this. During
teardown, each QP publishes its final issued TX boundary, then writes a
quiesce marker to tell the peer that this boundary is final and no more
TX will be issued for the session. The CPU-written marker can arrive
before earlier DMA writes when the two paths use different PCIe ordering
domains. The peer therefore waits until it has consumed completion words
through that boundary before acknowledging quiesce and releasing the RX
mappings. Local TX mappings remain owned until DMA completes or the
channel has been stopped and synchronized.

Completion ordering
-------------------

The payload and its completion word are submitted to the same ordered
DMA channel. A separate CPU MMIO completion would not provide that
ordering: its posted write could pass an earlier DMA write. The MSI or
doorbell sent after the DMA callback can still reach the peer before the
completion word is visible. A read-back from the completion destination
would close that window, but would add a non-posted PCIe round trip to
every packet. It hurt throughput in testing. Notifications are therefore
hints, while the completion word decides when an RX buffer can be
reused. Optional polling lets the receiver find a completion that
arrives after its notification.

Per-direction negotiation
-------------------------

A side may be able to map direct RX buffers even when it cannot reserve
all TX DMA channels. RX and TX are therefore negotiated separately,
allowing direct DMA in one direction while the other keeps using the
copy path.

Protocol compatibility
----------------------

The extension keeps protocol version 4 and the original QP layout for
peers that do not advertise direct DMA. Version 4 has been in use for
about 11 years.


Dependencies
============

The multi-function endpoint setup described above requires:

  1. [PATCH v6 0/6] PCI: endpoint: Expose endpoint DMA resources (part 2/3)
     (https://lore.kernel.org/r/20260804033855.2115817-1-den@valinux.co.jp/)

  2. [PATCH v5 0/3] PCI: endpoint: Add PCI DMA endpoint function (part 3/3)
     (https://lore.kernel.org/r/20260717050953.2145851-1-den@valinux.co.jp/)

  3. [PATCH 0/3] PCI: endpoint: Support vNTB as a non-first EPF
     (https://lore.kernel.org/r/20260728172306.2751813-1-den@valinux.co.jp/)

Series 1 and 2 will be combined in the next revision after the part 2
review settles, per Frank's feedback.


Performance improvement
=======================

The test used two R-Car S4 Spider boards connected with an OCuLink cable,
one as EP and the other as RC. The link was PCIe Gen4 x2 with DWC PCIe
controller IP v5.20 and eDMA (not HDMA). The workloads were:

- UDP: sudo iperf3 -ub0 -c $peer -l 65507 -w 512M -P 4 -t 10
- TCP: sudo iperf3 -Z -c $peer -l 65507 -P 8 -t 10

The results were:

  (unit: Gbps) (UL=EP->RC, DL=RC->EP)

          UL UDP  DL UDP   UL TCP  DL TCP
  ------- ------  -------  ------  ------
  Before  ~0.6     ~0.6     ~0.6     ~0.6
  After   ~19.5    ~17.3    ~12.3    ~10.8

Reaching the above throughput on R-Car S4 also requires:

- [PATCH 00/11] PCI/NTB: endpoint: packed vNTB memory windows
  (https://lore.kernel.org/r/20260803180439.1963864-1-den@valinux.co.jp/)
- [PATCH v4 00/24] dmaengine: dw-edma: Support dynamic LL appends
  (https://lore.kernel.org/r/20260729143036.3087722-1-den@valinux.co.jp/)

The test setup also used two unsubmitted platform changes: adding the PCIe
controller to the IPMMU allowlist and capping the EP DMA MRRS at 128 bytes.

The "After" result used:
- packed_mws=4 and `ethtool -L eth0 combined 4`
- `modprobe ntb_transport use_direct_dma=1 direct_dma_func=0 \
    direct_dma_ring_entries=256 direct_dma_poll=1`

Best regards,
Koichiro


Koichiro Den (16):
  NTB: ntb_transport: Abort link setup on QP MW allocation failure
  NTB: ntb_transport: Reject oversized TX buffers
  NTB: ntb_transport: Start TX offload thread after queue setup
  NTB: ntb_transport: Stop QP work before freeing a queue
  NTB: ntb_transport: Run RX processing on system workqueue
  NTB: ntb_transport: Define direct-DMA shared state
  NTB: ntb_transport: Negotiate direct-DMA queue layout
  NTB: ntb_transport: Add opt-in direct-DMA channel reservation
  NTB: ntb_transport: Allocate direct-DMA queue state
  NTB: ntb_transport: Implement direct-DMA QP session handshake
  NTB: ntb_transport: Implement direct-DMA RX buffer publication
  NTB: ntb_transport: Implement direct-DMA TX submission
  NTB: ntb_transport: Implement safe direct-DMA teardown
  NTB: ntb_transport: Enable direct-DMA queues
  NTB: ntb_transport: Report the direct-DMA payload limit
  NTB: ntb_transport: Add optional polling for direct-DMA RX

 drivers/net/ntb_netdev.c      |    4 +-
 drivers/ntb/ntb_transport.c   | 1523 +++++++++++++++++++++++++++++++--
 include/linux/ntb_transport.h |    1 +
 3 files changed, 1456 insertions(+), 72 deletions(-)

-- 
2.51.0

[PATCH 01/16] NTB: ntb_transport: Abort link setup on QP MW allocation failure

[Koichiro Den]

ntb_transport_setup_qp_mw() can fail while growing a QP's RX entry pool,
but the link worker ignores that error. The worker can then publish a QP
whose memory-window state is only partly initialized, and later work can
use stale or incomplete pointers.

Set up every QP memory window before publishing the transport link. On
failure, clear the QP pointers before releasing its MW backing and leave
the link down.

Fixes: a754a8fcaf38 ("NTB: allocate number transport entries depending on size of ring size")
Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
 drivers/ntb/ntb_transport.c | 20 ++++++++++++++++----
 1 file changed, 16 insertions(+), 4 deletions(-)

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index f59f926d4bfa..3efc50955253 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -1084,14 +1084,19 @@ static void ntb_transport_link_work(struct work_struct *work)
 			goto out1;
 	}
 
-	nt->link_is_up = true;
+	nt->link_is_up = false;
+	for (i = 0; i < nt->qp_count; i++) {
+		rc = ntb_transport_setup_qp_mw(nt, i);
+		if (rc)
+			goto out1;
+		ntb_transport_setup_qp_peer_msi(nt, i);
+	}
 
+	/* Publish the link only after every QP has been set up. */
+	nt->link_is_up = true;
 	for (i = 0; i < nt->qp_count; i++) {
 		struct ntb_transport_qp *qp = &nt->qp_vec[i];
 
-		ntb_transport_setup_qp_mw(nt, i);
-		ntb_transport_setup_qp_peer_msi(nt, i);
-
 		if (qp->client_ready)
 			schedule_delayed_work(&qp->link_work, 0);
 	}
@@ -1099,6 +1104,13 @@ static void ntb_transport_link_work(struct work_struct *work)
 	return;
 
 out1:
+	for (i = 0; i < nt->qp_count; i++) {
+		struct ntb_transport_qp *qp = &nt->qp_vec[i];
+
+		qp->rx_buff = NULL;
+		qp->remote_rx_info = NULL;
+	}
+
 	for (i = 0; i < nt->mw_count; i++)
 		ntb_free_mw(nt, i);
 
-- 
2.51.0

[PATCH 02/16] NTB: ntb_transport: Reject oversized TX buffers

[Koichiro Den]

ntb_process_tx() handles an oversized buffer by calling tx_handler()
with a NULL data pointer, returning the queue entry to tx_free_q, and
returning success. ntb_netdev therefore neither frees the skb from the
callback nor takes its enqueue error path, leaking it.

Return -EMSGSIZE instead. ntb_transport_tx_enqueue() puts the acquired
queue entry back on tx_free_q when ntb_process_tx() fails, while the
caller retains ownership of the rejected buffer.

Fixes: fce8a7bb5b4b ("PCI-Express Non-Transparent Bridge Support")
Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
 drivers/ntb/ntb_transport.c | 10 ++--------
 1 file changed, 2 insertions(+), 8 deletions(-)

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 3efc50955253..3a9635faad9e 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -1962,14 +1962,8 @@ static int ntb_process_tx(struct ntb_transport_qp *qp,
 		return -EAGAIN;
 	}
 
-	if (entry->len > qp->tx_max_frame - sizeof(struct ntb_payload_header)) {
-		if (qp->tx_handler)
-			qp->tx_handler(qp, qp->cb_data, NULL, -EIO);
-
-		ntb_list_add(&qp->ntb_tx_free_q_lock, &entry->entry,
-			     &qp->tx_free_q);
-		return 0;
-	}
+	if (entry->len > qp->tx_max_frame - sizeof(struct ntb_payload_header))
+		return -EMSGSIZE;
 
 	ntb_async_tx(qp, entry);
 
-- 
2.51.0

[PATCH 03/16] NTB: ntb_transport: Start TX offload thread after queue setup

[Koichiro Den]

ntb_transport_create_queue() starts the per-QP TX offload thread before
DMA mappings and queue entries are allocated. If later setup fails, the
error path returns the QP to the free bitmap without stopping the
thread. A retry can then reinitialize its waitqueue while the old thread
is still waiting on it.

Start the thread only after all fallible queue setup has completed.
Thread creation failure itself remains non-fatal and falls back to
inline copy as before.

Fixes: 322617a06c97 ("NTB: ntb_transport: Add 'tx_memcpy_offload' module option")
Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
 drivers/ntb/ntb_transport.c | 28 ++++++++++++++--------------
 1 file changed, 14 insertions(+), 14 deletions(-)

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 3a9635faad9e..044d8b5747fc 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -2063,20 +2063,6 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 	qp->tx_handler = handlers->tx_handler;
 	qp->event_handler = handlers->event_handler;
 
-	init_waitqueue_head(&qp->tx_offload_wq);
-	if (tx_memcpy_offload) {
-		qp->tx_offload_thread = kthread_run(ntb_tx_memcpy_kthread, qp,
-						    "ntb-txcpy/%s/%u",
-						    pci_name(ndev->pdev), qp->qp_num);
-		if (IS_ERR(qp->tx_offload_thread)) {
-			dev_warn(&nt->ndev->dev,
-				 "tx memcpy offload thread creation failed: %ld; falling back to inline copy\n",
-				 PTR_ERR(qp->tx_offload_thread));
-			qp->tx_offload_thread = NULL;
-		}
-	} else
-		qp->tx_offload_thread = NULL;
-
 	dma_cap_zero(dma_mask);
 	dma_cap_set(DMA_MEMCPY, dma_mask);
 
@@ -2137,6 +2123,20 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 			     &qp->tx_free_q);
 	}
 
+	init_waitqueue_head(&qp->tx_offload_wq);
+	qp->tx_offload_thread = NULL;
+	if (tx_memcpy_offload) {
+		qp->tx_offload_thread = kthread_run(ntb_tx_memcpy_kthread, qp,
+						    "ntb-txcpy/%s/%u",
+						    pci_name(ndev->pdev), qp->qp_num);
+		if (IS_ERR(qp->tx_offload_thread)) {
+			dev_warn(&nt->ndev->dev,
+				 "tx memcpy offload thread creation failed: %ld; falling back to inline copy\n",
+				 PTR_ERR(qp->tx_offload_thread));
+			qp->tx_offload_thread = NULL;
+		}
+	}
+
 	ntb_db_clear(qp->ndev, qp_bit);
 	ntb_db_clear_mask(qp->ndev, qp_bit);
 
-- 
2.51.0

[PATCH 04/16] NTB: ntb_transport: Stop QP work before freeing a queue

[Koichiro Den]

ntb_transport_link_down() clears client_ready, but asynchronous QP work
can outlive that transition. This leaves three teardown races:

  1. Queued link_work can mark the QP active again and schedule RX
     processing.
  2. A per-QP MSI can schedule RX processing after client link-down.
  3. link_cleanup can run while ntb_transport_free_queue() releases queue
     state and can rearm link_work. RX processing can also queue another
     cleanup while it is being drained.

Gate link_work and the MSI handler with client_ready. Before releasing
queue resources, drain link_cleanup and link_work. After stopping RX
processing, drain link_cleanup once more, then link_work because cleanup
can rearm it.

Fixes: fce8a7bb5b4b ("PCI-Express Non-Transparent Bridge Support")
Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
 drivers/ntb/ntb_transport.c | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 044d8b5747fc..4afad4489772 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -653,7 +653,8 @@ static irqreturn_t ntb_transport_isr(int irq, void *dev)
 {
 	struct ntb_transport_qp *qp = dev;
 
-	tasklet_schedule(&qp->rxc_db_work);
+	if (qp->client_ready)
+		tasklet_schedule(&qp->rxc_db_work);
 
 	return IRQ_HANDLED;
 }
@@ -1133,6 +1134,9 @@ static void ntb_qp_link_work(struct work_struct *work)
 	struct ntb_transport_ctx *nt = qp->transport;
 	int val;
 
+	if (!qp->client_ready)
+		return;
+
 	WARN_ON(!nt->link_is_up);
 
 	val = ntb_spad_read(nt->ndev, QP_LINKS);
@@ -2182,6 +2186,8 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 
 	pdev = qp->ndev->pdev;
 
+	cancel_work_sync(&qp->link_cleanup);
+	cancel_delayed_work_sync(&qp->link_work);
 	qp->active = false;
 
 	if (qp->tx_offload_thread) {
@@ -2229,6 +2235,8 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 	ntb_db_set_mask(qp->ndev, qp_bit);
 	tasklet_kill(&qp->rxc_db_work);
 
+	/* Catch cleanup queued while draining RX processing. */
+	cancel_work_sync(&qp->link_cleanup);
 	cancel_delayed_work_sync(&qp->link_work);
 
 	qp->cb_data = NULL;
-- 
2.51.0

[PATCH 05/16] NTB: ntb_transport: Run RX processing on system workqueue

[Koichiro Den]

Doorbell interrupts run every QP's RX tasklet on the IRQ CPU. With
several queues, completion and replenishment work piles up there and DMA
channels can run dry.

Run per-QP RX processing on system_dfl_wq. The hard IRQ only queues
work, and independent QPs can run on different CPUs.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 4afad4489772..463ee6f17fcb 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -179,7 +179,7 @@ struct ntb_transport_qp {
 	unsigned int rx_max_frame;
 	unsigned int rx_alloc_entry;
 	dma_cookie_t last_cookie;
-	struct tasklet_struct rxc_db_work;
+	struct work_struct rxc_db_work;
 
 	void (*event_handler)(void *data, int status);
 	struct delayed_work link_work;
@@ -289,7 +289,7 @@ enum {
 #define NTB_QP_DEF_NUM_ENTRIES	100
 #define NTB_LINK_DOWN_TIMEOUT	10
 
-static void ntb_transport_rxc_db(unsigned long data);
+static void ntb_transport_rxc_db(struct work_struct *work);
 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,
@@ -654,7 +654,7 @@ static irqreturn_t ntb_transport_isr(int irq, void *dev)
 	struct ntb_transport_qp *qp = dev;
 
 	if (qp->client_ready)
-		tasklet_schedule(&qp->rxc_db_work);
+		queue_work(system_dfl_wq, &qp->rxc_db_work);
 
 	return IRQ_HANDLED;
 }
@@ -1156,7 +1156,7 @@ static void ntb_qp_link_work(struct work_struct *work)
 			qp->event_handler(qp->cb_data, qp->link_is_up);
 
 		if (qp->active)
-			tasklet_schedule(&qp->rxc_db_work);
+			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));
@@ -1243,8 +1243,7 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 	INIT_LIST_HEAD(&qp->tx_free_q);
 	INIT_LIST_HEAD(&qp->tx_offl_q);
 
-	tasklet_init(&qp->rxc_db_work, ntb_transport_rxc_db,
-		     (unsigned long)qp);
+	INIT_WORK(&qp->rxc_db_work, ntb_transport_rxc_db);
 
 	return 0;
 }
@@ -1693,9 +1692,10 @@ static int ntb_process_rxc(struct ntb_transport_qp *qp)
 	return 0;
 }
 
-static void ntb_transport_rxc_db(unsigned long data)
+static void ntb_transport_rxc_db(struct work_struct *work)
 {
-	struct ntb_transport_qp *qp = (void *)data;
+	struct ntb_transport_qp *qp =
+		container_of(work, struct ntb_transport_qp, rxc_db_work);
 	int rc, i;
 
 	dev_dbg(&qp->ndev->pdev->dev, "%s: doorbell %d received\n",
@@ -1716,7 +1716,7 @@ static void ntb_transport_rxc_db(unsigned long data)
 	if (i == qp->rx_max_entry) {
 		/* there is more work to do */
 		if (qp->active)
-			tasklet_schedule(&qp->rxc_db_work);
+			queue_work(system_dfl_wq, &qp->rxc_db_work);
 	} else if (ntb_db_read(qp->ndev) & BIT_ULL(qp->qp_num)) {
 		/* the doorbell bit is set: clear it */
 		ntb_db_clear(qp->ndev, BIT_ULL(qp->qp_num));
@@ -1728,7 +1728,7 @@ static void ntb_transport_rxc_db(unsigned long data)
 		 * there might be some more work to do.
 		 */
 		if (qp->active)
-			tasklet_schedule(&qp->rxc_db_work);
+			queue_work(system_dfl_wq, &qp->rxc_db_work);
 	}
 }
 
@@ -2233,7 +2233,7 @@ 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);
-	tasklet_kill(&qp->rxc_db_work);
+	cancel_work_sync(&qp->rxc_db_work);
 
 	/* Catch cleanup queued while draining RX processing. */
 	cancel_work_sync(&qp->link_cleanup);
@@ -2334,7 +2334,7 @@ int ntb_transport_rx_enqueue(struct ntb_transport_qp *qp, void *cb, void *data,
 	ntb_list_add(&qp->ntb_rx_q_lock, &entry->entry, &qp->rx_pend_q);
 
 	if (qp->active)
-		tasklet_schedule(&qp->rxc_db_work);
+		queue_work(system_dfl_wq, &qp->rxc_db_work);
 
 	return 0;
 }
@@ -2528,7 +2528,7 @@ static void ntb_transport_doorbell_callback(void *data, int vector)
 		qp = &nt->qp_vec[qp_num];
 
 		if (qp->active)
-			tasklet_schedule(&qp->rxc_db_work);
+			queue_work(system_dfl_wq, &qp->rxc_db_work);
 
 		db_bits &= ~BIT_ULL(qp_num);
 	}
-- 
2.51.0

[PATCH 06/16] NTB: ntb_transport: Define direct-DMA shared state

[Koichiro Den]

Define the per-QP shared layout for session state, RX buffer
publications, completion reporting, and teardown boundaries.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 463ee6f17fcb..44957c11d1c8 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -55,6 +55,7 @@
 #include <linux/export.h>
 #include <linux/interrupt.h>
 #include <linux/kthread.h>
+#include <linux/limits.h>
 #include <linux/module.h>
 #include <linux/pci.h>
 #include <linux/slab.h>
@@ -264,6 +265,61 @@ enum {
 	LINK_DOWN_FLAG = BIT(1),
 };
 
+/* RX publications are pushed into the sender's local copy. */
+struct ntb_direct_pub {
+	u32 addr_lo;
+	u32 addr_hi;
+	u32 len;
+};
+
+/* During a session, the peer writes this area and the local CPU reads it. */
+struct ntb_direct_shared {
+	/* Peer session ID and the peer's ack of the local session. */
+	u32 session;
+	u32 session_ack;
+
+	/*
+	 * Tearing down a direct-DMA session must keep RX mappings valid until
+	 * completions for all in-flight transfers have been consumed. Each peer
+	 * publishes its final issued TX boundary, followed by a session-tagged
+	 * quiesce marker, using CPU MMIO writes to the peer's memory window.
+	 * The marker may arrive before earlier DMA writes because the CPU and
+	 * DMA paths may use different PCIe ordering domains. Quiesce is
+	 * therefore acknowledged only after completions through the published
+	 * boundary have been consumed:
+	 * - quiesce_issue: exclusive boundary of the peer's issued TX work that
+	 *                  local RX must consume.
+	 * - quiesce: local session ID written by the peer after quiesce_issue,
+	 *            validating that boundary for the current session.
+	 * - quiesce_ack: local session ID confirming that the peer consumed
+	 *                local TX through its final boundary.
+	 */
+	u32 quiesce_issue;
+	u32 quiesce;
+	u32 quiesce_ack;
+
+	/* Base address of the peer's RX completion array. */
+	u32 cpl_addr_lo;
+	u32 cpl_addr_hi;
+
+	/* Exclusive producer boundary for the peer RX publications below. */
+	u32 pub_head;
+
+	/* Peer RX buffer publications in ring order. */
+	struct ntb_direct_pub pub[];
+};
+
+/* Bound per-QP completion arrays and software queue allocations. */
+#define NTB_DIRECT_MAX_RING_ENTRIES	4096
+
+/* Zero is pending, U32_MAX is an error, and other values are lengths. */
+#define NTB_DIRECT_CPL_ERROR		U32_MAX
+
+static inline size_t ntb_direct_shared_size(unsigned int entries)
+{
+	return struct_size_t(struct ntb_direct_shared, pub, entries);
+}
+
 struct ntb_payload_header {
 	unsigned int ver;
 	unsigned int len;
-- 
2.51.0

[PATCH 07/16] NTB: ntb_transport: Negotiate direct-DMA queue layout

[Koichiro Den]

The direct-DMA shared area occupies part of each QP memory-window slice,
so both peers must agree on its size and placement. Exchange direct-DMA
capabilities and ring depth in spare scratchpads, and reserve the area
only when both peers advertise compatible settings. Otherwise retain the
protocol v4 copy layout.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 44957c11d1c8..f873eacd532b 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -98,6 +98,11 @@ static bool use_dma;
 module_param(use_dma, bool, 0644);
 MODULE_PARM_DESC(use_dma, "Use DMA engine to perform large data copy");
 
+static unsigned int direct_dma_ring_entries = 64;
+module_param(direct_dma_ring_entries, uint, 0644);
+MODULE_PARM_DESC(direct_dma_ring_entries,
+		 "Number of entries in each direct-DMA queue");
+
 static bool use_msi;
 #ifdef CONFIG_NTB_MSI
 module_param(use_msi, bool, 0644);
@@ -174,6 +179,15 @@ struct ntb_transport_qp {
 	struct list_head rx_free_q;
 	/* ntb_rx_q_lock: synchronize access to rx_XXXX_q */
 	spinlock_t ntb_rx_q_lock;
+	struct ntb_direct_shared *direct_shared;
+	struct ntb_direct_shared __iomem *peer_direct_shared;
+	unsigned int direct_ring_entries;
+	u32 *direct_tx_cpl;
+	dma_addr_t direct_tx_cpl_dma;
+	u32 direct_rx_prod;
+	u32 direct_rx_cons;
+	u32 direct_tx_issue;
+	u32 direct_tx_cons;
 	void *rx_buff;
 	unsigned int rx_index;
 	unsigned int rx_max_entry;
@@ -255,6 +269,9 @@ struct ntb_transport_ctx {
 	struct work_struct link_cleanup;
 
 	struct dentry *debugfs_node_dir;
+	u32 direct_features;
+	u32 peer_direct_features;
+	unsigned int direct_ring_entries;
 
 	/* Make sure workq of link event be executed serially */
 	struct mutex link_event_lock;
@@ -315,11 +332,26 @@ struct ntb_direct_shared {
 /* Zero is pending, U32_MAX is an error, and other values are lengths. */
 #define NTB_DIRECT_CPL_ERROR		U32_MAX
 
+enum {
+	NTB_DIRECT_FEAT_RX = BIT(0),
+	NTB_DIRECT_FEAT_TX = BIT(1),
+};
+
 static inline size_t ntb_direct_shared_size(unsigned int entries)
 {
 	return struct_size_t(struct ntb_direct_shared, pub, entries);
 }
 
+static inline u32 ntb_direct_ring_idx(struct ntb_transport_qp *qp, u32 val)
+{
+	return val & (qp->direct_ring_entries - 1);
+}
+
+static inline u32 ntb_direct_ring_used(u32 head, u32 tail)
+{
+	return head - tail;
+}
+
 struct ntb_payload_header {
 	unsigned int ver;
 	unsigned int len;
@@ -335,6 +367,32 @@ enum {
 	MW0_SZ_LOW,
 };
 
+enum {
+	DIRECT_SPAD_FEATURES,
+	DIRECT_SPAD_RING_ENTRIES,
+	DIRECT_SPAD_COUNT,
+};
+
+static unsigned int ntb_direct_spad_offset(struct ntb_transport_ctx *nt)
+{
+	/* Skip two SPADs per QP reserved for optional v4 MSI descriptors. */
+	return nt->msi_spad_offset + nt->qp_count * 2;
+}
+
+static bool ntb_direct_spads_available(struct ntb_transport_ctx *nt)
+{
+	return ntb_direct_spad_offset(nt) + DIRECT_SPAD_COUNT <=
+		ntb_spad_count(nt->ndev);
+}
+
+static bool ntb_direct_layout(struct ntb_transport_ctx *nt)
+{
+	return ((nt->direct_features & NTB_DIRECT_FEAT_RX) &&
+		(nt->peer_direct_features & NTB_DIRECT_FEAT_TX)) ||
+	       ((nt->direct_features & NTB_DIRECT_FEAT_TX) &&
+		(nt->peer_direct_features & NTB_DIRECT_FEAT_RX));
+}
+
 #define dev_client_dev(__dev) \
 	container_of((__dev), struct ntb_transport_client_dev, dev)
 
@@ -635,6 +693,38 @@ static struct ntb_queue_entry *ntb_list_mv(spinlock_t *lock,
 	return entry;
 }
 
+/*
+ * Protocol v4 uses the whole area before rx_info for copy slots. Reserve the
+ * tail for direct state only when both peers negotiated a direct direction.
+ */
+static int ntb_transport_setup_qp_tx_layout(struct ntb_transport_qp *qp)
+{
+	unsigned int tx_size = qp->tx_mw_size;
+	size_t direct_size = 0;
+
+	if (ntb_direct_layout(qp->transport))
+		direct_size = ntb_direct_shared_size(qp->direct_ring_entries);
+	if (direct_size &&
+	    tx_size < direct_size + sizeof(struct ntb_rx_info) +
+		      2 * sizeof(struct ntb_payload_header))
+		return -ENOSPC;
+
+	qp->rx_info = qp->tx_mw + tx_size - sizeof(struct ntb_rx_info);
+	tx_size -= sizeof(struct ntb_rx_info);
+	if (direct_size) {
+		tx_size -= direct_size;
+		qp->peer_direct_shared = qp->tx_mw + tx_size;
+	} else {
+		qp->peer_direct_shared = NULL;
+	}
+
+	/* Ring housekeeping requires at least two buffers. */
+	qp->tx_max_frame = min(transport_mtu, tx_size / 2);
+	qp->tx_max_entry = tx_size / qp->tx_max_frame;
+
+	return 0;
+}
+
 static int ntb_transport_setup_qp_mw(struct ntb_transport_ctx *nt,
 				     unsigned int qp_num)
 {
@@ -644,6 +734,8 @@ static int ntb_transport_setup_qp_mw(struct ntb_transport_ctx *nt,
 	struct ntb_queue_entry *entry;
 	unsigned int rx_size, num_qps_mw;
 	unsigned int mw_num, mw_count, qp_count;
+	size_t direct_size = 0;
+	void *rx_base;
 	unsigned int i;
 	int node;
 
@@ -662,10 +754,24 @@ static int ntb_transport_setup_qp_mw(struct ntb_transport_ctx *nt,
 		num_qps_mw = qp_count / mw_count;
 
 	rx_size = (unsigned int)mw->xlat_size / num_qps_mw;
-	qp->rx_buff = mw->virt_addr + rx_size * (qp_num / mw_count);
+	rx_base = mw->virt_addr + rx_size * (qp_num / mw_count);
+	if (ntb_direct_layout(nt))
+		direct_size = ntb_direct_shared_size(qp->direct_ring_entries);
+	if (direct_size &&
+	    rx_size < direct_size + sizeof(struct ntb_rx_info) +
+		      2 * sizeof(struct ntb_payload_header))
+		return -ENOSPC;
+
+	qp->rx_buff = rx_base;
 	rx_size -= sizeof(struct ntb_rx_info);
-
 	qp->remote_rx_info = qp->rx_buff + rx_size;
+	if (direct_size) {
+		rx_size -= direct_size;
+		qp->direct_shared = qp->rx_buff + rx_size;
+		memset(qp->direct_shared, 0, direct_size);
+	} else {
+		qp->direct_shared = NULL;
+	}
 
 	/* Due to housekeeping, there must be atleast 2 buffs */
 	qp->rx_max_frame = min(transport_mtu, rx_size / 2);
@@ -1070,6 +1176,7 @@ static void ntb_transport_link_work(struct work_struct *work)
 	struct ntb_dev *ndev = nt->ndev;
 	struct pci_dev *pdev = ndev->pdev;
 	resource_size_t size;
+	unsigned int direct_spad;
 	u32 val;
 	int rc = 0, i, spad;
 
@@ -1106,6 +1213,15 @@ static void ntb_transport_link_work(struct work_struct *work)
 	ntb_peer_spad_write(ndev, PIDX, NUM_MWS, nt->mw_count);
 
 	ntb_peer_spad_write(ndev, PIDX, NUM_QPS, nt->qp_count);
+	if (ntb_direct_spads_available(nt)) {
+		direct_spad = ntb_direct_spad_offset(nt);
+		ntb_peer_spad_write(ndev, PIDX,
+				    direct_spad + DIRECT_SPAD_FEATURES,
+				    nt->direct_features);
+		ntb_peer_spad_write(ndev, PIDX,
+				    direct_spad + DIRECT_SPAD_RING_ENTRIES,
+				    nt->direct_ring_entries);
+	}
 
 	ntb_peer_spad_write(ndev, PIDX, VERSION, NTB_TRANSPORT_VERSION);
 
@@ -1125,6 +1241,26 @@ static void ntb_transport_link_work(struct work_struct *work)
 	if (val != nt->mw_count)
 		goto out;
 
+	nt->peer_direct_features = 0;
+	if (ntb_direct_spads_available(nt)) {
+		direct_spad = ntb_direct_spad_offset(nt);
+		val = ntb_spad_read(ndev,
+				    direct_spad + DIRECT_SPAD_FEATURES);
+		val &= NTB_DIRECT_FEAT_RX | NTB_DIRECT_FEAT_TX;
+		if (val) {
+			u32 entries;
+
+			entries = ntb_spad_read(ndev,
+						direct_spad +
+						DIRECT_SPAD_RING_ENTRIES);
+			dev_dbg(&pdev->dev,
+				"Remote direct DMA features = %#x, ring entries = %u\n",
+				val, entries);
+			if (entries == nt->direct_ring_entries)
+				nt->peer_direct_features = val;
+		}
+	}
+
 	for (i = 0; i < nt->mw_count; i++) {
 		u64 val64;
 
@@ -1143,6 +1279,9 @@ static void ntb_transport_link_work(struct work_struct *work)
 
 	nt->link_is_up = false;
 	for (i = 0; i < nt->qp_count; i++) {
+		rc = ntb_transport_setup_qp_tx_layout(&nt->qp_vec[i]);
+		if (rc)
+			goto out1;
 		rc = ntb_transport_setup_qp_mw(nt, i);
 		if (rc)
 			goto out1;
@@ -1164,6 +1303,8 @@ static void ntb_transport_link_work(struct work_struct *work)
 	for (i = 0; i < nt->qp_count; i++) {
 		struct ntb_transport_qp *qp = &nt->qp_vec[i];
 
+		qp->direct_shared = NULL;
+		qp->peer_direct_shared = NULL;
 		qp->rx_buff = NULL;
 		qp->remote_rx_info = NULL;
 	}
@@ -1254,22 +1395,20 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 
 	tx_size = (unsigned int)mw_size / num_qps_mw;
 	qp_offset = tx_size * (qp_num / mw_count);
+	mw_base += qp_offset;
 
+	qp->direct_ring_entries = nt->direct_ring_entries;
 	qp->tx_mw_size = tx_size;
 	qp->tx_mw = nt->mw_vec[mw_num].vbase + qp_offset;
 	if (!qp->tx_mw)
 		return -EINVAL;
 
-	qp->tx_mw_phys = mw_base + qp_offset;
+	qp->tx_mw_phys = mw_base;
 	if (!qp->tx_mw_phys)
 		return -EINVAL;
 
-	tx_size -= sizeof(struct ntb_rx_info);
-	qp->rx_info = qp->tx_mw + tx_size;
-
-	/* Due to housekeeping, there must be atleast 2 buffs */
-	qp->tx_max_frame = min(transport_mtu, tx_size / 2);
-	qp->tx_max_entry = tx_size / qp->tx_max_frame;
+	if (ntb_transport_setup_qp_tx_layout(qp))
+		return -ENOSPC;
 
 	if (nt->debugfs_node_dir) {
 		char debugfs_name[8];
@@ -1337,6 +1476,7 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
 		return -ENOMEM;
 
 	nt->ndev = ndev;
+	nt->direct_ring_entries = direct_dma_ring_entries;
 
 	/*
 	 * If we are using MSI, and have at least one extra memory window,
@@ -1412,6 +1552,18 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
 	nt->qp_bitmap = qp_bitmap;
 	nt->qp_bitmap_free = qp_bitmap;
 
+	if (ntb_direct_spads_available(nt)) {
+		unsigned int spad = ntb_direct_spad_offset(nt);
+
+		/* Old v4 peers do not overwrite stale extension SPADs. */
+		rc = ntb_spad_write(ndev, spad + DIRECT_SPAD_FEATURES, 0);
+		if (rc)
+			goto err1;
+		rc = ntb_spad_write(ndev, spad + DIRECT_SPAD_RING_ENTRIES, 0);
+		if (rc)
+			goto err1;
+	}
+
 	nt->qp_vec = kcalloc_node(qp_count, sizeof(*nt->qp_vec),
 				  GFP_KERNEL, node);
 	if (!nt->qp_vec) {
-- 
2.51.0

[PATCH 08/16] NTB: ntb_transport: Add opt-in direct-DMA channel reservation

[Koichiro Den]

Direct DMA is opt-in. Add 'use_direct_dma' module parameter to request
it. Some endpoint integrations expose NTB and DMA through sibling PCI
functions, so add direct_dma_func to select the DMA function when the
NTB device cannot return the DMA device directly.

When use_direct_dma is set and negotiation scratchpads are available,
locate the DMA device and reserve one suitable channel for every QP
before registering clients. This avoids negotiating direct TX before a
later-created QP discovers that no channel is available. If the full set
is unavailable, leave direct TX disabled. Note that direct RX does not
require these channels.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index f873eacd532b..6051d8abfc0d 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -98,6 +98,16 @@ static bool use_dma;
 module_param(use_dma, bool, 0644);
 MODULE_PARM_DESC(use_dma, "Use DMA engine to perform large data copy");
 
+static bool use_direct_dma;
+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 unsigned int direct_dma_func;
+module_param(direct_dma_func, uint, 0644);
+MODULE_PARM_DESC(direct_dma_func,
+		 "PCI function number of a sibling endpoint DMA function");
+
 static unsigned int direct_dma_ring_entries = 64;
 module_param(direct_dma_ring_entries, uint, 0644);
 MODULE_PARM_DESC(direct_dma_ring_entries,
@@ -147,6 +157,7 @@ struct ntb_transport_qp {
 	void *cb_data;
 	struct dma_chan *tx_dma_chan;
 	struct dma_chan *rx_dma_chan;
+	struct dma_chan *direct_dma_chan;
 
 	bool client_ready;
 	bool link_is_up;
@@ -269,6 +280,7 @@ struct ntb_transport_ctx {
 	struct work_struct link_cleanup;
 
 	struct dentry *debugfs_node_dir;
+	struct device *direct_dma_dev;
 	u32 direct_features;
 	u32 peer_direct_features;
 	unsigned int direct_ring_entries;
@@ -352,6 +364,97 @@ static inline u32 ntb_direct_ring_used(u32 head, u32 tail)
 	return head - tail;
 }
 
+static bool ntb_direct_dma_filter_fn(struct dma_chan *chan, void *data)
+{
+	struct dma_slave_caps caps;
+
+	if (chan->device->dev != data || dma_get_slave_caps(chan, &caps))
+		return false;
+
+	/* Payload and completion descriptors must complete in order. */
+	return caps.cmd_terminate &&
+	       !dma_has_cap(DMA_COMPLETION_NO_ORDER, chan->device->cap_mask) &&
+	       (caps.directions & BIT(DMA_MEM_TO_DEV));
+}
+
+static struct device *ntb_direct_get_dma_dev(struct ntb_dev *ndev)
+{
+	struct pci_dev *pdev = ndev->pdev;
+	struct pci_dev *dma_pdev;
+	struct device *dev;
+
+	if (ndev->ops->get_dma_dev) {
+		dev = ntb_get_dma_dev(ndev);
+		return dev ? get_device(dev) : ERR_PTR(-ENODEV);
+	}
+
+	if (!pdev || !pdev->bus || direct_dma_func > 7)
+		return ERR_PTR(-ENODEV);
+
+	dma_pdev = pci_get_domain_bus_and_slot(pci_domain_nr(pdev->bus),
+					       pdev->bus->number,
+					       PCI_DEVFN(PCI_SLOT(pdev->devfn),
+							 direct_dma_func));
+	if (!dma_pdev)
+		return ERR_PTR(-ENODEV);
+
+	return &dma_pdev->dev;
+}
+
+static void ntb_direct_dma_release_channels(struct ntb_transport_ctx *nt)
+{
+	unsigned int i;
+
+	for (i = 0; i < nt->qp_count; i++) {
+		if (!nt->qp_vec[i].direct_dma_chan)
+			continue;
+
+		dma_release_channel(nt->qp_vec[i].direct_dma_chan);
+		nt->qp_vec[i].direct_dma_chan = NULL;
+	}
+}
+
+static void ntb_direct_dma_release(struct ntb_transport_ctx *nt)
+{
+	ntb_direct_dma_release_channels(nt);
+	if (nt->direct_dma_dev) {
+		put_device(nt->direct_dma_dev);
+		nt->direct_dma_dev = NULL;
+	}
+}
+
+static void ntb_direct_dma_init(struct ntb_transport_ctx *nt)
+{
+	dma_cap_mask_t mask;
+	unsigned int i;
+
+	if (!use_direct_dma)
+		return;
+
+	nt->direct_dma_dev = ntb_direct_get_dma_dev(nt->ndev);
+	if (IS_ERR(nt->direct_dma_dev)) {
+		dev_info(&nt->ndev->dev, "direct DMA device unavailable: %pe\n",
+			 nt->direct_dma_dev);
+		nt->direct_dma_dev = NULL;
+		return;
+	}
+
+	dma_cap_zero(mask);
+	dma_cap_set(DMA_SLAVE, mask);
+
+	for (i = 0; i < nt->qp_count; i++) {
+		nt->qp_vec[i].direct_dma_chan =
+			dma_request_channel(mask, ntb_direct_dma_filter_fn,
+					    nt->direct_dma_dev);
+		if (!nt->qp_vec[i].direct_dma_chan) {
+			dev_info(&nt->ndev->dev,
+				 "not enough direct DMA channels for all QPs\n");
+			ntb_direct_dma_release_channels(nt);
+			return;
+		}
+	}
+}
+
 struct ntb_payload_header {
 	unsigned int ver;
 	unsigned int len;
@@ -1476,6 +1579,16 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
 		return -ENOMEM;
 
 	nt->ndev = ndev;
+	if (use_direct_dma &&
+	    (direct_dma_ring_entries < 2 ||
+	     direct_dma_ring_entries > NTB_DIRECT_MAX_RING_ENTRIES ||
+	     !is_power_of_2(direct_dma_ring_entries))) {
+		dev_err(&ndev->dev,
+			"direct DMA ring entries must be a power of two between 2 and %u\n",
+			NTB_DIRECT_MAX_RING_ENTRIES);
+		rc = -EINVAL;
+		goto err;
+	}
 	nt->direct_ring_entries = direct_dma_ring_entries;
 
 	/*
@@ -1582,6 +1695,11 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
 		if (rc)
 			goto err2;
 	}
+	if (ntb_direct_spads_available(nt))
+		ntb_direct_dma_init(nt);
+	else if (use_direct_dma)
+		dev_info(&ndev->dev,
+			 "not enough scratchpads for direct DMA negotiation\n");
 
 	mutex_init(&nt->link_event_lock);
 	INIT_DELAYED_WORK(&nt->link_work, ntb_transport_link_work);
@@ -1605,6 +1723,7 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
 err3:
 	ntb_clear_ctx(ndev);
 err2:
+	ntb_direct_dma_release(nt);
 	kfree(nt->qp_vec);
 err1:
 	while (i--) {
@@ -1648,6 +1767,7 @@ static void ntb_transport_free(struct ntb_client *self, struct ntb_dev *ndev)
 		iounmap(nt->mw_vec[i].vbase);
 	}
 
+	ntb_direct_dma_release(nt);
 	kfree(nt->qp_vec);
 	kfree(nt->mw_vec);
 	kfree(nt);
-- 
2.51.0

[PATCH 09/16] NTB: ntb_transport: Allocate direct-DMA queue state

[Koichiro Den]

A direct-DMA ring can be deeper than the existing queue pools, and each
slot needs DMA-visible completion storage.

Allocate the completion arrays with the QP and size its software entries
for the larger ring. Expose the RX queue depth so clients can post
enough buffers.

Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
 drivers/net/ntb_netdev.c      |  4 +-
 drivers/ntb/ntb_transport.c   | 78 +++++++++++++++++++++++++++++++++--
 include/linux/ntb_transport.h |  1 +
 3 files changed, 78 insertions(+), 5 deletions(-)

diff --git a/drivers/net/ntb_netdev.c b/drivers/net/ntb_netdev.c
index c3a6ba96fc8a..62c5a5c1dfed 100644
--- a/drivers/net/ntb_netdev.c
+++ b/drivers/net/ntb_netdev.c
@@ -47,7 +47,6 @@ struct ntb_netdev {
 };
 
 #define	NTB_TX_TIMEOUT_MS	1000
-#define	NTB_RXQ_SIZE		100
 
 static void ntb_netdev_update_carrier(struct ntb_netdev *dev)
 {
@@ -82,10 +81,11 @@ static void ntb_netdev_queue_rx_drain(struct ntb_netdev_queue *queue)
 static int ntb_netdev_queue_rx_fill(struct net_device *ndev,
 				    struct ntb_netdev_queue *queue)
 {
+	unsigned int rxq_size = ntb_transport_rx_queue_size(queue->qp);
 	struct sk_buff *skb;
 	int rc, i;
 
-	for (i = 0; i < NTB_RXQ_SIZE; i++) {
+	for (i = 0; i < rxq_size; i++) {
 		skb = netdev_alloc_skb(ndev, ndev->mtu + ETH_HLEN);
 		if (!skb)
 			return -ENOMEM;
diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 6051d8abfc0d..67044d0ea0ff 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -139,6 +139,8 @@ struct ntb_queue_entry {
 	int errors;
 	unsigned int tx_index;
 	unsigned int rx_index;
+	dma_addr_t direct_dma_addr;
+	dma_addr_t direct_cpl_addr;
 
 	struct ntb_transport_qp *qp;
 	union {
@@ -193,6 +195,8 @@ struct ntb_transport_qp {
 	struct ntb_direct_shared *direct_shared;
 	struct ntb_direct_shared __iomem *peer_direct_shared;
 	unsigned int direct_ring_entries;
+	u32 *direct_rx_cpl;
+	dma_addr_t direct_rx_cpl_dma;
 	u32 *direct_tx_cpl;
 	dma_addr_t direct_tx_cpl_dma;
 	u32 direct_rx_prod;
@@ -506,6 +510,23 @@ static bool ntb_direct_layout(struct ntb_transport_ctx *nt)
 #define NTB_QP_DEF_NUM_ENTRIES	100
 #define NTB_LINK_DOWN_TIMEOUT	10
 
+/**
+ * ntb_transport_rx_queue_size - Query the RX queue depth
+ * @qp: NTB transport layer queue to be queried
+ *
+ * Return: Number of RX buffers the client should allocate
+ */
+unsigned int ntb_transport_rx_queue_size(struct ntb_transport_qp *qp)
+{
+	if (!qp)
+		return 0;
+
+	return qp->transport->direct_dma_dev ?
+		max(NTB_QP_DEF_NUM_ENTRIES, qp->direct_ring_entries - 1) :
+		NTB_QP_DEF_NUM_ENTRIES;
+}
+EXPORT_SYMBOL_GPL(ntb_transport_rx_queue_size);
+
 static void ntb_transport_rxc_db(struct work_struct *work);
 static const struct ntb_ctx_ops ntb_transport_ops;
 static struct ntb_client ntb_transport_client;
@@ -1179,6 +1200,10 @@ static void ntb_qp_link_context_reset(struct ntb_transport_qp *qp)
 	qp->tx_err_no_buf = 0;
 	qp->tx_memcpy = 0;
 	qp->tx_async = 0;
+	qp->direct_rx_prod = 0;
+	qp->direct_rx_cons = 0;
+	qp->direct_tx_issue = 0;
+	qp->direct_tx_cons = 0;
 }
 
 static void ntb_qp_link_down_reset(struct ntb_transport_qp *qp)
@@ -2343,6 +2368,23 @@ static bool ntb_dma_filter_fn(struct dma_chan *chan, void *node)
 	return dev_to_node(&chan->dev->device) == (int)(unsigned long)node;
 }
 
+static void ntb_direct_free_queue(struct ntb_transport_qp *qp)
+{
+	size_t cpl_size = array_size(qp->direct_ring_entries, sizeof(u32));
+
+	if (qp->direct_tx_cpl) {
+		dma_free_coherent(dmaengine_get_dma_device(qp->direct_dma_chan),
+				  cpl_size,
+				  qp->direct_tx_cpl, qp->direct_tx_cpl_dma);
+		qp->direct_tx_cpl = NULL;
+	}
+	if (qp->direct_rx_cpl) {
+		dma_free_coherent(qp->transport->direct_dma_dev, cpl_size,
+				  qp->direct_rx_cpl, qp->direct_rx_cpl_dma);
+		qp->direct_rx_cpl = NULL;
+	}
+}
+
 /**
  * ntb_transport_create_queue - Create a new NTB transport layer queue
  * @data: pointer for callback data
@@ -2368,7 +2410,9 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 	struct ntb_transport_qp *qp;
 	u64 qp_bit;
 	unsigned int free_queue;
+	unsigned int tx_entries;
 	dma_cap_mask_t dma_mask;
+	size_t direct_cpl_size;
 	int node;
 	int i;
 
@@ -2415,6 +2459,28 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 		qp->rx_dma_chan = NULL;
 	}
 
+	if (nt->direct_dma_dev) {
+		direct_cpl_size = array_size(qp->direct_ring_entries,
+					     sizeof(*qp->direct_rx_cpl));
+		qp->direct_rx_cpl =
+			dma_alloc_coherent(nt->direct_dma_dev, direct_cpl_size,
+					   &qp->direct_rx_cpl_dma, GFP_KERNEL);
+		if (!qp->direct_rx_cpl)
+			goto err1;
+	}
+
+	if (qp->direct_dma_chan) {
+		direct_cpl_size = array_size(qp->direct_ring_entries,
+					     sizeof(*qp->direct_tx_cpl));
+
+		qp->direct_tx_cpl =
+			dma_alloc_coherent(dmaengine_get_dma_device(qp->direct_dma_chan),
+					   direct_cpl_size,
+					   &qp->direct_tx_cpl_dma, GFP_KERNEL);
+		if (!qp->direct_tx_cpl)
+			goto err1;
+	}
+
 	qp->tx_mw_dma_addr = 0;
 	if (qp->tx_dma_chan) {
 		qp->tx_mw_dma_addr =
@@ -2434,7 +2500,7 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 	dev_dbg(&pdev->dev, "Using %s memcpy for RX\n",
 		qp->rx_dma_chan ? "DMA" : "CPU");
 
-	for (i = 0; i < NTB_QP_DEF_NUM_ENTRIES; i++) {
+	for (i = 0; i < ntb_transport_rx_queue_size(qp); i++) {
 		entry = kzalloc_node(sizeof(*entry), GFP_KERNEL, node);
 		if (!entry)
 			goto err1;
@@ -2443,9 +2509,12 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 		ntb_list_add(&qp->ntb_rx_q_lock, &entry->entry,
 			     &qp->rx_free_q);
 	}
-	qp->rx_alloc_entry = NTB_QP_DEF_NUM_ENTRIES;
+	qp->rx_alloc_entry = i;
 
-	for (i = 0; i < qp->tx_max_entry; i++) {
+	tx_entries = qp->tx_max_entry;
+	if (qp->direct_dma_chan)
+		tx_entries = max(tx_entries, qp->direct_ring_entries - 1);
+	for (i = 0; i < tx_entries; i++) {
 		entry = kzalloc_node(sizeof(*entry), GFP_KERNEL, node);
 		if (!entry)
 			goto err2;
@@ -2480,6 +2549,7 @@ ntb_transport_create_queue(void *data, struct device *client_dev,
 	while ((entry = ntb_list_rm(&qp->ntb_tx_free_q_lock, &qp->tx_free_q)))
 		kfree(entry);
 err1:
+	ntb_direct_free_queue(qp);
 	qp->rx_alloc_entry = 0;
 	while ((entry = ntb_list_rm(&qp->ntb_rx_q_lock, &qp->rx_free_q)))
 		kfree(entry);
@@ -2591,6 +2661,8 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 	while ((entry = ntb_list_rm(&qp->ntb_tx_offl_q_lock, &qp->tx_offl_q)))
 		kfree(entry);
 
+	ntb_direct_free_queue(qp);
+
 	qp->transport->qp_bitmap_free |= qp_bit;
 
 	dev_info(&pdev->dev, "NTB Transport QP %d freed\n", qp->qp_num);
diff --git a/include/linux/ntb_transport.h b/include/linux/ntb_transport.h
index 7243eb98a722..41ba7c51cf75 100644
--- a/include/linux/ntb_transport.h
+++ b/include/linux/ntb_transport.h
@@ -71,6 +71,7 @@ struct ntb_queue_handlers {
 
 unsigned char ntb_transport_qp_num(struct ntb_transport_qp *qp);
 unsigned int ntb_transport_max_size(struct ntb_transport_qp *qp);
+unsigned int ntb_transport_rx_queue_size(struct ntb_transport_qp *qp);
 struct ntb_transport_qp *
 ntb_transport_create_queue(void *data, struct device *client_dev,
 			   const struct ntb_queue_handlers *handlers);
-- 
2.51.0

[PATCH 10/16] NTB: ntb_transport: Implement direct-DMA QP session handshake

[Koichiro Den]

A logical QP can be reused while its shared memory still contains RX
addresses and completions from the previous instance. Without a session
ID, the new QP could accept that stale state.

Give each QP instance a fresh session ID and enable direct DMA only
after both peers acknowledge it. Teardown also exchanges the final
issued boundary so RX mappings remain valid until all outstanding
transfers have completed.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 67044d0ea0ff..ca08cb690311 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -58,6 +58,7 @@
 #include <linux/limits.h>
 #include <linux/module.h>
 #include <linux/pci.h>
+#include <linux/random.h>
 #include <linux/slab.h>
 #include <linux/seq_file.h>
 #include <linux/types.h>
@@ -153,6 +154,14 @@ struct ntb_rx_info {
 	unsigned int entry;
 };
 
+enum ntb_direct_state {
+	NTB_DIRECT_DOWN,
+	NTB_DIRECT_HANDSHAKE,
+	NTB_DIRECT_ACTIVE,
+	NTB_DIRECT_QUIESCING,
+	NTB_DIRECT_QUIESCED,
+};
+
 struct ntb_transport_qp {
 	struct ntb_transport_ctx *transport;
 	struct ntb_dev *ndev;
@@ -195,6 +204,8 @@ struct ntb_transport_qp {
 	struct ntb_direct_shared *direct_shared;
 	struct ntb_direct_shared __iomem *peer_direct_shared;
 	unsigned int direct_ring_entries;
+	/* Serialize direct session, TX ring, and RX publication state. */
+	spinlock_t direct_lock;
 	u32 *direct_rx_cpl;
 	dma_addr_t direct_rx_cpl_dma;
 	u32 *direct_tx_cpl;
@@ -203,6 +214,9 @@ struct ntb_transport_qp {
 	u32 direct_rx_cons;
 	u32 direct_tx_issue;
 	u32 direct_tx_cons;
+	u32 direct_session;
+	u32 direct_peer_session;
+	enum ntb_direct_state direct_state;
 	void *rx_buff;
 	unsigned int rx_index;
 	unsigned int rx_max_entry;
@@ -509,6 +523,228 @@ static bool ntb_direct_layout(struct ntb_transport_ctx *nt)
 #define QP_TO_MW(nt, qp)	((qp) % nt->mw_count)
 #define NTB_QP_DEF_NUM_ENTRIES	100
 #define NTB_LINK_DOWN_TIMEOUT	10
+#define NTB_DIRECT_TEARDOWN_RETRY_INTERVAL_MS	10
+
+static bool ntb_direct_rx_mode(struct ntb_transport_qp *qp)
+{
+	struct ntb_transport_ctx *nt = qp->transport;
+
+	return (nt->direct_features & NTB_DIRECT_FEAT_RX) &&
+	       (nt->peer_direct_features & NTB_DIRECT_FEAT_TX);
+}
+
+static bool ntb_direct_tx_mode(struct ntb_transport_qp *qp)
+{
+	struct ntb_transport_ctx *nt = qp->transport;
+
+	return (nt->direct_features & NTB_DIRECT_FEAT_TX) &&
+	       (nt->peer_direct_features & NTB_DIRECT_FEAT_RX);
+}
+
+static bool ntb_direct_link_capable(struct ntb_transport_qp *qp)
+{
+	return ntb_direct_rx_mode(qp) || ntb_direct_tx_mode(qp);
+}
+
+static void ntb_transport_notify_peer(struct ntb_transport_qp *qp)
+{
+	if (qp->use_msi)
+		ntb_msi_peer_trigger(qp->ndev, PIDX, &qp->peer_msi_desc);
+	else
+		ntb_peer_db_set(qp->ndev, BIT_ULL(qp->qp_num));
+}
+
+static bool ntb_direct_tx_idle(struct ntb_transport_qp *qp)
+{
+	return qp->direct_tx_issue == qp->direct_tx_cons;
+}
+
+static bool ntb_direct_rx_drained(struct ntb_transport_qp *qp)
+{
+	struct ntb_direct_shared *shared = qp->direct_shared;
+	u32 session = qp->direct_session;
+
+	if (!shared || !session || READ_ONCE(shared->quiesce) != session)
+		return false;
+
+	/* quiesce is written after its final issue boundary. */
+	dma_rmb();
+	return READ_ONCE(qp->direct_rx_cons) ==
+	       READ_ONCE(shared->quiesce_issue);
+}
+
+static bool ntb_direct_tx_acked(struct ntb_transport_qp *qp)
+{
+	struct ntb_direct_shared *shared = qp->direct_shared;
+	u32 session = qp->direct_session;
+
+	return shared && session &&
+	       READ_ONCE(shared->quiesce_ack) == session;
+}
+
+static bool ntb_direct_control_pending(struct ntb_transport_qp *qp)
+{
+	struct ntb_direct_shared *shared = qp->direct_shared;
+	enum ntb_direct_state state = READ_ONCE(qp->direct_state);
+	u32 peer_session = READ_ONCE(qp->direct_peer_session);
+	u32 session = READ_ONCE(qp->direct_session);
+
+	if (!shared)
+		return false;
+	if (state == NTB_DIRECT_HANDSHAKE)
+		return READ_ONCE(shared->session) != peer_session ||
+		       READ_ONCE(shared->session_ack) == session;
+	if (state != NTB_DIRECT_ACTIVE)
+		return false;
+
+	return READ_ONCE(shared->session) != peer_session ||
+	       READ_ONCE(shared->quiesce) == session;
+}
+
+static void ntb_direct_control_publish_locked(struct ntb_transport_qp *qp)
+{
+	struct ntb_direct_shared __iomem *peer = qp->peer_direct_shared;
+	u32 peer_session = qp->direct_peer_session;
+
+	lockdep_assert_held(&qp->direct_lock);
+
+	/* Publish the completion array address before its session. */
+	iowrite32(lower_32_bits(qp->direct_rx_cpl_dma), &peer->cpl_addr_lo);
+	iowrite32(upper_32_bits(qp->direct_rx_cpl_dma), &peer->cpl_addr_hi);
+
+	iowrite32(qp->direct_session, &peer->session);
+	if (!peer_session)
+		return;
+
+	iowrite32(peer_session, &peer->session_ack);
+	if (qp->direct_state == NTB_DIRECT_QUIESCING) {
+		/* Publish the final exclusive TX boundary before its marker. */
+		iowrite32(qp->direct_tx_issue, &peer->quiesce_issue);
+		iowrite32(peer_session, &peer->quiesce);
+	}
+	if (ntb_direct_rx_drained(qp))
+		iowrite32(peer_session, &peer->quiesce_ack);
+}
+
+/*
+ * Accept the peer session during HANDSHAKE. In ACTIVE or QUIESCING, a
+ * replacement peer session or quiesce request enters the local teardown path.
+ */
+static bool ntb_direct_control_progress(struct ntb_transport_qp *qp)
+{
+	struct ntb_direct_shared *shared = qp->direct_shared;
+	u32 peer_session, session_ack, quiesce;
+	bool published = false;
+	bool cleanup = false;
+	bool ready = false;
+
+	if (!qp->transport->link_is_up || !shared ||
+	    !qp->peer_direct_shared ||
+	    !ntb_direct_link_capable(qp) ||
+	    ntb_link_is_up(qp->ndev, NULL, NULL) != 1)
+		return true;
+
+	peer_session = READ_ONCE(shared->session);
+	session_ack = READ_ONCE(shared->session_ack);
+	quiesce = READ_ONCE(shared->quiesce);
+
+	scoped_guard(spinlock_bh, &qp->direct_lock) {
+		if (!qp->direct_session)
+			goto out;
+
+		if (qp->direct_state == NTB_DIRECT_HANDSHAKE && peer_session) {
+			qp->direct_peer_session = peer_session;
+		} else if (qp->direct_state == NTB_DIRECT_ACTIVE && peer_session &&
+			   peer_session != qp->direct_peer_session) {
+			qp->direct_state = NTB_DIRECT_QUIESCING;
+		}
+
+		if (quiesce == qp->direct_session &&
+		    (qp->direct_state == NTB_DIRECT_HANDSHAKE ||
+		     qp->direct_state == NTB_DIRECT_ACTIVE)) {
+			qp->direct_state = NTB_DIRECT_QUIESCING;
+			cleanup = qp->client_ready;
+		}
+
+		ntb_direct_control_publish_locked(qp);
+		published = true;
+		if (qp->direct_state == NTB_DIRECT_HANDSHAKE &&
+		    qp->direct_peer_session &&
+		    session_ack == qp->direct_session)
+			qp->direct_state = NTB_DIRECT_ACTIVE;
+		ready = qp->direct_state == NTB_DIRECT_ACTIVE;
+	}
+
+out:
+	if (published)
+		ntb_transport_notify_peer(qp);
+	if (cleanup)
+		schedule_work(&qp->link_cleanup);
+
+	return ready;
+}
+
+static void ntb_direct_session_start(struct ntb_transport_qp *qp)
+{
+	u32 session;
+
+	if (!ntb_direct_link_capable(qp) || !qp->direct_shared)
+		return;
+
+	session = get_random_u32_above(0);
+
+	memset(qp->direct_shared, 0, sizeof(*qp->direct_shared));
+	/* Complete local shared-state reset before publishing the new session. */
+	dma_wmb();
+
+	guard(spinlock_bh)(&qp->direct_lock);
+	qp->direct_session = session;
+	qp->direct_peer_session = 0;
+	qp->direct_state = NTB_DIRECT_HANDSHAKE;
+}
+
+static void ntb_direct_quiesce(struct ntb_transport_qp *qp)
+{
+	bool done;
+
+	if (!ntb_direct_link_capable(qp))
+		return;
+
+	scoped_guard(spinlock_bh, &qp->direct_lock) {
+		if (qp->direct_state == NTB_DIRECT_DOWN ||
+		    qp->direct_state == NTB_DIRECT_QUIESCED)
+			return;
+		if (!qp->direct_peer_session) {
+			/*
+			 * No peer session means RX publication and TX
+			 * submission never became active.
+			 */
+			qp->direct_state = NTB_DIRECT_QUIESCED;
+			return;
+		}
+		qp->direct_state = NTB_DIRECT_QUIESCING;
+	}
+
+	/*
+	 * Keep the mappings until the peer acknowledges the final boundaries,
+	 * or until the link is down or the peer starts a new session.
+	 */
+	for (;;) {
+		ntb_direct_control_progress(qp);
+
+		scoped_guard(spinlock_bh, &qp->direct_lock)
+			done = ntb_direct_tx_idle(qp) &&
+			       ntb_direct_tx_acked(qp) &&
+			       ntb_direct_rx_drained(qp);
+		if (done || ntb_link_is_up(qp->ndev, NULL, NULL) != 1)
+			break;
+
+		msleep(NTB_DIRECT_TEARDOWN_RETRY_INTERVAL_MS);
+	}
+
+	guard(spinlock_bh)(&qp->direct_lock);
+	qp->direct_state = NTB_DIRECT_QUIESCED;
+}
 
 /**
  * ntb_transport_rx_queue_size - Query the RX queue depth
@@ -1204,6 +1440,9 @@ static void ntb_qp_link_context_reset(struct ntb_transport_qp *qp)
 	qp->direct_rx_cons = 0;
 	qp->direct_tx_issue = 0;
 	qp->direct_tx_cons = 0;
+	qp->direct_session = 0;
+	qp->direct_peer_session = 0;
+	qp->direct_state = NTB_DIRECT_DOWN;
 }
 
 static void ntb_qp_link_down_reset(struct ntb_transport_qp *qp)
@@ -1221,6 +1460,7 @@ static void ntb_qp_link_cleanup(struct ntb_transport_qp *qp)
 	dev_info(&pdev->dev, "qp %d: Link Cleanup\n", qp->qp_num);
 
 	cancel_delayed_work_sync(&qp->link_work);
+	ntb_direct_quiesce(qp);
 	ntb_qp_link_down_reset(qp);
 
 	if (qp->event_handler)
@@ -1457,6 +1697,7 @@ static void ntb_qp_link_work(struct work_struct *work)
 						   link_work.work);
 	struct pci_dev *pdev = qp->ndev->pdev;
 	struct ntb_transport_ctx *nt = qp->transport;
+	bool direct_ready;
 	int val;
 
 	if (!qp->client_ready)
@@ -1466,13 +1707,17 @@ static void ntb_qp_link_work(struct work_struct *work)
 
 	val = ntb_spad_read(nt->ndev, QP_LINKS);
 
+	if (qp->direct_state == NTB_DIRECT_DOWN)
+		ntb_direct_session_start(qp);
+
 	ntb_peer_spad_write(nt->ndev, PIDX, QP_LINKS, val | BIT(qp->qp_num));
+	direct_ready = ntb_direct_control_progress(qp);
 
 	/* query remote spad for qp ready bits */
 	dev_dbg_ratelimited(&pdev->dev, "Remote QP link status = %x\n", val);
 
 	/* See if the remote side is up */
-	if (val & BIT(qp->qp_num)) {
+	if ((val & BIT(qp->qp_num)) && direct_ready) {
 		dev_info(&pdev->dev, "qp %d: Link Up\n", qp->qp_num);
 		qp->link_is_up = true;
 		qp->active = true;
@@ -1559,6 +1804,7 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 	spin_lock_init(&qp->ntb_rx_q_lock);
 	spin_lock_init(&qp->ntb_tx_free_q_lock);
 	spin_lock_init(&qp->ntb_tx_offl_q_lock);
+	spin_lock_init(&qp->direct_lock);
 
 	INIT_LIST_HEAD(&qp->rx_post_q);
 	INIT_LIST_HEAD(&qp->rx_pend_q);
@@ -2054,6 +2300,11 @@ static void ntb_transport_rxc_db(struct work_struct *work)
 	dev_dbg(&qp->ndev->pdev->dev, "%s: doorbell %d received\n",
 		__func__, qp->qp_num);
 
+	if (ntb_direct_control_pending(qp))
+		ntb_direct_control_progress(qp);
+	if (!qp->active)
+		goto clear_db;
+
 	/* Limit the number of packets processed in a single interrupt to
 	 * provide fairness to others
 	 */
@@ -2070,7 +2321,11 @@ static void ntb_transport_rxc_db(struct work_struct *work)
 		/* there is more work to do */
 		if (qp->active)
 			queue_work(system_dfl_wq, &qp->rxc_db_work);
-	} else if (ntb_db_read(qp->ndev) & BIT_ULL(qp->qp_num)) {
+		return;
+	}
+
+clear_db:
+	if (ntb_db_read(qp->ndev) & BIT_ULL(qp->qp_num)) {
 		/* the doorbell bit is set: clear it */
 		ntb_db_clear(qp->ndev, BIT_ULL(qp->qp_num));
 		/* ntb_db_read ensures ntb_db_clear write is committed */
@@ -2080,7 +2335,9 @@ static void ntb_transport_rxc_db(struct work_struct *work)
 		 * ntb_process_rxc and clearing the doorbell bit:
 		 * there might be some more work to do.
 		 */
-		if (qp->active)
+		if (qp->active ||
+		    (qp->client_ready &&
+		     READ_ONCE(qp->direct_state) == NTB_DIRECT_HANDSHAKE))
 			queue_work(system_dfl_wq, &qp->rxc_db_work);
 	}
 }
@@ -2128,10 +2385,7 @@ static void ntb_tx_copy_callback(void *data,
 	dma_mb();
 	ioread32(&hdr->flags);
 
-	if (qp->use_msi)
-		ntb_msi_peer_trigger(qp->ndev, PIDX, &qp->peer_msi_desc);
-	else
-		ntb_peer_db_set(qp->ndev, BIT_ULL(qp->qp_num));
+	ntb_transport_notify_peer(qp);
 
 	/* The entry length can only be zero if the packet is intended to be a
 	 * "link down" or similar.  Since no payload is being sent in these
@@ -2586,6 +2840,7 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 
 	cancel_work_sync(&qp->link_cleanup);
 	cancel_delayed_work_sync(&qp->link_work);
+	ntb_direct_quiesce(qp);
 	qp->active = false;
 
 	if (qp->tx_offload_thread) {
@@ -2636,6 +2891,7 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 	/* Catch cleanup queued while draining RX processing. */
 	cancel_work_sync(&qp->link_cleanup);
 	cancel_delayed_work_sync(&qp->link_work);
+	ntb_qp_link_context_reset(qp);
 
 	qp->cb_data = NULL;
 	qp->rx_handler = NULL;
@@ -2822,6 +3078,7 @@ void ntb_transport_link_down(struct ntb_transport_qp *qp)
 		return;
 
 	qp->client_ready = false;
+	ntb_direct_quiesce(qp);
 
 	val = ntb_spad_read(qp->ndev, QP_LINKS);
 
@@ -2927,7 +3184,9 @@ static void ntb_transport_doorbell_callback(void *data, int vector)
 		qp_num = __ffs(db_bits);
 		qp = &nt->qp_vec[qp_num];
 
-		if (qp->active)
+		if (qp->active ||
+		    (qp->client_ready &&
+		     READ_ONCE(qp->direct_state) == NTB_DIRECT_HANDSHAKE))
 			queue_work(system_dfl_wq, &qp->rxc_db_work);
 
 		db_bits &= ~BIT_ULL(qp_num);
-- 
2.51.0

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

[Koichiro Den]

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

[PATCH 12/16] NTB: ntb_transport: Implement direct-DMA TX submission

[Koichiro Den]

Completing a payload and then writing its completion from the CPU does
not guarantee that the peer observes those writes in the same order. It
could release the RX buffer while the payload is still in flight.

Submit each payload and its completion word on the same in-order DMA
channel. Track source mappings in ring order, and keep outstanding
mappings owned by the queue after a DMA error so teardown can stop the
channel before releasing them.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 152400feb6d6..4c4741b8f9b1 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -213,6 +213,8 @@ struct ntb_transport_qp {
 	struct list_head direct_rx_q;
 	u32 *direct_rx_cpl;
 	dma_addr_t direct_rx_cpl_dma;
+	/* Entries on the direct RX/TX queues own their DMA mappings. */
+	struct list_head direct_tx_q;
 	u32 *direct_tx_cpl;
 	dma_addr_t direct_tx_cpl_dma;
 	u32 direct_rx_prod;
@@ -222,6 +224,7 @@ struct ntb_transport_qp {
 	u32 direct_session;
 	u32 direct_peer_session;
 	enum ntb_direct_state direct_state;
+	bool direct_tx_failed;
 	void *rx_buff;
 	unsigned int rx_index;
 	unsigned int rx_max_entry;
@@ -707,6 +710,7 @@ static void ntb_direct_session_start(struct ntb_transport_qp *qp)
 	guard(spinlock_bh)(&qp->direct_lock);
 	qp->direct_session = session;
 	qp->direct_peer_session = 0;
+	qp->direct_tx_failed = false;
 	qp->direct_state = NTB_DIRECT_HANDSHAKE;
 }
 
@@ -1452,6 +1456,7 @@ static void ntb_qp_link_context_reset(struct ntb_transport_qp *qp)
 	qp->direct_session = 0;
 	qp->direct_peer_session = 0;
 	qp->direct_state = NTB_DIRECT_DOWN;
+	qp->direct_tx_failed = false;
 }
 
 static void ntb_qp_link_down_reset(struct ntb_transport_qp *qp)
@@ -1830,6 +1835,7 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 	INIT_LIST_HEAD(&qp->tx_free_q);
 	INIT_LIST_HEAD(&qp->tx_offl_q);
 	INIT_LIST_HEAD(&qp->direct_rx_q);
+	INIT_LIST_HEAD(&qp->direct_tx_q);
 
 	INIT_WORK(&qp->rxc_db_work, ntb_transport_rxc_db);
 	INIT_DELAYED_WORK(&qp->direct_rx_retry, ntb_direct_rx_retry_work);
@@ -2743,6 +2749,15 @@ static int ntb_async_tx_submit(struct ntb_transport_qp *qp,
 	return -ENXIO;
 }
 
+static unsigned int
+ntb_transport_mw_tx_free_entry(struct ntb_transport_qp *qp)
+{
+	unsigned int head = qp->tx_index;
+	unsigned int tail = qp->remote_rx_info->entry;
+
+	return tail >= head ? tail - head : qp->tx_max_entry + tail - head;
+}
+
 static void ntb_async_tx(struct ntb_transport_qp *qp,
 			 struct ntb_queue_entry *entry)
 {
@@ -2756,7 +2771,7 @@ static void ntb_async_tx(struct ntb_transport_qp *qp,
 	hdr = offset + qp->tx_max_frame - sizeof(struct ntb_payload_header);
 	entry->tx_hdr = hdr;
 
-	WARN_ON_ONCE(!ntb_transport_tx_free_entry(qp));
+	WARN_ON_ONCE(!ntb_transport_mw_tx_free_entry(qp));
 	WRITE_ONCE(qp->tx_index, (qp->tx_index + 1) % qp->tx_max_entry);
 
 	iowrite32(entry->len, &hdr->len);
@@ -2780,10 +2795,229 @@ static void ntb_async_tx(struct ntb_transport_qp *qp,
 	qp->tx_memcpy++;
 }
 
+static bool ntb_direct_tx_enabled(struct ntb_transport_qp *qp)
+{
+	return ntb_direct_tx_mode(qp) &&
+	       READ_ONCE(qp->direct_state) == NTB_DIRECT_ACTIVE;
+}
+
+static int ntb_direct_tx_publication(struct ntb_transport_qp *qp, u32 issue,
+				     dma_addr_t *dst, dma_addr_t *cpl_dst,
+				     unsigned int *len)
+{
+	struct ntb_direct_shared *shared = qp->direct_shared;
+	struct ntb_direct_pub *pub;
+	u32 head, idx, used;
+
+	lockdep_assert_held(&qp->direct_lock);
+
+	head = READ_ONCE(shared->pub_head);
+	used = ntb_direct_ring_used(head, issue);
+	if (!used)
+		return -ENOSPC;
+	if (used > qp->direct_ring_entries - 1)
+		return -EPROTO;
+
+	/* pub_head is written after the publication slot. */
+	dma_rmb();
+	idx = ntb_direct_ring_idx(qp, issue);
+	pub = &shared->pub[idx];
+
+	*dst = (u64)READ_ONCE(pub->addr_hi) << 32 |
+	       READ_ONCE(pub->addr_lo);
+	*cpl_dst = ((u64)READ_ONCE(shared->cpl_addr_hi) << 32 |
+		     READ_ONCE(shared->cpl_addr_lo)) + idx * sizeof(u32);
+	*len = READ_ONCE(pub->len);
+
+	return 0;
+}
+
+static void ntb_direct_tx_complete(struct ntb_transport_qp *qp,
+				   struct ntb_queue_entry *entry)
+{
+	int cb_len;
+
+	cb_len = entry->errors ? -EIO : entry->len;
+	if (!entry->errors) {
+		qp->tx_bytes += entry->len;
+		qp->tx_pkts++;
+	}
+
+	if (qp->tx_handler)
+		qp->tx_handler(qp, qp->cb_data, entry->cb_data, cb_len);
+	ntb_list_add(&qp->ntb_tx_free_q_lock, &entry->entry, &qp->tx_free_q);
+
+	if (ntb_link_is_up(qp->ndev, NULL, NULL) == 1)
+		ntb_transport_notify_peer(qp);
+}
+
+static bool ntb_direct_tx_fail(struct ntb_transport_qp *qp)
+{
+	bool cleanup = !qp->direct_tx_failed && qp->client_ready;
+
+	lockdep_assert_held(&qp->direct_lock);
+
+	qp->direct_tx_failed = true;
+	qp->direct_state = NTB_DIRECT_QUIESCING;
+
+	return cleanup;
+}
+
+static void ntb_direct_tx_callback(void *data,
+				   const struct dmaengine_result *res)
+{
+	struct ntb_queue_entry *entry = data;
+	struct ntb_transport_qp *qp = entry->qp;
+	struct device *dma_dev = dmaengine_get_dma_device(qp->direct_dma_chan);
+	bool error;
+
+	error = res && res->result != DMA_TRANS_NOERROR;
+	if (error) {
+		bool cleanup;
+
+		scoped_guard(spinlock_bh, &qp->direct_lock) {
+			entry->errors++;
+			cleanup = ntb_direct_tx_fail(qp);
+		}
+
+		ntb_direct_control_progress(qp);
+		if (cleanup && ntb_link_is_up(qp->ndev, NULL, NULL) == 1)
+			schedule_work(&qp->link_cleanup);
+		return;
+	}
+
+	dma_unmap_single(dma_dev, entry->direct_dma_addr, entry->len,
+			 DMA_TO_DEVICE);
+
+	/* The channel filter excludes out-of-order completion. */
+	scoped_guard(spinlock_bh, &qp->direct_lock)
+		list_del(&entry->entry);
+
+	ntb_direct_tx_complete(qp, entry);
+
+	scoped_guard(spinlock_bh, &qp->direct_lock)
+		qp->direct_tx_cons++;
+}
+
+static int ntb_direct_tx_submit(struct ntb_transport_qp *qp,
+				struct ntb_queue_entry *entry)
+{
+	struct dma_async_tx_descriptor *payload, *completion;
+	struct dma_chan *chan = qp->direct_dma_chan;
+	struct device *dma_dev = dmaengine_get_dma_device(chan);
+	struct dma_slave_config config = {};
+	dma_addr_t cpl_dma_addr, cpl_dst, dma_addr, dst;
+	unsigned int published_len;
+	bool schedule_cleanup = false;
+	bool terminate = false;
+	u32 issue, idx;
+	int rc;
+
+	if (entry->len > INT_MAX)
+		return -EMSGSIZE;
+
+	dma_addr = dma_map_single(dma_dev, entry->buf, entry->len,
+				  DMA_TO_DEVICE);
+	if (dma_mapping_error(dma_dev, dma_addr))
+		return -EIO;
+
+	scoped_guard(spinlock_bh, &qp->direct_lock) {
+		if (!ntb_direct_tx_enabled(qp)) {
+			rc = -ENOLINK;
+			goto unmap;
+		}
+
+		issue = qp->direct_tx_issue;
+		if (ntb_direct_ring_used(issue, qp->direct_tx_cons) >=
+		    qp->direct_ring_entries - 1) {
+			rc = -ENOSPC;
+			goto unmap;
+		}
+
+		rc = ntb_direct_tx_publication(qp, issue, &dst, &cpl_dst,
+					       &published_len);
+		if (rc)
+			goto unmap;
+		if (entry->len > published_len) {
+			rc = -EMSGSIZE;
+			goto unmap;
+		}
+
+		idx = ntb_direct_ring_idx(qp, issue);
+		entry->direct_dma_addr = dma_addr;
+		entry->direct_cpl_addr = cpl_dst;
+		qp->direct_tx_cpl[idx] = entry->len;
+		dma_wmb();
+
+		config.direction = DMA_MEM_TO_DEV;
+		config.dst_addr = dst;
+		payload = dmaengine_prep_config_single_safe(chan, dma_addr,
+							    entry->len,
+							    DMA_MEM_TO_DEV,
+							    DMA_CTRL_ACK,
+							    &config);
+
+		if (!payload) {
+			rc = -EIO;
+			goto unmap;
+		}
+
+		rc = dma_submit_error(dmaengine_submit(payload));
+		if (rc)
+			goto unmap;
+
+		list_add_tail(&entry->entry, &qp->direct_tx_q);
+		qp->direct_tx_issue = issue + 1;
+
+		config.dst_addr = cpl_dst;
+		cpl_dma_addr = qp->direct_tx_cpl_dma + idx * sizeof(u32);
+		completion = dmaengine_prep_config_single_safe(chan,
+							       cpl_dma_addr,
+							       sizeof(u32),
+							       DMA_MEM_TO_DEV,
+							       DMA_CTRL_ACK |
+							       DMA_PREP_INTERRUPT,
+							       &config);
+		if (!completion) {
+			schedule_cleanup = ntb_direct_tx_fail(qp);
+			terminate = true;
+		} else {
+			completion->callback_result = ntb_direct_tx_callback;
+			completion->callback_param = entry;
+
+			if (dma_submit_error(dmaengine_submit(completion))) {
+				schedule_cleanup = ntb_direct_tx_fail(qp);
+				terminate = true;
+			} else {
+				dma_async_issue_pending(chan);
+			}
+		}
+	}
+
+	if (terminate) {
+		dmaengine_terminate_async(chan);
+		if (schedule_cleanup && ntb_link_is_up(qp->ndev, NULL, NULL) == 1)
+			schedule_work(&qp->link_cleanup);
+	}
+
+	return 0;
+
+unmap:
+	dma_unmap_single(dma_dev, dma_addr, entry->len, DMA_TO_DEVICE);
+	return rc;
+}
+
 static int ntb_process_tx(struct ntb_transport_qp *qp,
 			  struct ntb_queue_entry *entry)
 {
-	if (!ntb_transport_tx_free_entry(qp)) {
+	if (entry->len && ntb_direct_tx_mode(qp)) {
+		if (!ntb_direct_tx_enabled(qp))
+			return -ENOLINK;
+
+		return ntb_direct_tx_submit(qp, entry);
+	}
+
+	if (!ntb_transport_mw_tx_free_entry(qp)) {
 		qp->tx_ring_full++;
 		return -EAGAIN;
 	}
@@ -3377,10 +3611,29 @@ EXPORT_SYMBOL_GPL(ntb_transport_max_size);
 
 unsigned int ntb_transport_tx_free_entry(struct ntb_transport_qp *qp)
 {
-	unsigned int head = qp->tx_index;
-	unsigned int tail = qp->remote_rx_info->entry;
+	u32 available, capacity, used;
 
-	return tail >= head ? tail - head : qp->tx_max_entry + tail - head;
+	if (ntb_direct_tx_mode(qp)) {
+		guard(spinlock_bh)(&qp->direct_lock);
+		if (!ntb_direct_tx_enabled(qp))
+			return 0;
+
+		capacity = qp->direct_ring_entries - 1;
+		used = ntb_direct_ring_used(qp->direct_tx_issue,
+					    qp->direct_tx_cons);
+		if (used >= capacity)
+			return 0;
+
+		available =
+			ntb_direct_ring_used(READ_ONCE(qp->direct_shared->pub_head),
+					     qp->direct_tx_issue);
+		if (available > capacity)
+			return 0;
+
+		return min(available, capacity - used);
+	}
+
+	return ntb_transport_mw_tx_free_entry(qp);
 }
 EXPORT_SYMBOL_GPL(ntb_transport_tx_free_entry);
 
-- 
2.51.0

[PATCH 13/16] NTB: ntb_transport: Implement safe direct-DMA teardown

[Koichiro Den]

A logical link-down does not prove that local DMA or the peer has
stopped using direct mappings. Releasing them can therefore allow DMA
access after unmapping.

Serialize quiesce and keep mappings until both directions reach their
final issued boundaries. On DMA failure or physical link loss, stop and
synchronize the local channel first. If the link remains usable, send
error completions on that channel before releasing outstanding mappings.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 4c4741b8f9b1..f2fa145994b9 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -225,6 +225,7 @@ struct ntb_transport_qp {
 	u32 direct_peer_session;
 	enum ntb_direct_state direct_state;
 	bool direct_tx_failed;
+	struct mutex direct_quiesce_lock; /* Serialize direct teardown */
 	void *rx_buff;
 	unsigned int rx_index;
 	unsigned int rx_max_entry;
@@ -569,6 +570,19 @@ static bool ntb_direct_tx_idle(struct ntb_transport_qp *qp)
 	return qp->direct_tx_issue == qp->direct_tx_cons;
 }
 
+static bool ntb_direct_peer_restarted(struct ntb_transport_qp *qp)
+{
+	struct ntb_direct_shared *shared = qp->direct_shared;
+	u32 peer_session = qp->direct_peer_session;
+	u32 session;
+
+	if (!shared || !peer_session)
+		return false;
+
+	session = READ_ONCE(shared->session);
+	return session && session != peer_session;
+}
+
 static bool ntb_direct_rx_drained(struct ntb_transport_qp *qp)
 {
 	struct ntb_direct_shared *shared = qp->direct_shared;
@@ -611,6 +625,8 @@ static bool ntb_direct_control_pending(struct ntb_transport_qp *qp)
 	       READ_ONCE(shared->quiesce) == session;
 }
 
+static void ntb_direct_tx_terminate(struct ntb_transport_qp *qp);
+
 static void ntb_direct_control_publish_locked(struct ntb_transport_qp *qp)
 {
 	struct ntb_direct_shared __iomem *peer = qp->peer_direct_shared;
@@ -664,9 +680,13 @@ static bool ntb_direct_control_progress(struct ntb_transport_qp *qp)
 
 		if (qp->direct_state == NTB_DIRECT_HANDSHAKE && peer_session) {
 			qp->direct_peer_session = peer_session;
-		} else if (qp->direct_state == NTB_DIRECT_ACTIVE && peer_session &&
+		} else if ((qp->direct_state == NTB_DIRECT_ACTIVE ||
+			    qp->direct_state == NTB_DIRECT_QUIESCING) &&
+			   peer_session &&
 			   peer_session != qp->direct_peer_session) {
 			qp->direct_state = NTB_DIRECT_QUIESCING;
+			cleanup = qp->client_ready;
+			goto out;
 		}
 
 		if (quiesce == qp->direct_session &&
@@ -714,13 +734,28 @@ static void ntb_direct_session_start(struct ntb_transport_qp *qp)
 	qp->direct_state = NTB_DIRECT_HANDSHAKE;
 }
 
+static void ntb_direct_begin_quiesce(struct ntb_transport_qp *qp)
+{
+	if (!ntb_direct_link_capable(qp))
+		return;
+
+	guard(spinlock_bh)(&qp->direct_lock);
+	if (qp->direct_state == NTB_DIRECT_HANDSHAKE)
+		qp->direct_state = qp->direct_peer_session ?
+			NTB_DIRECT_QUIESCING : NTB_DIRECT_QUIESCED;
+	else if (qp->direct_state == NTB_DIRECT_ACTIVE)
+		qp->direct_state = NTB_DIRECT_QUIESCING;
+}
+
 static void ntb_direct_quiesce(struct ntb_transport_qp *qp)
 {
-	bool done;
+	bool link_down, peer_reset, terminate, done;
 
 	if (!ntb_direct_link_capable(qp))
 		return;
 
+	guard(mutex)(&qp->direct_quiesce_lock);
+
 	scoped_guard(spinlock_bh, &qp->direct_lock) {
 		if (qp->direct_state == NTB_DIRECT_DOWN ||
 		    qp->direct_state == NTB_DIRECT_QUIESCED)
@@ -734,7 +769,16 @@ static void ntb_direct_quiesce(struct ntb_transport_qp *qp)
 			return;
 		}
 		qp->direct_state = NTB_DIRECT_QUIESCING;
+		link_down = ntb_link_is_up(qp->ndev, NULL, NULL) != 1;
+		peer_reset = ntb_direct_peer_restarted(qp);
+		terminate = qp->direct_tx_failed || link_down || peer_reset;
 	}
+	if (terminate)
+		ntb_direct_tx_terminate(qp);
+
+	/* A peer starts a new session only after draining the old boundary. */
+	if (link_down || peer_reset)
+		goto quiesced;
 
 	/*
 	 * Keep the mappings until the peer acknowledges the final boundaries,
@@ -743,16 +787,31 @@ static void ntb_direct_quiesce(struct ntb_transport_qp *qp)
 	for (;;) {
 		ntb_direct_control_progress(qp);
 
-		scoped_guard(spinlock_bh, &qp->direct_lock)
+		scoped_guard(spinlock_bh, &qp->direct_lock) {
+			peer_reset = ntb_direct_peer_restarted(qp);
+			terminate = qp->direct_tx_failed || peer_reset;
 			done = ntb_direct_tx_idle(qp) &&
 			       ntb_direct_tx_acked(qp) &&
 			       ntb_direct_rx_drained(qp);
-		if (done || ntb_link_is_up(qp->ndev, NULL, NULL) != 1)
+		}
+
+		if (terminate) {
+			ntb_direct_tx_terminate(qp);
+			if (peer_reset)
+				goto quiesced;
+			continue;
+		}
+		if (ntb_link_is_up(qp->ndev, NULL, NULL) != 1) {
+			ntb_direct_tx_terminate(qp);
+			break;
+		}
+		if (done)
 			break;
 
 		msleep(NTB_DIRECT_TEARDOWN_RETRY_INTERVAL_MS);
 	}
 
+quiesced:
 	guard(spinlock_bh)(&qp->direct_lock);
 	qp->direct_state = NTB_DIRECT_QUIESCED;
 }
@@ -1521,6 +1580,7 @@ static void ntb_transport_link_cleanup(struct ntb_transport_ctx *nt)
 	for (i = 0; i < nt->qp_count; i++)
 		if (qp_bitmap_alloc & BIT_ULL(i)) {
 			qp = &nt->qp_vec[i];
+			ntb_direct_begin_quiesce(qp);
 			ntb_qp_link_cleanup(qp);
 			cancel_work_sync(&qp->link_cleanup);
 			cancel_delayed_work_sync(&qp->link_work);
@@ -1828,6 +1888,7 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 	spin_lock_init(&qp->ntb_tx_free_q_lock);
 	spin_lock_init(&qp->ntb_tx_offl_q_lock);
 	spin_lock_init(&qp->direct_lock);
+	mutex_init(&qp->direct_quiesce_lock);
 
 	INIT_LIST_HEAD(&qp->rx_post_q);
 	INIT_LIST_HEAD(&qp->rx_pend_q);
@@ -3007,6 +3068,118 @@ static int ntb_direct_tx_submit(struct ntb_transport_qp *qp,
 	return rc;
 }
 
+static void ntb_direct_tx_stop(struct ntb_transport_qp *qp)
+{
+	struct dma_chan *chan = qp->direct_dma_chan;
+	int rc;
+
+	/* Do not release mappings until the DMA channel is confirmed stopped. */
+	do {
+		rc = dmaengine_terminate_sync(chan);
+		if (rc) {
+			dev_err_ratelimited(&qp->ndev->dev,
+					    "QP%u direct DMA termination failed: %d\n",
+					    qp->qp_num, rc);
+			msleep(NTB_DIRECT_TEARDOWN_RETRY_INTERVAL_MS);
+		}
+	} while (rc);
+}
+
+/*
+ * A CPU MMIO completion could pass an earlier DMA payload. Once the failed
+ * channel is stopped, publish terminal completions through that channel too.
+ */
+static bool ntb_direct_tx_publish_error(struct ntb_transport_qp *qp,
+					struct ntb_queue_entry *entry,
+					u32 idx)
+{
+	struct dma_async_tx_descriptor *completion;
+	struct dma_slave_config config = {
+		.direction = DMA_MEM_TO_DEV,
+		.dst_addr = entry->direct_cpl_addr,
+	};
+	struct dma_chan *chan = qp->direct_dma_chan;
+	dma_cookie_t cookie;
+	enum dma_status status;
+	dma_addr_t src;
+
+	qp->direct_tx_cpl[idx] = NTB_DIRECT_CPL_ERROR;
+	dma_wmb();
+	src = qp->direct_tx_cpl_dma + idx * sizeof(u32);
+
+	while (!ntb_direct_peer_restarted(qp) &&
+	       ntb_link_is_up(qp->ndev, NULL, NULL) == 1) {
+		completion = dmaengine_prep_config_single_safe(chan, src, sizeof(u32),
+							       DMA_MEM_TO_DEV,
+							       DMA_CTRL_ACK |
+							       DMA_PREP_INTERRUPT,
+							       &config);
+		if (completion) {
+			cookie = dmaengine_submit(completion);
+			if (!dma_submit_error(cookie)) {
+				dma_async_issue_pending(chan);
+				status = dma_sync_wait(chan, cookie);
+				if (status == DMA_COMPLETE)
+					return true;
+			}
+		}
+
+		ntb_direct_tx_stop(qp);
+		dev_err_ratelimited(&qp->ndev->dev,
+				    "QP%u direct DMA error completion failed; retrying\n",
+				    qp->qp_num);
+		msleep(NTB_DIRECT_TEARDOWN_RETRY_INTERVAL_MS);
+	}
+
+	return false;
+}
+
+static void ntb_direct_tx_terminate(struct ntb_transport_qp *qp)
+{
+	struct dma_chan *chan = qp->direct_dma_chan;
+	struct ntb_queue_entry *entry;
+	struct device *dma_dev;
+	bool notify = false;
+	u32 idx;
+
+	lockdep_assert_held(&qp->direct_quiesce_lock);
+	WARN_ON_ONCE(ntb_direct_tx_enabled(qp));
+
+	if (!chan)
+		return;
+
+	dma_dev = dmaengine_get_dma_device(chan);
+	ntb_direct_tx_stop(qp);
+
+	/*
+	 * QUIESCING blocks new submissions, and terminate_sync() has drained all
+	 * callbacks, so this function exclusively owns the issued queue.
+	 * Therefore, qp->direct_lock is not needed while draining it.
+	 */
+	while (!ntb_direct_tx_idle(qp)) {
+		idx = ntb_direct_ring_idx(qp, qp->direct_tx_cons);
+		entry = list_first_entry(&qp->direct_tx_q,
+					 struct ntb_queue_entry, entry);
+
+		notify |= ntb_direct_tx_publish_error(qp, entry, idx);
+
+		list_del(&entry->entry);
+		qp->direct_tx_cons++;
+
+		dma_unmap_single(dma_dev, entry->direct_dma_addr, entry->len,
+				 DMA_TO_DEVICE);
+		entry->errors++;
+		if (qp->tx_handler)
+			qp->tx_handler(qp, qp->cb_data, entry->cb_data, -EIO);
+		ntb_list_add(&qp->ntb_tx_free_q_lock, &entry->entry,
+			     &qp->tx_free_q);
+	}
+	qp->direct_tx_failed = false;
+
+	if (notify && ntb_link_is_up(qp->ndev, NULL, NULL) == 1)
+		ntb_transport_notify_peer(qp);
+}
+
 static int ntb_process_tx(struct ntb_transport_qp *qp,
 			  struct ntb_queue_entry *entry)
 {
@@ -3287,9 +3460,12 @@ void ntb_transport_free_queue(struct ntb_transport_qp *qp)
 
 	pdev = qp->ndev->pdev;
 
+	ntb_direct_begin_quiesce(qp);
 	cancel_work_sync(&qp->link_cleanup);
 	cancel_delayed_work_sync(&qp->link_work);
 	ntb_direct_quiesce(qp);
+	if (qp->direct_dma_chan)
+		dmaengine_synchronize(qp->direct_dma_chan);
 	qp->active = false;
 
 	if (qp->tx_offload_thread) {
@@ -3525,22 +3701,37 @@ EXPORT_SYMBOL_GPL(ntb_transport_link_up);
  */
 void ntb_transport_link_down(struct ntb_transport_qp *qp)
 {
+	bool direct;
 	int val;
 
 	if (!qp)
 		return;
 
 	qp->client_ready = false;
+	direct = ntb_direct_link_capable(qp);
+	if (direct) {
+		ntb_direct_begin_quiesce(qp);
+		disable_work_sync(&qp->link_cleanup);
+		cancel_delayed_work_sync(&qp->link_work);
+	}
 	ntb_direct_quiesce(qp);
 
 	val = ntb_spad_read(qp->ndev, QP_LINKS);
 
 	ntb_peer_spad_write(qp->ndev, PIDX, QP_LINKS, val & ~BIT(qp->qp_num));
 
-	if (qp->link_is_up)
-		ntb_send_link_down(qp);
-	else
+	if (qp->link_is_up) {
+		/* An established direct session uses QUIESCE as link-down. */
+		if (direct && ntb_direct_tx_acked(qp))
+			ntb_qp_link_down_reset(qp);
+		else
+			ntb_send_link_down(qp);
+	} else {
 		cancel_delayed_work_sync(&qp->link_work);
+	}
+
+	if (direct)
+		enable_work(&qp->link_cleanup);
 }
 EXPORT_SYMBOL_GPL(ntb_transport_link_down);
 
-- 
2.51.0

[PATCH 14/16] NTB: ntb_transport: Enable direct-DMA queues

[Koichiro Den]

Advertise direct RX when a DMA mapping device is available, and direct
TX only when every QP has a suitable channel. Negotiation can then
enable either direction independently.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index f2fa145994b9..9f5d6a407be7 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -554,7 +554,7 @@ static bool ntb_direct_tx_mode(struct ntb_transport_qp *qp)
 
 static bool ntb_direct_link_capable(struct ntb_transport_qp *qp)
 {
-	return ntb_direct_rx_mode(qp) || ntb_direct_tx_mode(qp);
+	return ntb_direct_layout(qp->transport);
 }
 
 static void ntb_transport_notify_peer(struct ntb_transport_qp *qp)
@@ -2058,6 +2058,11 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
 	else if (use_direct_dma)
 		dev_info(&ndev->dev,
 			 "not enough scratchpads for direct DMA negotiation\n");
+	if (nt->direct_dma_dev) {
+		nt->direct_features = NTB_DIRECT_FEAT_RX;
+		if (nt->qp_count && nt->qp_vec[0].direct_dma_chan)
+			nt->direct_features |= NTB_DIRECT_FEAT_TX;
+	}
 
 	mutex_init(&nt->link_event_lock);
 	INIT_DELAYED_WORK(&nt->link_work, ntb_transport_link_work);
-- 
2.51.0

[PATCH 15/16] NTB: ntb_transport: Report the direct-DMA payload limit

[Koichiro Den]

ntb_transport_max_size() currently reports the copy-path limit, which
reserves space for struct ntb_payload_header. Direct DMA bypasses that
layout, so the same limit is unnecessarily restrictive.

For active direct-DMA TX, derive the limit from the DMA device's segment
and mapping limits. Keep it within INT_MAX because transport callbacks
report lengths as int. Continue reporting the copy-path limit until direct
TX is active.

This lets ntb_netdev use ETH_MAX_MTU over direct DMA. If the queue later
falls back to the copy path, larger frames are rejected.

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

diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 9f5d6a407be7..2f69328d5201 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -3787,10 +3787,19 @@ unsigned int ntb_transport_max_size(struct ntb_transport_qp *qp)
 	unsigned int max_size;
 	unsigned int copy_align;
 	struct dma_chan *rx_chan, *tx_chan;
+	struct device *dma_dev;
 
 	if (!qp)
 		return 0;
 
+	if (ntb_transport_link_query(qp) && ntb_direct_tx_enabled(qp)) {
+		dma_dev = dmaengine_get_dma_device(qp->direct_dma_chan);
+		max_size = min_t(size_t, dma_get_max_seg_size(dma_dev),
+				 dma_max_mapping_size(dma_dev));
+
+		return min_t(unsigned int, max_size, INT_MAX);
+	}
+
 	rx_chan = qp->rx_dma_chan;
 	tx_chan = qp->tx_dma_chan;
 
-- 
2.51.0

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

[Koichiro Den]

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

Re: [PATCH 14/16] NTB: ntb_transport: Enable direct-DMA queues

[Koichiro Den]

On Tue, Aug 11, 2026 at 01:51:33AM +0900, Koichiro Den wrote:
> Advertise direct RX when a DMA mapping device is available, and direct
> TX only when every QP has a suitable channel. Negotiation can then
> enable either direction independently.
> 
> Signed-off-by: Koichiro Den <den@valinux.co.jp>
> ---
>  drivers/ntb/ntb_transport.c | 7 ++++++-
>  1 file changed, 6 insertions(+), 1 deletion(-)
> 
> diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
> index f2fa145994b9..9f5d6a407be7 100644
> --- a/drivers/ntb/ntb_transport.c
> +++ b/drivers/ntb/ntb_transport.c
> @@ -554,7 +554,7 @@ static bool ntb_direct_tx_mode(struct ntb_transport_qp *qp)
>  
>  static bool ntb_direct_link_capable(struct ntb_transport_qp *qp)
>  {
> -	return ntb_direct_rx_mode(qp) || ntb_direct_tx_mode(qp);
> +	return ntb_direct_layout(qp->transport);
>  }

Sorry, I got distracted while rearranging and polishing the series at the last
minute, leaving this meaningless intermediate diff here. :(

I'll properly fold it into the earlier commit in the next revision, so please
disregard this hunk for now. Sorry for the noise.

Best regards,
Koichiro

>  
>  static void ntb_transport_notify_peer(struct ntb_transport_qp *qp)
> @@ -2058,6 +2058,11 @@ static int ntb_transport_probe(struct ntb_client *self, struct ntb_dev *ndev)
>  	else if (use_direct_dma)
>  		dev_info(&ndev->dev,
>  			 "not enough scratchpads for direct DMA negotiation\n");
> +	if (nt->direct_dma_dev) {
> +		nt->direct_features = NTB_DIRECT_FEAT_RX;
> +		if (nt->qp_count && nt->qp_vec[0].direct_dma_chan)
> +			nt->direct_features |= NTB_DIRECT_FEAT_TX;
> +	}
>  
>  	mutex_init(&nt->link_event_lock);
>  	INIT_DELAYED_WORK(&nt->link_work, ntb_transport_link_work);
> -- 
> 2.51.0
> 

Re: [PATCH 01/16] NTB: ntb_transport: Abort link setup on QP MW allocation failure

[Frank Li]

On Tue, Aug 11, 2026 at 01:51:20AM +0900, Koichiro Den wrote:
> ntb_transport_setup_qp_mw() can fail while growing a QP's RX entry pool,
> but the link worker ignores that error. The worker can then publish a QP
> whose memory-window state is only partly initialized, and later work can
> use stale or incomplete pointers.
>
> Set up every QP memory window before publishing the transport link. On
> failure, clear the QP pointers before releasing its MW backing and leave
> the link down.
>
> Fixes: a754a8fcaf38 ("NTB: allocate number transport entries depending on size of ring size")
> Signed-off-by: Koichiro Den <den@valinux.co.jp>
> ---
>  drivers/ntb/ntb_transport.c | 20 ++++++++++++++++----
>  1 file changed, 16 insertions(+), 4 deletions(-)
>
> diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
> index f59f926d4bfa..3efc50955253 100644
> --- a/drivers/ntb/ntb_transport.c
> +++ b/drivers/ntb/ntb_transport.c
> @@ -1084,14 +1084,19 @@ static void ntb_transport_link_work(struct work_struct *work)
>  			goto out1;
>  	}
>
> -	nt->link_is_up = true;
> +	nt->link_is_up = false;
> +	for (i = 0; i < nt->qp_count; i++) {
> +		rc = ntb_transport_setup_qp_mw(nt, i);
> +		if (rc)
> +			goto out1;
> +		ntb_transport_setup_qp_peer_msi(nt, i);
> +	}
>
> +	/* Publish the link only after every QP has been set up. */
> +	nt->link_is_up = true;

Not sure if need WRITE_ONCE() or other memory barrier to make sure
ntb_transport_setup_qp_mw() and ntb_transport_setup_qp_peer_msi() actually
complete before set this flag.

Frank

>  	for (i = 0; i < nt->qp_count; i++) {
>  		struct ntb_transport_qp *qp = &nt->qp_vec[i];
>
> -		ntb_transport_setup_qp_mw(nt, i);
> -		ntb_transport_setup_qp_peer_msi(nt, i);
> -
>  		if (qp->client_ready)
>  			schedule_delayed_work(&qp->link_work, 0);
>  	}
> @@ -1099,6 +1104,13 @@ static void ntb_transport_link_work(struct work_struct *work)
>  	return;
>
>  out1:
> +	for (i = 0; i < nt->qp_count; i++) {
> +		struct ntb_transport_qp *qp = &nt->qp_vec[i];
> +
> +		qp->rx_buff = NULL;
> +		qp->remote_rx_info = NULL;
> +	}
> +
>  	for (i = 0; i < nt->mw_count; i++)
>  		ntb_free_mw(nt, i);
>
> --
> 2.51.0
>