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

[Dave Jiang <dave.jiang@intel.com>]

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 <den@valinux.co.jp>
>>> ---
>>>  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.

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);
>>