[PATCH net-next v4 05/10] NTB: ntb_transport: Exchange client capabilities at link-up

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

Clients need to learn peer capabilities before sending data. Exchange
opaque 31-bit capabilities during QP link setup and pass them to the
client with the link-up event.

Use one scratchpad per QP after the MW sizes and MSI descriptors. The
top bit marks a valid advertisement. Write capabilities to the peer
before advertising QP readiness. Read peer capabilities from the local
scratchpad once the peer reports ready.

Clear the local capability scratchpad at probe, not on QP open or reset,
so early peer advertisements are not lost.
Withdraw advertisements on link-down. The existing transport link
cleanup clears all local scratchpads.

The transport version 4 layout is kept as-is, including MSI slots even
without local MSI. Report zero capabilities without extra retries for
legacy peers or QPs without spare scratchpads.

Adapt ntb_netdev to the API without using capabilities yet.

Signed-off-by: Koichiro Den <den@valinux.co.jp>
---
Changes in v4:
  - New patch to address Sashiko's feedback:
    https://lore.kernel.org/r/xsaocp32a6siggahziu4jqbwgmdasmp6v64pd5aujfkoxhiqws@n5luestlntzc/

For reviewers: we could also protect qp->client_ready and qp->local_caps
with ntb_rx_q_lock. I find the current approach easier to read, but I'm
happy to change it if preferred.

 drivers/net/ntb_netdev.c      |  8 ++---
 drivers/ntb/ntb_transport.c   | 60 +++++++++++++++++++++++++++++++----
 include/linux/ntb_transport.h |  7 ++--
 3 files changed, 62 insertions(+), 13 deletions(-)

diff --git a/drivers/net/ntb_netdev.c b/drivers/net/ntb_netdev.c
index 7a0d5e892a1a..869fc9a7f9e8 100644
--- a/drivers/net/ntb_netdev.c
+++ b/drivers/net/ntb_netdev.c
@@ -101,7 +101,7 @@ static int ntb_netdev_queue_rx_fill(struct net_device *ndev,
 	return 0;
 }
 
-static void ntb_netdev_event_handler(void *data, int link_is_up)
+static void ntb_netdev_event_handler(void *data, int link_is_up, u32 peer_caps)
 {
 	struct ntb_netdev_queue *q = data;
 	struct ntb_netdev *dev = q->ntdev;
@@ -346,7 +346,7 @@ static int ntb_netdev_open(struct net_device *ndev)
 	netif_tx_stop_all_queues(ndev);
 
 	for (q = 0; q < dev->num_queues; q++)
-		ntb_transport_link_up(dev->queues[q].qp);
+		ntb_transport_link_up(dev->queues[q].qp, 0);
 
 	return 0;
 
@@ -430,7 +430,7 @@ static int ntb_netdev_change_mtu(struct net_device *ndev, int new_mtu)
 	WRITE_ONCE(ndev->mtu, new_mtu);
 
 	for (q = 0; q < dev->num_queues; q++)
-		ntb_transport_link_up(dev->queues[q].qp);
+		ntb_transport_link_up(dev->queues[q].qp, 0);
 
 	return 0;
 
@@ -538,7 +538,7 @@ static int ntb_inc_channels(struct net_device *ndev,
 
 	if (running)
 		for (q = old; q < new; q++)
-			ntb_transport_link_up(dev->queues[q].qp);
+			ntb_transport_link_up(dev->queues[q].qp, 0);
 
 	return 0;
 
diff --git a/drivers/ntb/ntb_transport.c b/drivers/ntb/ntb_transport.c
index 0b47285ef48b..ea89eb336472 100644
--- a/drivers/ntb/ntb_transport.c
+++ b/drivers/ntb/ntb_transport.c
@@ -145,6 +145,8 @@ struct ntb_transport_qp {
 	bool client_ready;
 	bool link_is_up;
 	bool active;
+	u32 local_caps;
+	unsigned int caps_spad;
 
 	u8 qp_num;	/* Only 64 QP's are allowed.  0-63 */
 	u64 qp_bit;
@@ -181,7 +183,7 @@ struct ntb_transport_qp {
 	dma_cookie_t last_cookie;
 	struct tasklet_struct rxc_db_work;
 
-	void (*event_handler)(void *data, int status);
+	void (*event_handler)(void *data, int status, u32 peer_caps);
 	struct delayed_work link_work;
 	struct work_struct link_cleanup;
 
@@ -282,6 +284,9 @@ enum {
 	MW0_SZ_LOW,
 };
 
+/* One per-QP scratchpad, with the remaining bits owned by the client. */
+#define QP_CAPS_VALID		BIT(31)
+
 #define dev_client_dev(__dev) \
 	container_of((__dev), struct ntb_transport_client_dev, dev)
 
@@ -937,7 +942,7 @@ static void ntb_qp_link_cleanup(struct ntb_transport_qp *qp)
 	ntb_qp_link_down_reset(qp);
 
 	if (qp->event_handler)
-		qp->event_handler(qp->cb_data, qp->link_is_up);
+		qp->event_handler(qp->cb_data, qp->link_is_up, 0);
 }
 
 static void ntb_qp_link_cleanup_work(struct work_struct *work)
@@ -1141,10 +1146,19 @@ 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;
+	u32 peer_caps = 0;
 	int val;
 
 	WARN_ON(!nt->link_is_up);
 
+	/* Pair with the release store in ntb_transport_link_up(). */
+	if (!smp_load_acquire(&qp->client_ready))
+		return;
+
+	/* Publish capabilities before QP readiness. */
+	if (qp->caps_spad)
+		ntb_peer_spad_write(nt->ndev, PIDX, qp->caps_spad,
+				    READ_ONCE(qp->local_caps) | QP_CAPS_VALID);
 	val = ntb_spad_read(nt->ndev, QP_LINKS);
 
 	ntb_qp_up_request(qp, true);
@@ -1154,12 +1168,26 @@ static void ntb_qp_link_work(struct work_struct *work)
 
 	/* See if the remote side is up */
 	if (val & BIT(qp->qp_num)) {
+		if (qp->caps_spad) {
+			u32 caps;
+
+			/*
+			 * Order the readiness read before the capability read
+			 * for memory-backed SPADs.
+			 */
+			dma_rmb();
+			caps = ntb_spad_read(nt->ndev, qp->caps_spad);
+
+			if (caps & QP_CAPS_VALID)
+				peer_caps = caps & ~QP_CAPS_VALID;
+		}
+
 		dev_info(&pdev->dev, "qp %d: Link Up\n", qp->qp_num);
 		qp->link_is_up = true;
 		qp->active = true;
 
 		if (qp->event_handler)
-			qp->event_handler(qp->cb_data, qp->link_is_up);
+			qp->event_handler(qp->cb_data, qp->link_is_up, peer_caps);
 
 		if (qp->active)
 			tasklet_schedule(&qp->rxc_db_work);
@@ -1189,6 +1217,12 @@ static int ntb_transport_init_queue(struct ntb_transport_ctx *nt,
 	qp->ndev = nt->ndev;
 	qp->client_ready = false;
 	qp->event_handler = NULL;
+	/* Reserve MSI slots even when only the peer might use them. */
+	qp->caps_spad = nt->msi_spad_offset + 2 * qp_count + qp_num;
+	if (qp->caps_spad >= ntb_spad_count(nt->ndev))
+		qp->caps_spad = 0;
+	else
+		ntb_spad_write(qp->ndev, qp->caps_spad, 0);
 	ntb_qp_link_context_reset(qp);
 
 	if (mw_num < qp_count % mw_count)
@@ -2409,15 +2443,22 @@ EXPORT_SYMBOL_GPL(ntb_transport_tx_enqueue);
 /**
  * ntb_transport_link_up - Notify NTB transport of client readiness to use queue
  * @qp: NTB transport layer queue to be enabled
+ * @local_caps: Opaque client capabilities in bits 0..30, unchanged until
+ *              ntb_transport_link_down()
  *
  * Notify NTB transport layer of client readiness to use queue
+ *
+ * Exchange capabilities before reporting link-up through event_handler.
+ * Report zero peer capabilities for legacy peers or insufficient scratchpads.
  */
-void ntb_transport_link_up(struct ntb_transport_qp *qp)
+void ntb_transport_link_up(struct ntb_transport_qp *qp, u32 local_caps)
 {
 	if (!qp)
 		return;
 
-	qp->client_ready = true;
+	WRITE_ONCE(qp->local_caps, local_caps & ~QP_CAPS_VALID);
+	/* Publish local_caps before QP link work sees client_ready. */
+	smp_store_release(&qp->client_ready, true);
 
 	if (qp->transport->link_is_up)
 		schedule_delayed_work(&qp->link_work, 0);
@@ -2439,12 +2480,17 @@ void ntb_transport_link_down(struct ntb_transport_qp *qp)
 
 	qp->client_ready = false;
 
+	if (!qp->link_is_up)
+		cancel_delayed_work_sync(&qp->link_work);
+
+	/* Stop advertising capabilities before withdrawing QP readiness. */
+	if (qp->caps_spad)
+		ntb_peer_spad_write(qp->ndev, PIDX, qp->caps_spad, 0);
+
 	ntb_qp_up_request(qp, false);
 
 	if (qp->link_is_up)
 		ntb_send_link_down(qp);
-	else
-		cancel_delayed_work_sync(&qp->link_work);
 }
 EXPORT_SYMBOL_GPL(ntb_transport_link_down);
 
diff --git a/include/linux/ntb_transport.h b/include/linux/ntb_transport.h
index 7243eb98a722..685dde629a48 100644
--- a/include/linux/ntb_transport.h
+++ b/include/linux/ntb_transport.h
@@ -48,6 +48,8 @@
  * Jon Mason <jon.mason@intel.com>
  */
 
+#include <linux/types.h>
+
 struct ntb_transport_qp;
 
 struct ntb_transport_client {
@@ -66,7 +68,8 @@ struct ntb_queue_handlers {
 			   void *data, int len);
 	void (*tx_handler)(struct ntb_transport_qp *qp, void *qp_data,
 			   void *data, int len);
-	void (*event_handler)(void *data, int status);
+	/* peer_caps is 31-bit, zero on link-down or without peer support. */
+	void (*event_handler)(void *data, int status, u32 peer_caps);
 };
 
 unsigned char ntb_transport_qp_num(struct ntb_transport_qp *qp);
@@ -80,7 +83,7 @@ int ntb_transport_rx_enqueue(struct ntb_transport_qp *qp, void *cb, void *data,
 int ntb_transport_tx_enqueue(struct ntb_transport_qp *qp, void *cb, void *data,
 			     unsigned int len);
 void *ntb_transport_rx_remove(struct ntb_transport_qp *qp, unsigned int *len);
-void ntb_transport_link_up(struct ntb_transport_qp *qp);
+void ntb_transport_link_up(struct ntb_transport_qp *qp, u32 local_caps);
 void ntb_transport_link_down(struct ntb_transport_qp *qp);
 bool ntb_transport_link_query(struct ntb_transport_qp *qp);
 unsigned int ntb_transport_tx_free_entry(struct ntb_transport_qp *qp);
-- 
2.51.0