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

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

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.

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