# Re: \[PATCH 10/16\] NTB: ntb\_transport: Implement direct-DMA QP session handshake

- From: Koichiro Den \<hidden\>
- Date: 2026-08-28 17:12:02 
- List: netdev
- Page: https://ratatoskr.run/netdev/2026/08/17476293
- Thread: https://ratatoskr.run/netdev/2026/08/17392799/t.md

```text
On Fri, Aug 28, 2026 at 07:47:35AM -0700, Dave Jiang wrote:
> 
> 
> On 8/27/26 6:32 PM, Koichiro Den wrote:
> > On Thu, Aug 27, 2026 at 09:03:42AM -0700, Dave Jiang wrote:
> >>
> >>
> >> On 8/10/26 9:51 AM, Koichiro Den wrote:
> >>> 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 <redacted>
> >>> ---
> >>>  drivers/ntb/ntb_transport.c | 275 ++++++++++++++++++++++++++++++++++--
> >>
> >> Will you considered moving most of the ntb_direct code to ntb_transport_direct.c instead of increasing the size of ntb_transport.c? Just from an organization perspective this may be better for maintenance and reading in the future.
> > 
> > Thanks for the comment. Let me rethink how best to split it out.
> > 
> > (For v1, I thought it would be fine to keep it in ntb_transport.c, because,
> > compared to the previous ntb_transport_edma RFC, it can now fall back to the
> > standard MW memcpy path, and the vendor-specific parts are gone after moving
> > them under the standard dmaengine API. I had thought keeping it together would
> > be simpler, so thanks for the feedback.)
> 
> I just mean a different C source file with a shared private header file. Still same module.

I see. Thanks for the suggestion!

Best regards,
Koichiro

> 
> DJ> 
> > Best regards,
> > Koichiro
> > 
> >>
> >> DJ
> >>
> >>>  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);
> >>
> 
> 
```