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

@@ -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) {