15 files changed, 7379 insertions, 0 deletions
diff --git a/fs/smb/smbdirect/Kconfig b/fs/smb/smbdirect/Kconfig
new file mode 100644
index 000000000000..2bf1b3350f7d
--- /dev/null
+++ b/fs/smb/smbdirect/Kconfig
@@ -0,0 +1,9 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# smbdirect configuration
+
+config SMBDIRECT
+	def_tristate n
+	depends on INFINIBAND && INFINIBAND_ADDR_TRANS
+	depends on m || INFINIBAND=y
+	select SG_POOL
diff --git a/fs/smb/smbdirect/Makefile b/fs/smb/smbdirect/Makefile
new file mode 100644
index 000000000000..80d6025984d1
--- /dev/null
+++ b/fs/smb/smbdirect/Makefile
@@ -0,0 +1,18 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Makefile for smbdirect support
+#
+
+obj-$(CONFIG_SMBDIRECT) += smbdirect.o
+
+smbdirect-y := \
+	socket.o	\
+	connection.o	\
+	mr.o		\
+	rw.o		\
+	debug.o		\
+	connect.o	\
+	listen.o	\
+	accept.o	\
+	devices.o	\
+	main.o
diff --git a/fs/smb/smbdirect/accept.c b/fs/smb/smbdirect/accept.c
new file mode 100644
index 000000000000..1b30ca8476c1
--- /dev/null
+++ b/fs/smb/smbdirect/accept.c
@@ -0,0 +1,868 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (C) 2018, LG Electronics.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+#include <net/sock.h>
+#include "../common/smb2status.h"
+
+static int smbdirect_accept_rdma_event_handler(struct rdma_cm_id *id,
+					       struct rdma_cm_event *event);
+static int smbdirect_accept_init_params(struct smbdirect_socket *sc);
+static void smbdirect_accept_negotiate_recv_done(struct ib_cq *cq, struct ib_wc *wc);
+static void smbdirect_accept_negotiate_send_done(struct ib_cq *cq, struct ib_wc *wc);
+
+int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
+				     const struct rdma_conn_param *param)
+{
+	struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_recv_io *recv_io;
+	u8 peer_initiator_depth;
+	u8 peer_responder_resources;
+	struct rdma_conn_param conn_param;
+	__be32 ird_ord_hdr[2];
+	int ret;
+
+	if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_CREATED))
+		return -EINVAL;
+
+	/*
+	 * First set what the we as server are able to support
+	 */
+	sp->initiator_depth = min3(sp->initiator_depth,
+				   sc->ib.dev->attrs.max_qp_rd_atom,
+				   U8_MAX);
+
+	peer_initiator_depth = param->initiator_depth;
+	peer_responder_resources = param->responder_resources;
+	smbdirect_connection_negotiate_rdma_resources(sc,
+						      peer_initiator_depth,
+						      peer_responder_resources,
+						      param);
+
+	ret = smbdirect_accept_init_params(sc);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_accept_init_params() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto init_params_failed;
+	}
+
+	ret = smbdirect_connection_create_qp(sc);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_create_qp() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto create_qp_failed;
+	}
+
+	ret = smbdirect_connection_create_mem_pools(sc);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_create_mem_pools() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto create_mem_failed;
+	}
+
+	recv_io = smbdirect_connection_get_recv_io(sc);
+	if (WARN_ON_ONCE(!recv_io)) {
+		ret = -EINVAL;
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_get_recv_io() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto get_recv_io_failed;
+	}
+	recv_io->cqe.done = smbdirect_accept_negotiate_recv_done;
+
+	/*
+	 * Now post the recv_io buffer in order to get
+	 * the negotiate request
+	 */
+	sc->recv_io.expected = SMBDIRECT_EXPECT_NEGOTIATE_REQ;
+	ret = smbdirect_connection_post_recv_io(recv_io);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_post_recv_io() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto post_recv_io_failed;
+	}
+	/*
+	 * From here recv_io is known to the RDMA QP and needs ib_drain_qp and
+	 * smbdirect_accept_negotiate_recv_done to cleanup...
+	 */
+	recv_io = NULL;
+
+	/* already checked with SMBDIRECT_CHECK_STATUS_WARN above */
+	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED);
+	sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED;
+
+	/*
+	 * We already negotiated sp->initiator_depth
+	 * and sp->responder_resources above.
+	 */
+	memset(&conn_param, 0, sizeof(conn_param));
+	conn_param.initiator_depth = sp->initiator_depth;
+	conn_param.responder_resources = sp->responder_resources;
+
+	if (sc->rdma.legacy_iwarp) {
+		ird_ord_hdr[0] = cpu_to_be32(conn_param.responder_resources);
+		ird_ord_hdr[1] = cpu_to_be32(conn_param.initiator_depth);
+		conn_param.private_data = ird_ord_hdr;
+		conn_param.private_data_len = sizeof(ird_ord_hdr);
+	} else {
+		conn_param.private_data = NULL;
+		conn_param.private_data_len = 0;
+	}
+	conn_param.retry_count = SMBDIRECT_RDMA_CM_RETRY;
+	conn_param.rnr_retry_count = SMBDIRECT_RDMA_CM_RNR_RETRY;
+	conn_param.flow_control = 0;
+
+	/* explicitly set above */
+	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED);
+	sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING;
+	sc->rdma.expected_event = RDMA_CM_EVENT_ESTABLISHED;
+	sc->rdma.cm_id->event_handler = smbdirect_accept_rdma_event_handler;
+	ret = rdma_accept(sc->rdma.cm_id, &conn_param);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"rdma_accept() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto rdma_accept_failed;
+	}
+
+	/*
+	 * start with the negotiate timeout and SMBDIRECT_KEEPALIVE_PENDING
+	 * so that the timer will cause a disconnect.
+	 */
+	INIT_DELAYED_WORK(&sc->idle.timer_work, smbdirect_connection_idle_timer_work);
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->negotiate_timeout_msec));
+
+	return 0;
+
+rdma_accept_failed:
+	/*
+	 * The recv_io posted above is now owned by the QP (recv_io was set to
+	 * NULL after a successful post).  smbdirect_connection_destroy_qp()
+	 * calls ib_drain_qp(), whose completion
+	 * (smbdirect_accept_negotiate_recv_done) returns the recv_io to the
+	 * free list via smbdirect_connection_put_recv_io().  It therefore MUST
+	 * run BEFORE smbdirect_connection_destroy_mem_pools(): otherwise the
+	 * posted recv_io is still outstanding when kmem_cache_destroy() runs
+	 * ("Slab cache still has objects") and is later freed into an
+	 * already-destroyed mempool (mempool_free_bulk NULL-ptr-deref).
+	 */
+	smbdirect_connection_destroy_qp(sc);
+	smbdirect_connection_destroy_mem_pools(sc);
+	return ret;
+post_recv_io_failed:
+	/* post failed: recv_io was not accepted by the QP, still in hand */
+	if (recv_io)
+		smbdirect_connection_put_recv_io(recv_io);
+get_recv_io_failed:
+	smbdirect_connection_destroy_mem_pools(sc);
+create_mem_failed:
+	smbdirect_connection_destroy_qp(sc);
+create_qp_failed:
+init_params_failed:
+	return ret;
+}
+
+static int smbdirect_accept_init_params(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	int max_send_sges;
+	unsigned int maxpages;
+
+	/* need 3 more sge. because a SMB_DIRECT header, SMB2 header,
+	 * SMB2 response could be mapped.
+	 */
+	max_send_sges = DIV_ROUND_UP(sp->max_send_size, PAGE_SIZE) + 3;
+	if (max_send_sges > SMBDIRECT_SEND_IO_MAX_SGE) {
+		pr_err("max_send_size %d is too large\n", sp->max_send_size);
+		return -EINVAL;
+	}
+
+	/*
+	 * There is only a single batch credit
+	 */
+	atomic_set(&sc->send_io.bcredits.count, 1);
+
+	/*
+	 * Initialize the local credits to post
+	 * IB_WR_SEND[_WITH_INV].
+	 */
+	atomic_set(&sc->send_io.lcredits.count, sp->send_credit_target);
+
+	if (sp->max_read_write_size) {
+		maxpages = DIV_ROUND_UP(sp->max_read_write_size, PAGE_SIZE);
+		sc->rw_io.credits.max = rdma_rw_mr_factor(sc->ib.dev,
+							  sc->rdma.cm_id->port_num,
+							  maxpages);
+		sc->rw_io.credits.num_pages = DIV_ROUND_UP(maxpages, sc->rw_io.credits.max);
+		/* add one extra in order to handle unaligned pages */
+		sc->rw_io.credits.max += 1;
+	}
+
+	sc->recv_io.credits.target = 1;
+
+	atomic_set(&sc->rw_io.credits.count, sc->rw_io.credits.max);
+
+	return 0;
+}
+
+static void smbdirect_accept_negotiate_recv_work(struct work_struct *work);
+
+static void smbdirect_accept_negotiate_recv_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_recv_io *recv_io =
+		container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
+	struct smbdirect_socket *sc = recv_io->socket;
+	unsigned long flags;
+
+	if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_RECV))) {
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
+				"wc->status=%s (%d) wc->opcode=%d\n",
+				ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		goto error;
+	}
+
+	smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_INFO,
+		"smbdirect_recv_io completed. status='%s (%d)', opcode=%d\n",
+		ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+
+	/*
+	 * This is an internal error!
+	 */
+	if (WARN_ON_ONCE(sc->recv_io.expected != SMBDIRECT_EXPECT_NEGOTIATE_REQ))
+		goto error;
+
+	/*
+	 * Don't reset timer to the keepalive interval in
+	 * this will be done in smbdirect_accept_direct_negotiate_recv_work.
+	 */
+
+	ib_dma_sync_single_for_cpu(sc->ib.dev,
+				   recv_io->sge.addr,
+				   recv_io->sge.length,
+				   DMA_FROM_DEVICE);
+
+	/*
+	 * Only remember recv_io if it has enough bytes,
+	 * this gives smbdirect_accept_negotiate_recv_work enough
+	 * information in order to disconnect if it was not
+	 * valid.
+	 */
+	sc->recv_io.reassembly.full_packet_received = true;
+	if (wc->byte_len >= sizeof(struct smbdirect_negotiate_req))
+		smbdirect_connection_reassembly_append_recv_io(sc, recv_io, 0);
+	else
+		smbdirect_connection_put_recv_io(recv_io);
+
+	/*
+	 * Some drivers (at least mlx5_ib and irdma) might post a
+	 * recv completion before RDMA_CM_EVENT_ESTABLISHED,
+	 * we need to adjust our expectation in that case.
+	 *
+	 * So we defer further processing of the negotiation
+	 * to smbdirect_accept_negotiate_recv_work().
+	 *
+	 * If we are already in SMBDIRECT_SOCKET_NEGOTIATE_NEEDED
+	 * we queue the work directly otherwise
+	 * smbdirect_accept_rdma_event_handler() will do it, when
+	 * RDMA_CM_EVENT_ESTABLISHED arrived.
+	 */
+	spin_lock_irqsave(&sc->connect.lock, flags);
+	if (!sc->first_error) {
+		INIT_WORK(&sc->connect.work, smbdirect_accept_negotiate_recv_work);
+		if (sc->status == SMBDIRECT_SOCKET_NEGOTIATE_NEEDED)
+			queue_work(sc->workqueues.accept, &sc->connect.work);
+	}
+	spin_unlock_irqrestore(&sc->connect.lock, flags);
+
+	return;
+
+error:
+	/*
+	 * recv_io.posted.refill_work is still disabled,
+	 * so smbdirect_connection_put_recv_io() won't
+	 * start it.
+	 */
+	smbdirect_connection_put_recv_io(recv_io);
+	smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+}
+
+static void smbdirect_accept_negotiate_recv_work(struct work_struct *work)
+{
+	struct smbdirect_socket *sc =
+		container_of(work, struct smbdirect_socket, connect.work);
+	struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_recv_io *recv_io;
+	struct smbdirect_negotiate_req *nreq;
+	unsigned long flags;
+	u16 min_version;
+	u16 max_version;
+	u16 credits_requested;
+	u32 preferred_send_size;
+	u32 max_receive_size;
+	u32 max_fragmented_size;
+	u32 ntstatus;
+
+	if (sc->first_error)
+		return;
+
+	/*
+	 * make sure we won't start again...
+	 */
+	disable_work(work);
+
+	/*
+	 * Reset timer to the keepalive interval in
+	 * order to trigger our next keepalive message.
+	 */
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->keepalive_interval_msec));
+
+	/*
+	 * If smbdirect_accept_negotiate_recv_done() detected an
+	 * invalid request we want to disconnect.
+	 */
+	recv_io = smbdirect_connection_reassembly_first_recv_io(sc);
+	if (!recv_io) {
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
+	sc->recv_io.reassembly.queue_length--;
+	list_del(&recv_io->list);
+	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
+	smbdirect_connection_put_recv_io(recv_io);
+
+	if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_NEGOTIATE_NEEDED))
+		return;
+	sc->status = SMBDIRECT_SOCKET_NEGOTIATE_RUNNING;
+
+	/*
+	 * Note recv_io is already part of the free list,
+	 * as we just called smbdirect_connection_put_recv_io(),
+	 * but it won't be reused before we call
+	 * smbdirect_connection_recv_io_refill() below.
+	 */
+
+	nreq = (struct smbdirect_negotiate_req *)recv_io->packet;
+	min_version = le16_to_cpu(nreq->min_version);
+	max_version = le16_to_cpu(nreq->max_version);
+	credits_requested = le16_to_cpu(nreq->credits_requested);
+	preferred_send_size = le32_to_cpu(nreq->preferred_send_size);
+	max_receive_size = le32_to_cpu(nreq->max_receive_size);
+	max_fragmented_size = le32_to_cpu(nreq->max_fragmented_size);
+
+	smbdirect_log_negotiate(sc, SMBDIRECT_LOG_INFO,
+		"ReqIn: %s%x, %s%x, %s%u, %s%u, %s%u, %s%u\n",
+		"MinVersion=0x",
+		le16_to_cpu(nreq->min_version),
+		"MaxVersion=0x",
+		le16_to_cpu(nreq->max_version),
+		"CreditsRequested=",
+		le16_to_cpu(nreq->credits_requested),
+		"PreferredSendSize=",
+		le32_to_cpu(nreq->preferred_send_size),
+		"MaxRecvSize=",
+		le32_to_cpu(nreq->max_receive_size),
+		"MaxFragmentedSize=",
+		le32_to_cpu(nreq->max_fragmented_size));
+
+	if (!(min_version <= SMBDIRECT_V1 && max_version >= SMBDIRECT_V1)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: min_version=0x%x max_version=0x%x\n",
+			min_version, max_version);
+		ntstatus = le32_to_cpu(STATUS_NOT_SUPPORTED);
+		goto not_supported;
+	}
+
+	if (credits_requested == 0) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: credits_requested == 0\n");
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	if (max_receive_size < SMBDIRECT_MIN_RECEIVE_SIZE) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: max_receive_size=%u < %u\n",
+			max_receive_size,
+			SMBDIRECT_MIN_RECEIVE_SIZE);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	if (max_fragmented_size < SMBDIRECT_MIN_FRAGMENTED_SIZE) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: max_fragmented_size=%u < %u\n",
+			max_fragmented_size,
+			SMBDIRECT_MIN_FRAGMENTED_SIZE);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	/*
+	 * At least the value of SMBDIRECT_MIN_RECEIVE_SIZE is used.
+	 */
+	sp->max_recv_size = min_t(u32, sp->max_recv_size, preferred_send_size);
+	sp->max_recv_size = max_t(u32, sp->max_recv_size, SMBDIRECT_MIN_RECEIVE_SIZE);
+
+	/*
+	 * The maximum fragmented upper-layer payload receive size supported
+	 *
+	 * Assume max_payload_per_credit is
+	 * smb_direct_receive_credit_max - 24 = 1340
+	 *
+	 * The maximum number would be
+	 * smb_direct_receive_credit_max * max_payload_per_credit
+	 *
+	 *                       1340 * 255 = 341700 (0x536C4)
+	 *
+	 * The minimum value from the spec is 131072 (0x20000)
+	 *
+	 * For now we use the logic we used in ksmbd before:
+	 *                 (1364 * 255) / 2 = 173910 (0x2A756)
+	 *
+	 * We need to adjust this here in case the peer
+	 * lowered sp->max_recv_size.
+	 *
+	 * TODO: instead of adjusting max_fragmented_recv_size
+	 * we should adjust the number of available buffers,
+	 * but for now we keep the logic as it was used
+	 * in ksmbd before.
+	 */
+	sp->max_fragmented_recv_size = (sp->recv_credit_max * sp->max_recv_size) / 2;
+
+	/*
+	 * We take the value from the peer, which is checked to be higher than 0,
+	 * but we limit it to the max value we support in order to have
+	 * the main logic simpler.
+	 */
+	sc->recv_io.credits.target = credits_requested;
+	sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target,
+					   sp->recv_credit_max);
+
+	/*
+	 * Note nreq->max_receive_size was already checked against
+	 * SMBDIRECT_MIN_RECEIVE_SIZE above.
+	 */
+	sp->max_send_size = min_t(u32, sp->max_send_size, max_receive_size);
+
+	/*
+	 * Note nreq->max_fragmented_size was already checked against
+	 * SMBDIRECT_MIN_FRAGMENTED_SIZE above.
+	 */
+	sp->max_fragmented_send_size = max_fragmented_size;
+
+	if (sc->accept.listener) {
+		struct smbdirect_socket *lsc = sc->accept.listener;
+		unsigned long flags;
+
+		spin_lock_irqsave(&lsc->listen.lock, flags);
+		list_del(&sc->accept.list);
+		list_add_tail(&sc->accept.list, &lsc->listen.ready);
+		wake_up(&lsc->listen.wait_queue);
+		spin_unlock_irqrestore(&lsc->listen.lock, flags);
+
+		/*
+		 * smbdirect_socket_accept() will call
+		 * smbdirect_accept_negotiate_finish(nsc, 0);
+		 *
+		 * So that we don't send the negotiation
+		 * response that grants credits to the peer
+		 * before the socket is accepted by the
+		 * application.
+		 */
+		return;
+	}
+
+	ntstatus = le32_to_cpu(STATUS_SUCCESS);
+
+not_supported:
+	smbdirect_accept_negotiate_finish(sc, ntstatus);
+}
+
+void smbdirect_accept_negotiate_finish(struct smbdirect_socket *sc, u32 ntstatus)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_recv_io *recv_io;
+	struct smbdirect_send_io *send_io;
+	struct smbdirect_negotiate_resp *nrep;
+	int posted;
+	u16 new_credits;
+	int ret;
+
+	if (ntstatus)
+		goto not_supported;
+
+	/*
+	 * Prepare for receiving data_transfer messages
+	 */
+	sc->recv_io.reassembly.full_packet_received = true;
+	sc->recv_io.expected = SMBDIRECT_EXPECT_DATA_TRANSFER;
+	list_for_each_entry(recv_io, &sc->recv_io.free.list, list)
+		recv_io->cqe.done = smbdirect_connection_recv_io_done;
+	recv_io = NULL;
+
+	/*
+	 * We should at least post 1 smbdirect_recv_io!
+	 */
+	posted = smbdirect_connection_recv_io_refill(sc);
+	if (posted < 1) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_recv_io_refill() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(posted));
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	/*
+	 * The response will grant credits for all posted
+	 * smbdirect_recv_io messages.
+	 */
+	new_credits = smbdirect_connection_grant_recv_credits(sc);
+
+not_supported:
+	send_io = smbdirect_connection_alloc_send_io(sc);
+	if (IS_ERR(send_io)) {
+		ret = PTR_ERR(send_io);
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_alloc_send_io() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		smbdirect_socket_schedule_cleanup(sc, ret);
+		return;
+	}
+	send_io->cqe.done = smbdirect_accept_negotiate_send_done;
+
+	nrep = (struct smbdirect_negotiate_resp *)send_io->packet;
+	nrep->min_version = cpu_to_le16(SMBDIRECT_V1);
+	nrep->max_version = cpu_to_le16(SMBDIRECT_V1);
+	if (ntstatus == 0) {
+		nrep->negotiated_version = cpu_to_le16(SMBDIRECT_V1);
+		nrep->reserved = 0;
+		nrep->credits_requested = cpu_to_le16(sp->send_credit_target);
+		nrep->credits_granted = cpu_to_le16(new_credits);
+		nrep->status = cpu_to_le32(ntstatus);
+		nrep->max_readwrite_size = cpu_to_le32(sp->max_read_write_size);
+		nrep->preferred_send_size = cpu_to_le32(sp->max_send_size);
+		nrep->max_receive_size = cpu_to_le32(sp->max_recv_size);
+		nrep->max_fragmented_size = cpu_to_le32(sp->max_fragmented_recv_size);
+	} else {
+		nrep->negotiated_version = 0;
+		nrep->reserved = 0;
+		nrep->credits_requested = 0;
+		nrep->credits_granted = 0;
+		nrep->status = cpu_to_le32(ntstatus);
+		nrep->max_readwrite_size = 0;
+		nrep->preferred_send_size = 0;
+		nrep->max_receive_size = 0;
+		nrep->max_fragmented_size = 0;
+	}
+
+	smbdirect_log_negotiate(sc, SMBDIRECT_LOG_INFO,
+		"RepOut: %s%x, %s%x, %s%x, %s%u, %s%u, %s%x, %s%u, %s%u, %s%u, %s%u\n",
+		"MinVersion=0x",
+		le16_to_cpu(nrep->min_version),
+		"MaxVersion=0x",
+		le16_to_cpu(nrep->max_version),
+		"NegotiatedVersion=0x",
+		le16_to_cpu(nrep->negotiated_version),
+		"CreditsRequested=",
+		le16_to_cpu(nrep->credits_requested),
+		"CreditsGranted=",
+		le16_to_cpu(nrep->credits_granted),
+		"Status=0x",
+		le32_to_cpu(nrep->status),
+		"MaxReadWriteSize=",
+		le32_to_cpu(nrep->max_readwrite_size),
+		"PreferredSendSize=",
+		le32_to_cpu(nrep->preferred_send_size),
+		"MaxRecvSize=",
+		le32_to_cpu(nrep->max_receive_size),
+		"MaxFragmentedSize=",
+		le32_to_cpu(nrep->max_fragmented_size));
+
+	send_io->sge[0].addr = ib_dma_map_single(sc->ib.dev,
+						 nrep,
+						 sizeof(*nrep),
+						 DMA_TO_DEVICE);
+	ret = ib_dma_mapping_error(sc->ib.dev, send_io->sge[0].addr);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"ib_dma_mapping_error() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		smbdirect_connection_free_send_io(send_io);
+		smbdirect_socket_schedule_cleanup(sc, ret);
+		return;
+	}
+
+	send_io->sge[0].length = sizeof(*nrep);
+	send_io->sge[0].lkey = sc->ib.pd->local_dma_lkey;
+	send_io->num_sge = 1;
+
+	ib_dma_sync_single_for_device(sc->ib.dev,
+				      send_io->sge[0].addr,
+				      send_io->sge[0].length,
+				      DMA_TO_DEVICE);
+
+	send_io->wr.next = NULL;
+	send_io->wr.wr_cqe = &send_io->cqe;
+	send_io->wr.sg_list = send_io->sge;
+	send_io->wr.num_sge = send_io->num_sge;
+	send_io->wr.opcode = IB_WR_SEND;
+	send_io->wr.send_flags = IB_SEND_SIGNALED;
+
+	ret = smbdirect_connection_post_send_wr(sc, &send_io->wr);
+	if (ret) {
+		/* if we reach here, post send failed */
+		smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_post_send_wr() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		/*
+		 * Note smbdirect_connection_free_send_io()
+		 * does ib_dma_unmap_page()
+		 */
+		smbdirect_connection_free_send_io(send_io);
+		smbdirect_socket_schedule_cleanup(sc, ret);
+		return;
+	}
+
+	/*
+	 * smbdirect_accept_negotiate_send_done
+	 * will do all remaining work...
+	 */
+}
+
+static void smbdirect_accept_negotiate_send_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_send_io *send_io =
+		container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
+	struct smbdirect_socket *sc = send_io->socket;
+	struct smbdirect_negotiate_resp *nrep;
+	u32 ntstatus;
+
+	smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_INFO,
+		"smbdirect_send_io completed. status='%s (%d)', opcode=%d\n",
+		ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+
+	nrep = (struct smbdirect_negotiate_resp *)send_io->packet;
+	ntstatus = le32_to_cpu(nrep->status);
+
+	/* Note this frees wc->wr_cqe, but not wc */
+	smbdirect_connection_free_send_io(send_io);
+	atomic_dec(&sc->send_io.pending.count);
+
+	if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_SEND))) {
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+				"wc->status=%s (%d) wc->opcode=%d\n",
+				ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	/*
+	 * If we send a smbdirect_negotiate_resp without NT_STATUS_OK (0)
+	 * we need to disconnect now.
+	 *
+	 * Otherwise smbdirect_connection_negotiation_done()
+	 * will setup all required things and wake up
+	 * the waiter.
+	 */
+	if (ntstatus)
+		smbdirect_socket_schedule_cleanup(sc, -EOPNOTSUPP);
+	else
+		smbdirect_connection_negotiation_done(sc);
+}
+
+static int smbdirect_accept_rdma_event_handler(struct rdma_cm_id *id,
+					       struct rdma_cm_event *event)
+{
+	struct smbdirect_socket *sc = id->context;
+	unsigned long flags;
+
+	/*
+	 * cma_cm_event_handler() has
+	 * lockdep_assert_held(&id_priv->handler_mutex);
+	 *
+	 * Mutexes are not allowed in interrupts,
+	 * and we rely on not being in an interrupt here,
+	 * as we might sleep.
+	 *
+	 * We didn't timeout so we cancel our idle timer,
+	 * it will be scheduled again if needed.
+	 */
+	WARN_ON_ONCE(in_interrupt());
+
+	if (event->status || event->event != sc->rdma.expected_event) {
+		int ret = -ECONNABORTED;
+
+		if (event->event == RDMA_CM_EVENT_REJECTED)
+			ret = -ECONNREFUSED;
+		if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+			ret = -ENETDOWN;
+		if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
+			ret = event->status;
+
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"%s (first_error=%1pe, expected=%s) => event=%s status=%d => ret=%1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			rdma_event_msg(sc->rdma.expected_event),
+			rdma_event_msg(event->event),
+			event->status,
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+
+		smbdirect_socket_schedule_cleanup(sc, ret);
+		return 0;
+	}
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"%s (first_error=%1pe) event=%s\n",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+		rdma_event_msg(event->event));
+
+	if (sc->first_error)
+		return 0;
+
+	switch (event->event) {
+	case RDMA_CM_EVENT_ESTABLISHED:
+		smbdirect_connection_rdma_established(sc);
+
+		/*
+		 * Some drivers (at least mlx5_ib and irdma) might post a
+		 * recv completion before RDMA_CM_EVENT_ESTABLISHED,
+		 * we need to adjust our expectation in that case.
+		 *
+		 * If smbdirect_accept_negotiate_recv_done was called first
+		 * it initialized sc->connect.work only for us to
+		 * start, so that we turned into
+		 * SMBDIRECT_SOCKET_NEGOTIATE_NEEDED, before
+		 * smbdirect_accept_negotiate_recv_work() runs.
+		 *
+		 * If smbdirect_accept_negotiate_recv_done didn't happen
+		 * yet. sc->connect.work is still be disabled and
+		 * queue_work() is a no-op.
+		 */
+		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING))
+			return 0;
+		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_NEEDED;
+		spin_lock_irqsave(&sc->connect.lock, flags);
+		if (!sc->first_error)
+			queue_work(sc->workqueues.accept, &sc->connect.work);
+		spin_unlock_irqrestore(&sc->connect.lock, flags);
+
+		/*
+		 * wait for smbdirect_accept_negotiate_recv_done()
+		 * to get the negotiate request.
+		 */
+		return 0;
+
+	default:
+		break;
+	}
+
+	/*
+	 * This is an internal error
+	 */
+	WARN_ON_ONCE(sc->rdma.expected_event != RDMA_CM_EVENT_ESTABLISHED);
+	smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+	return 0;
+}
+
+static long smbdirect_socket_wait_for_accept(struct smbdirect_socket *lsc, long timeo)
+{
+	long ret;
+
+	ret = wait_event_interruptible_timeout(lsc->listen.wait_queue,
+					       !list_empty_careful(&lsc->listen.ready) ||
+					       lsc->status != SMBDIRECT_SOCKET_LISTENING ||
+					       lsc->first_error,
+					       timeo);
+	if (lsc->status != SMBDIRECT_SOCKET_LISTENING)
+		return -EINVAL;
+	if (lsc->first_error)
+		return lsc->first_error;
+	if (!ret)
+		ret = -ETIMEDOUT;
+	if (ret < 0)
+		return ret;
+
+	return 0;
+}
+
+struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
+						 long timeo,
+						 struct proto_accept_arg *arg)
+{
+	struct smbdirect_socket *nsc;
+	unsigned long flags;
+
+	if (lsc->status != SMBDIRECT_SOCKET_LISTENING) {
+		arg->err = -EINVAL;
+		return NULL;
+	}
+
+	if (lsc->first_error) {
+		arg->err = lsc->first_error;
+		return NULL;
+	}
+
+	if (list_empty_careful(&lsc->listen.ready)) {
+		int ret;
+
+		if (timeo == 0) {
+			arg->err = -EAGAIN;
+			return NULL;
+		}
+
+		ret = smbdirect_socket_wait_for_accept(lsc, timeo);
+		if (ret) {
+			arg->err = ret;
+			return NULL;
+		}
+	}
+
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	nsc = list_first_entry_or_null(&lsc->listen.ready,
+				       struct smbdirect_socket,
+				       accept.list);
+	if (nsc) {
+		nsc->accept.listener = NULL;
+		list_del_init_careful(&nsc->accept.list);
+		arg->is_empty = list_empty_careful(&lsc->listen.ready);
+	}
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+	if (!nsc) {
+		arg->err = -EAGAIN;
+		return NULL;
+	}
+
+	/*
+	 * We did not send the negotiation response
+	 * yet, so we did not grant any credits to the client,
+	 * so it didn't grant any credits to us.
+	 *
+	 * The caller expects a connected socket
+	 * now as there are no credits anyway.
+	 *
+	 * Then we send the negotiation response in
+	 * order to grant credits to the peer.
+	 */
+	nsc->status = SMBDIRECT_SOCKET_CONNECTED;
+	smbdirect_accept_negotiate_finish(nsc, 0);
+
+	return nsc;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_accept);
diff --git a/fs/smb/smbdirect/connect.c b/fs/smb/smbdirect/connect.c
new file mode 100644
index 000000000000..34a3e72c38fb
--- /dev/null
+++ b/fs/smb/smbdirect/connect.c
@@ -0,0 +1,926 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (c) 2012,2016,2017,2025 Stefan Metzmacher
+ */
+
+#include "internal.h"
+#include "../common/smb2status.h"
+
+static int smbdirect_connect_setup_connection(struct smbdirect_socket *sc);
+static int smbdirect_connect_resolve_addr(struct smbdirect_socket *sc,
+					  const struct sockaddr *src,
+					  const struct sockaddr *dst);
+static int smbdirect_connect_rdma_event_handler(struct rdma_cm_id *id,
+						struct rdma_cm_event *event);
+static int smbdirect_connect_negotiate_start(struct smbdirect_socket *sc);
+static void smbdirect_connect_negotiate_send_done(struct ib_cq *cq, struct ib_wc *wc);
+static void smbdirect_connect_negotiate_recv_done(struct ib_cq *cq, struct ib_wc *wc);
+
+int smbdirect_connect(struct smbdirect_socket *sc, const struct sockaddr *dst)
+{
+	const struct sockaddr *src = NULL;
+	union {
+		struct sockaddr sa;
+		struct sockaddr_storage ss;
+	} src_addr = {
+		.sa = {
+			.sa_family = AF_UNSPEC,
+		},
+	};
+	int ret;
+
+	if (sc->first_error)
+		return -ENOTCONN;
+
+	if (sc->status != SMBDIRECT_SOCKET_CREATED)
+		return -EALREADY;
+
+	if (WARN_ON_ONCE(!sc->rdma.cm_id))
+		return -EINVAL;
+
+	src_addr.ss = sc->rdma.cm_id->route.addr.src_addr;
+	if (src_addr.sa.sa_family != AF_UNSPEC)
+		src = &src_addr.sa;
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"connect: src: %pISpsfc dst: %pISpsfc\n",
+		src, dst);
+
+	ret = smbdirect_connect_setup_connection(sc);
+	if (ret)
+		return ret;
+
+	ret = smbdirect_connect_resolve_addr(sc, src, dst);
+	if (ret)
+		return ret;
+
+	/*
+	 * The rest happens async via smbdirect_connect_rdma_event_handler()
+	 * the caller will decide to wait or not.
+	 */
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connect);
+
+static int smbdirect_connect_setup_connection(struct smbdirect_socket *sc)
+{
+	rdma_lock_handler(sc->rdma.cm_id);
+	sc->rdma.cm_id->event_handler = smbdirect_connect_rdma_event_handler;
+	rdma_unlock_handler(sc->rdma.cm_id);
+
+	if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_CREATED))
+		return -EINVAL;
+	sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED;
+
+	return 0;
+}
+
+static int smbdirect_connect_resolve_addr(struct smbdirect_socket *sc,
+					  const struct sockaddr *src,
+					  const struct sockaddr *dst)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct sockaddr *src_addr = NULL;
+	struct sockaddr *dst_addr = NULL;
+	int ret;
+
+	src_addr = (struct sockaddr *)src;
+	if (src_addr && src_addr->sa_family == AF_UNSPEC)
+		src_addr = NULL;
+	dst_addr = (struct sockaddr *)dst;
+
+	if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED))
+		return -EINVAL;
+	sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING;
+	sc->rdma.expected_event = RDMA_CM_EVENT_ADDR_RESOLVED;
+	ret = rdma_resolve_addr(sc->rdma.cm_id, src_addr, dst_addr,
+				sp->resolve_addr_timeout_msec);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"rdma_resolve_addr() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	return 0;
+}
+
+static int smbdirect_connect_resolve_route(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	int ret;
+
+	if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED))
+		return sc->first_error;
+	sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING;
+	sc->rdma.expected_event = RDMA_CM_EVENT_ROUTE_RESOLVED;
+	ret = rdma_resolve_route(sc->rdma.cm_id, sp->resolve_route_timeout_msec);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"rdma_resolve_route() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	return 0;
+}
+
+static int smbdirect_connect_rdma_connect(struct smbdirect_socket *sc)
+{
+	struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct rdma_conn_param conn_param;
+	__be32 ird_ord_hdr[2];
+	int ret;
+
+	sc->ib.dev = sc->rdma.cm_id->device;
+
+	if (!smbdirect_frwr_is_supported(&sc->ib.dev->attrs)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"Fast Registration Work Requests (FRWR) is not supported device %.*s\n",
+			IB_DEVICE_NAME_MAX,
+			sc->ib.dev->name);
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"Device capability flags = %llx max_fast_reg_page_list_len = %u\n",
+			sc->ib.dev->attrs.device_cap_flags,
+			sc->ib.dev->attrs.max_fast_reg_page_list_len);
+		return -EPROTONOSUPPORT;
+	}
+
+	if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IB &&
+	    !rdma_ib_or_roce(sc->ib.dev, sc->rdma.cm_id->port_num)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"Not IB: device: %.*s IW:%u local: %pISpsfc remote: %pISpsfc\n",
+			IB_DEVICE_NAME_MAX,
+			sc->ib.dev->name,
+			rdma_protocol_iwarp(sc->ib.dev, sc->rdma.cm_id->port_num),
+			&sc->rdma.cm_id->route.addr.src_addr,
+			&sc->rdma.cm_id->route.addr.dst_addr);
+		return -EPROTONOSUPPORT;
+	}
+	if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IW &&
+	    !rdma_protocol_iwarp(sc->ib.dev, sc->rdma.cm_id->port_num)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"Not IW: device: %.*s IB:%u local: %pISpsfc remote: %pISpsfc\n",
+			IB_DEVICE_NAME_MAX,
+			sc->ib.dev->name,
+			rdma_ib_or_roce(sc->ib.dev, sc->rdma.cm_id->port_num),
+			&sc->rdma.cm_id->route.addr.src_addr,
+			&sc->rdma.cm_id->route.addr.dst_addr);
+		return -EPROTONOSUPPORT;
+	}
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"rdma connect: device: %.*s local: %pISpsfc remote: %pISpsfc\n",
+		IB_DEVICE_NAME_MAX,
+		sc->ib.dev->name,
+		&sc->rdma.cm_id->route.addr.src_addr,
+		&sc->rdma.cm_id->route.addr.dst_addr);
+
+	sp->max_frmr_depth = min_t(u32, sp->max_frmr_depth,
+				   sc->ib.dev->attrs.max_fast_reg_page_list_len);
+	sc->mr_io.type = IB_MR_TYPE_MEM_REG;
+	if (sc->ib.dev->attrs.kernel_cap_flags & IBK_SG_GAPS_REG)
+		sc->mr_io.type = IB_MR_TYPE_SG_GAPS;
+
+	sp->responder_resources = min3(sp->responder_resources,
+				       sc->ib.dev->attrs.max_qp_rd_atom,
+				       U8_MAX);
+	smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_INFO,
+		"responder_resources=%d\n",
+		sp->responder_resources);
+
+	ret = smbdirect_connection_create_qp(sc);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_create_qp() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	memset(&conn_param, 0, sizeof(conn_param));
+	conn_param.initiator_depth = sp->initiator_depth;
+	conn_param.responder_resources = sp->responder_resources;
+
+	/* Need to send IRD/ORD in private data for iWARP */
+	if (rdma_protocol_iwarp(sc->ib.dev, sc->rdma.cm_id->port_num)) {
+		ird_ord_hdr[0] = cpu_to_be32(conn_param.responder_resources);
+		ird_ord_hdr[1] = cpu_to_be32(conn_param.initiator_depth);
+		conn_param.private_data = ird_ord_hdr;
+		conn_param.private_data_len = sizeof(ird_ord_hdr);
+	} else {
+		conn_param.private_data = NULL;
+		conn_param.private_data_len = 0;
+	}
+
+	conn_param.retry_count = SMBDIRECT_RDMA_CM_RETRY;
+	conn_param.rnr_retry_count = SMBDIRECT_RDMA_CM_RNR_RETRY;
+	conn_param.flow_control = 0;
+
+	if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED))
+		return sc->first_error;
+	sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING;
+	sc->rdma.expected_event = RDMA_CM_EVENT_ESTABLISHED;
+	ret = rdma_connect_locked(sc->rdma.cm_id, &conn_param);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"rdma_connect_locked() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	/*
+	 * start with the rdma connect timeout and SMBDIRECT_KEEPALIVE_PENDING
+	 * so that the timer will cause a disconnect.
+	 */
+	INIT_DELAYED_WORK(&sc->idle.timer_work, smbdirect_connection_idle_timer_work);
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->rdma_connect_timeout_msec));
+
+	return 0;
+}
+
+static int smbdirect_connect_rdma_event_handler(struct rdma_cm_id *id,
+						struct rdma_cm_event *event)
+{
+	struct smbdirect_socket *sc = id->context;
+	u8 peer_initiator_depth;
+	u8 peer_responder_resources;
+	int ret;
+
+	/*
+	 * cma_cm_event_handler() has
+	 * lockdep_assert_held(&id_priv->handler_mutex);
+	 *
+	 * Mutexes are not allowed in interrupts,
+	 * and we rely on not being in an interrupt here,
+	 * as we might sleep.
+	 *
+	 * We didn't timeout so we cancel our idle timer,
+	 * it will be scheduled again if needed.
+	 */
+	WARN_ON_ONCE(in_interrupt());
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
+	cancel_delayed_work_sync(&sc->idle.timer_work);
+
+	if (event->status || event->event != sc->rdma.expected_event) {
+		int lvl = SMBDIRECT_LOG_ERR;
+
+		ret = -ECONNABORTED;
+
+		if (event->event == RDMA_CM_EVENT_REJECTED)
+			ret = -ECONNREFUSED;
+		if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+			ret = -ENETDOWN;
+		if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
+			ret = event->status;
+
+		if (ret == -ENODEV)
+			lvl = SMBDIRECT_LOG_INFO;
+
+		smbdirect_log_rdma_event(sc, lvl,
+			"%s (first_error=%1pe, expected=%s) => event=%s status=%d => ret=%1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			rdma_event_msg(sc->rdma.expected_event),
+			rdma_event_msg(event->event),
+			event->status,
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+
+		smbdirect_socket_schedule_cleanup_lvl(sc,
+						      lvl,
+						      ret);
+		return 0;
+	}
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"%s (first_error=%1pe) event=%s\n",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+		rdma_event_msg(event->event));
+
+	if (sc->first_error)
+		return 0;
+
+	switch (event->event) {
+	case RDMA_CM_EVENT_ADDR_RESOLVED:
+		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING))
+			return 0;
+		sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED;
+
+		ret = smbdirect_connect_resolve_route(sc);
+		if (ret)
+			smbdirect_socket_schedule_cleanup(sc, ret);
+		return 0;
+
+	case RDMA_CM_EVENT_ROUTE_RESOLVED:
+		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING))
+			return 0;
+		sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED;
+
+		ret = smbdirect_connect_rdma_connect(sc);
+		if (ret)
+			smbdirect_socket_schedule_cleanup(sc, ret);
+		return 0;
+
+	case RDMA_CM_EVENT_ESTABLISHED:
+		smbdirect_connection_rdma_established(sc);
+
+		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING))
+			return 0;
+		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_NEEDED;
+
+		/*
+		 * Here we work around an inconsistency between
+		 * iWarp and other devices (at least rxe and irdma using RoCEv2)
+		 */
+		if (rdma_protocol_iwarp(id->device, id->port_num)) {
+			/*
+			 * iWarp devices report the peer's values
+			 * with the perspective of the peer here.
+			 * Tested with siw and irdma (in iwarp mode)
+			 * We need to change to our perspective here,
+			 * so we need to switch the values.
+			 */
+			peer_initiator_depth = event->param.conn.responder_resources;
+			peer_responder_resources = event->param.conn.initiator_depth;
+		} else {
+			/*
+			 * Non iWarp devices report the peer's values
+			 * already changed to our perspective here.
+			 * Tested with rxe and irdma (in roce mode).
+			 */
+			peer_initiator_depth = event->param.conn.initiator_depth;
+			peer_responder_resources = event->param.conn.responder_resources;
+		}
+		smbdirect_connection_negotiate_rdma_resources(sc,
+							      peer_initiator_depth,
+							      peer_responder_resources,
+							      &event->param.conn);
+
+		ret = smbdirect_connect_negotiate_start(sc);
+		if (ret)
+			smbdirect_socket_schedule_cleanup(sc, ret);
+		return 0;
+
+	default:
+		break;
+	}
+
+	/*
+	 * This is an internal error
+	 */
+	WARN_ON_ONCE(sc->rdma.expected_event != RDMA_CM_EVENT_ESTABLISHED);
+	smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+	return 0;
+}
+
+static int smbdirect_connect_negotiate_start(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_recv_io *recv_io = NULL;
+	struct smbdirect_send_io *send_io = NULL;
+	struct smbdirect_negotiate_req *nreq = NULL;
+	int ret;
+
+	if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_NEGOTIATE_NEEDED))
+		return sc->first_error;
+	sc->status = SMBDIRECT_SOCKET_NEGOTIATE_RUNNING;
+
+	ret = smbdirect_connection_create_mem_pools(sc);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_create_mem_pools() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto create_mem_pools_failed;
+	}
+
+	/*
+	 * There is only a single batch credit
+	 */
+	atomic_set(&sc->send_io.bcredits.count, 1);
+
+	/*
+	 * Initialize the local credits to post
+	 * IB_WR_SEND[_WITH_INV].
+	 */
+	atomic_set(&sc->send_io.lcredits.count, sp->send_credit_target);
+
+	recv_io = smbdirect_connection_get_recv_io(sc);
+	if (WARN_ON_ONCE(!recv_io)) {
+		ret = -EINVAL;
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_get_recv_io() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto get_recv_io_failed;
+	}
+	recv_io->cqe.done = smbdirect_connect_negotiate_recv_done;
+
+	send_io = smbdirect_connection_alloc_send_io(sc);
+	if (IS_ERR(send_io)) {
+		ret = PTR_ERR(send_io);
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_alloc_send_io() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto alloc_send_io_failed;
+	}
+	send_io->cqe.done = smbdirect_connect_negotiate_send_done;
+
+	nreq = (struct smbdirect_negotiate_req *)send_io->packet;
+	nreq->min_version = cpu_to_le16(SMBDIRECT_V1);
+	nreq->max_version = cpu_to_le16(SMBDIRECT_V1);
+	nreq->reserved = 0;
+	nreq->credits_requested = cpu_to_le16(sp->send_credit_target);
+	nreq->preferred_send_size = cpu_to_le32(sp->max_send_size);
+	nreq->max_receive_size = cpu_to_le32(sp->max_recv_size);
+	nreq->max_fragmented_size = cpu_to_le32(sp->max_fragmented_recv_size);
+
+	smbdirect_log_negotiate(sc, SMBDIRECT_LOG_INFO,
+		"ReqOut: %s%x, %s%x, %s%u, %s%u, %s%u, %s%u\n",
+		"MinVersion=0x",
+		le16_to_cpu(nreq->min_version),
+		"MaxVersion=0x",
+		le16_to_cpu(nreq->max_version),
+		"CreditsRequested=",
+		le16_to_cpu(nreq->credits_requested),
+		"PreferredSendSize=",
+		le32_to_cpu(nreq->preferred_send_size),
+		"MaxRecvSize=",
+		le32_to_cpu(nreq->max_receive_size),
+		"MaxFragmentedSize=",
+		le32_to_cpu(nreq->max_fragmented_size));
+
+	send_io->sge[0].addr = ib_dma_map_single(sc->ib.dev,
+						 nreq,
+						 sizeof(*nreq),
+						 DMA_TO_DEVICE);
+	ret = ib_dma_mapping_error(sc->ib.dev, send_io->sge[0].addr);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"ib_dma_mapping_error() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto dma_mapping_failed;
+	}
+
+	send_io->sge[0].length = sizeof(*nreq);
+	send_io->sge[0].lkey = sc->ib.pd->local_dma_lkey;
+	send_io->num_sge = 1;
+
+	ib_dma_sync_single_for_device(sc->ib.dev,
+				      send_io->sge[0].addr,
+				      send_io->sge[0].length,
+				      DMA_TO_DEVICE);
+
+	smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_INFO,
+		"sge addr=0x%llx length=%u lkey=0x%x\n",
+		send_io->sge[0].addr,
+		send_io->sge[0].length,
+		send_io->sge[0].lkey);
+
+	/*
+	 * Now post the recv_io buffer in order to get
+	 * the negotiate response
+	 */
+	sc->recv_io.expected = SMBDIRECT_EXPECT_NEGOTIATE_REP;
+	ret = smbdirect_connection_post_recv_io(recv_io);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_post_recv_io() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto post_recv_io_failed;
+	}
+
+	send_io->wr.next = NULL;
+	send_io->wr.wr_cqe = &send_io->cqe;
+	send_io->wr.sg_list = send_io->sge;
+	send_io->wr.num_sge = send_io->num_sge;
+	send_io->wr.opcode = IB_WR_SEND;
+	send_io->wr.send_flags = IB_SEND_SIGNALED;
+
+	ret = smbdirect_connection_post_send_wr(sc, &send_io->wr);
+	if (ret) {
+		/* if we reach here, post send failed */
+		smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_post_send_wr() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto post_send_wr_failed;
+	}
+
+	/*
+	 * start with the negotiate timeout and SMBDIRECT_KEEPALIVE_PENDING
+	 * so that the timer will cause a disconnect.
+	 */
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->negotiate_timeout_msec));
+
+	return 0;
+
+post_send_wr_failed:
+	/*
+	 * ib_dma_unmap_single is called in
+	 * smbdirect_connection_free_send_io()
+	 */
+	smbdirect_connection_free_send_io(send_io);
+	/*
+	 * recv_io is given to the rdma layer,
+	 * we should not put it even on error
+	 * nor call smbdirect_connection_destroy_mem_pools()
+	 * it will be cleaned up during disconnect.
+	 */
+	return ret;
+
+post_recv_io_failed:
+	/*
+	 * ib_dma_unmap_single is called in
+	 * smbdirect_connection_free_send_io()
+	 */
+dma_mapping_failed:
+	smbdirect_connection_free_send_io(send_io);
+
+alloc_send_io_failed:
+	smbdirect_connection_put_recv_io(recv_io);
+
+get_recv_io_failed:
+	smbdirect_connection_destroy_mem_pools(sc);
+
+create_mem_pools_failed:
+	return ret;
+}
+
+static void smbdirect_connect_negotiate_send_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_send_io *send_io =
+		container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
+	struct smbdirect_socket *sc = send_io->socket;
+
+	smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_INFO,
+		"smbdirect_send_io completed. status='%s (%d)', opcode=%d\n",
+		ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+
+	/* Note this frees wc->wr_cqe, but not wc */
+	smbdirect_connection_free_send_io(send_io);
+	atomic_dec(&sc->send_io.pending.count);
+
+	if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_SEND))) {
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+				"wc->status=%s (%d) wc->opcode=%d\n",
+				ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+}
+
+static void smbdirect_connect_negotiate_recv_work(struct work_struct *work);
+
+static void smbdirect_connect_negotiate_recv_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_recv_io *recv_io =
+		container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
+	struct smbdirect_socket *sc = recv_io->socket;
+	unsigned long flags;
+
+	if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_RECV))) {
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
+				"wc->status=%s (%d) wc->opcode=%d\n",
+				ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		goto error;
+	}
+
+	smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_INFO,
+		"smbdirect_recv_io completed. status='%s (%d)', opcode=%d\n",
+		ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+
+	/*
+	 * This is an internal error!
+	 */
+	if (WARN_ON_ONCE(sc->recv_io.expected != SMBDIRECT_EXPECT_NEGOTIATE_REP))
+		goto error;
+
+	/*
+	 * Don't reset timer to the keepalive interval in
+	 * this will be done in smbdirect_accept_direct_negotiate_recv_work.
+	 */
+
+	ib_dma_sync_single_for_cpu(sc->ib.dev,
+				   recv_io->sge.addr,
+				   recv_io->sge.length,
+				   DMA_FROM_DEVICE);
+
+	/*
+	 * Only remember recv_io if it has enough bytes,
+	 * this gives smbdirect_accept_negotiate_recv_work enough
+	 * information in order to disconnect if it was not
+	 * valid.
+	 */
+	sc->recv_io.reassembly.full_packet_received = true;
+	if (wc->byte_len >= sizeof(struct smbdirect_negotiate_resp))
+		smbdirect_connection_reassembly_append_recv_io(sc, recv_io, 0);
+	else
+		smbdirect_connection_put_recv_io(recv_io);
+
+	/*
+	 * We continue via the workqueue as we may have
+	 * complex work that might sleep.
+	 *
+	 * So we defer further processing of the negotiation
+	 * to smbdirect_connect_negotiate_recv_work().
+	 */
+	spin_lock_irqsave(&sc->connect.lock, flags);
+	if (!sc->first_error) {
+		INIT_WORK(&sc->connect.work, smbdirect_connect_negotiate_recv_work);
+		if (sc->status == SMBDIRECT_SOCKET_NEGOTIATE_RUNNING)
+			queue_work(sc->workqueues.connect, &sc->connect.work);
+	}
+	spin_unlock_irqrestore(&sc->connect.lock, flags);
+
+	return;
+
+error:
+	/*
+	 * recv_io.posted.refill_work is still disabled,
+	 * so smbdirect_connection_put_recv_io() won't
+	 * start it.
+	 */
+	smbdirect_connection_put_recv_io(recv_io);
+	smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+}
+
+static void smbdirect_connect_negotiate_recv_work(struct work_struct *work)
+{
+	struct smbdirect_socket *sc =
+		container_of(work, struct smbdirect_socket, connect.work);
+	struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_recv_io *recv_io;
+	struct smbdirect_negotiate_resp *nrep;
+	unsigned long flags;
+	u16 negotiated_version;
+	u16 credits_requested;
+	u16 credits_granted;
+	u32 status;
+	u32 max_readwrite_size;
+	u32 preferred_send_size;
+	u32 max_receive_size;
+	u32 max_fragmented_size;
+	int posted;
+	int ret;
+
+	if (sc->first_error)
+		return;
+
+	/*
+	 * make sure we won't start again...
+	 */
+	disable_work(work);
+
+	/*
+	 * Reset timer to the keepalive interval in
+	 * order to trigger our next keepalive message.
+	 */
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->keepalive_interval_msec));
+
+	/*
+	 * If smbdirect_connect_negotiate_recv_done() detected an
+	 * invalid request we want to disconnect.
+	 */
+	recv_io = smbdirect_connection_reassembly_first_recv_io(sc);
+	if (!recv_io) {
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
+	sc->recv_io.reassembly.queue_length--;
+	list_del(&recv_io->list);
+	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
+	smbdirect_connection_put_recv_io(recv_io);
+
+	if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_NEGOTIATE_RUNNING))
+		return;
+
+	/*
+	 * Note recv_io is already part of the free list,
+	 * as we just called smbdirect_connection_put_recv_io(),
+	 * but it won't be reused before we call
+	 * smbdirect_connection_recv_io_refill() below.
+	 */
+
+	nrep = (struct smbdirect_negotiate_resp *)recv_io->packet;
+	negotiated_version = le16_to_cpu(nrep->negotiated_version);
+	credits_requested = le16_to_cpu(nrep->credits_requested);
+	credits_granted = le16_to_cpu(nrep->credits_granted);
+	status = le32_to_cpu(nrep->status);
+	max_readwrite_size = le32_to_cpu(nrep->max_readwrite_size);
+	preferred_send_size = le32_to_cpu(nrep->preferred_send_size);
+	max_receive_size = le32_to_cpu(nrep->max_receive_size);
+	max_fragmented_size = le32_to_cpu(nrep->max_fragmented_size);
+
+	smbdirect_log_negotiate(sc, SMBDIRECT_LOG_INFO,
+		"RepIn: %s%x, %s%x, %s%x, %s%u, %s%u, %s%x, %s%u, %s%u, %s%u, %s%u\n",
+		"MinVersion=0x",
+		le16_to_cpu(nrep->min_version),
+		"MaxVersion=0x",
+		le16_to_cpu(nrep->max_version),
+		"NegotiatedVersion=0x",
+		le16_to_cpu(nrep->negotiated_version),
+		"CreditsRequested=",
+		le16_to_cpu(nrep->credits_requested),
+		"CreditsGranted=",
+		le16_to_cpu(nrep->credits_granted),
+		"Status=0x",
+		le32_to_cpu(nrep->status),
+		"MaxReadWriteSize=",
+		le32_to_cpu(nrep->max_readwrite_size),
+		"PreferredSendSize=",
+		le32_to_cpu(nrep->preferred_send_size),
+		"MaxRecvSize=",
+		le32_to_cpu(nrep->max_receive_size),
+		"MaxFragmentedSize=",
+		le32_to_cpu(nrep->max_fragmented_size));
+
+	if (negotiated_version != SMBDIRECT_V1) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: negotiated_version=0x%x\n",
+			negotiated_version);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNREFUSED);
+		return;
+	}
+
+	if (status != le32_to_cpu(STATUS_SUCCESS)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: status=0x%x != 0x0\n",
+			status);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNREFUSED);
+		return;
+	}
+
+	if (max_receive_size < SMBDIRECT_MIN_RECEIVE_SIZE) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: max_receive_size=%u < %u\n",
+			max_receive_size,
+			SMBDIRECT_MIN_RECEIVE_SIZE);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	if (max_fragmented_size < SMBDIRECT_MIN_FRAGMENTED_SIZE) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: max_fragmented_size=%u < %u\n",
+			max_fragmented_size,
+			SMBDIRECT_MIN_FRAGMENTED_SIZE);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	if (credits_granted == 0) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: credits_granted == 0\n");
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	if (credits_requested == 0) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: credits_requested == 0\n");
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	if (preferred_send_size > sp->max_recv_size) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: preferred_send_size=%u > max_recv_size=%u\n",
+			preferred_send_size,
+			sp->max_recv_size);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	/*
+	 * We take the value from the peer, which is checked to be higher than 0,
+	 * but we limit it to the max value we support in order to have
+	 * the main logic simpler.
+	 */
+	sc->recv_io.credits.target = credits_requested;
+	sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target,
+					   sp->recv_credit_max);
+
+	/*
+	 * At least the value of SMBDIRECT_MIN_RECEIVE_SIZE is used.
+	 */
+	sp->max_recv_size = min_t(u32, sp->max_recv_size, preferred_send_size);
+	sp->max_recv_size = max_t(u32, sp->max_recv_size, SMBDIRECT_MIN_RECEIVE_SIZE);
+
+	/*
+	 * We already sent our sp->max_fragmented_recv_size
+	 * to the peer, so we can't lower it here any more.
+	 *
+	 * TODO: but if the peer lowered sp->max_recv_size
+	 * we will have to adjust our number of buffers.
+	 *
+	 * But for now we keep it as the cifs.ko code
+	 * worked before.
+	 */
+
+	/*
+	 * Note nrep->max_receive_size was already checked against
+	 * SMBDIRECT_MIN_RECEIVE_SIZE above.
+	 */
+	sp->max_send_size = min_t(u32, sp->max_send_size, max_receive_size);
+
+	/*
+	 * Make sure the resulting max_frmr_depth is at least 1,
+	 * which means max_read_write_size needs to be at least PAGE_SIZE.
+	 */
+	sp->max_read_write_size = min_t(u32, sp->max_frmr_depth * PAGE_SIZE,
+					max_readwrite_size);
+	if (sp->max_read_write_size < PAGE_SIZE) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: max_readwrite_size=%u < PAGE_SIZE(%lu)\n",
+			max_readwrite_size,
+			PAGE_SIZE);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+	sp->max_frmr_depth = sp->max_read_write_size / PAGE_SIZE;
+
+	/*
+	 * Note nrep->credits_granted was already checked against 0 above.
+	 */
+	atomic_set(&sc->send_io.credits.count, credits_granted);
+
+	/*
+	 * Note nrep->max_fragmented_size was already checked against
+	 * SMBDIRECT_MIN_FRAGMENTED_SIZE above.
+	 */
+	sp->max_fragmented_send_size = max_fragmented_size;
+
+	ret = smbdirect_connection_create_mr_list(sc);
+	if (ret) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_create_mr_list() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		smbdirect_socket_schedule_cleanup(sc, ret);
+		return;
+	}
+
+	/*
+	 * Prepare for receiving data_transfer messages
+	 */
+	sc->recv_io.reassembly.full_packet_received = true;
+	sc->recv_io.expected = SMBDIRECT_EXPECT_DATA_TRANSFER;
+	list_for_each_entry(recv_io, &sc->recv_io.free.list, list)
+		recv_io->cqe.done = smbdirect_connection_recv_io_done;
+	recv_io = NULL;
+
+	/*
+	 * We should at least post 1 smbdirect_recv_io!
+	 */
+	posted = smbdirect_connection_recv_io_refill(sc);
+	if (posted < 1) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_recv_io_refill() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	/*
+	 * smbdirect_connection_negotiation_done()
+	 * will setup all required things and wake up
+	 * the waiter.
+	 */
+	smbdirect_connection_negotiation_done(sc);
+}
+
+int smbdirect_connect_sync(struct smbdirect_socket *sc,
+			   const struct sockaddr *dst)
+{
+	int ret;
+
+	ret = smbdirect_connect(sc, dst);
+	if (ret) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connect(%pISpsfc) failed %1pe\n",
+			dst, SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	ret = smbdirect_connection_wait_for_connected(sc);
+	if (ret) {
+		int lvl = SMBDIRECT_LOG_ERR;
+
+		if (ret == -ENODEV)
+			lvl = SMBDIRECT_LOG_INFO;
+
+		smbdirect_log_rdma_event(sc, lvl,
+			"wait for smbdirect_connect(%pISpsfc) failed %1pe\n",
+			dst, SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connect_sync);
diff --git a/fs/smb/smbdirect/connection.c b/fs/smb/smbdirect/connection.c
new file mode 100644
index 000000000000..afd31fa12a36
--- /dev/null
+++ b/fs/smb/smbdirect/connection.c
@@ -0,0 +1,2190 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+#include <linux/folio_queue.h>
+
+struct smbdirect_map_sges {
+	struct ib_sge *sge;
+	size_t num_sge;
+	size_t max_sge;
+	struct ib_device *device;
+	u32 local_dma_lkey;
+	enum dma_data_direction direction;
+};
+
+static ssize_t smbdirect_map_sges_from_iter(struct iov_iter *iter, size_t len,
+					    struct smbdirect_map_sges *state);
+
+static void smbdirect_connection_recv_io_refill_work(struct work_struct *work);
+static void smbdirect_connection_send_immediate_work(struct work_struct *work);
+
+static void smbdirect_connection_qp_event_handler(struct ib_event *event, void *context)
+{
+	struct smbdirect_socket *sc = context;
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+		"%s on device %.*s socket %p (cm_id=%p) status %s first_error %1pe\n",
+		ib_event_msg(event->event),
+		IB_DEVICE_NAME_MAX,
+		event->device->name,
+		sc, sc->rdma.cm_id,
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+
+	switch (event->event) {
+	case IB_EVENT_CQ_ERR:
+	case IB_EVENT_QP_FATAL:
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		break;
+
+	default:
+		break;
+	}
+}
+
+static int smbdirect_connection_rdma_event_handler(struct rdma_cm_id *id,
+						   struct rdma_cm_event *event)
+{
+	struct smbdirect_socket *sc = id->context;
+	int ret = -ECONNRESET;
+
+	if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+		ret = -ENETDOWN;
+	if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
+		ret = event->status;
+
+	/*
+	 * cma_cm_event_handler() has
+	 * lockdep_assert_held(&id_priv->handler_mutex);
+	 *
+	 * Mutexes are not allowed in interrupts,
+	 * and we rely on not being in an interrupt here.
+	 */
+	WARN_ON_ONCE(in_interrupt());
+
+	if (event->event != sc->rdma.expected_event) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"%s (first_error=%1pe, expected=%s) => event=%s status=%d => ret=%1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			rdma_event_msg(sc->rdma.expected_event),
+			rdma_event_msg(event->event),
+			event->status,
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+
+		/*
+		 * If we get RDMA_CM_EVENT_DEVICE_REMOVAL,
+		 * we should change to SMBDIRECT_SOCKET_DISCONNECTED,
+		 * so that rdma_disconnect() is avoided later via
+		 * smbdirect_socket_schedule_cleanup[_status]() =>
+		 * smbdirect_socket_cleanup_work().
+		 *
+		 * As otherwise we'd set SMBDIRECT_SOCKET_DISCONNECTING,
+		 * but never ever get RDMA_CM_EVENT_DISCONNECTED and
+		 * never reach SMBDIRECT_SOCKET_DISCONNECTED.
+		 */
+		if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+			smbdirect_socket_schedule_cleanup_status(sc,
+								 SMBDIRECT_LOG_ERR,
+								 ret,
+								 SMBDIRECT_SOCKET_DISCONNECTED);
+		else
+			smbdirect_socket_schedule_cleanup(sc, ret);
+		if (sc->ib.qp)
+			ib_drain_qp(sc->ib.qp);
+		return 0;
+	}
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"%s (first_error=%1pe) event=%s\n",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+		rdma_event_msg(event->event));
+
+	switch (event->event) {
+	case RDMA_CM_EVENT_DISCONNECTED:
+		/*
+		 * We need to change to SMBDIRECT_SOCKET_DISCONNECTED,
+		 * so that rdma_disconnect() is avoided later via
+		 * smbdirect_socket_schedule_cleanup_status() =>
+		 * smbdirect_socket_cleanup_work().
+		 *
+		 * As otherwise we'd set SMBDIRECT_SOCKET_DISCONNECTING,
+		 * but never ever get RDMA_CM_EVENT_DISCONNECTED and
+		 * never reach SMBDIRECT_SOCKET_DISCONNECTED.
+		 *
+		 * This is also a normal disconnect so
+		 * SMBDIRECT_LOG_INFO should be good enough
+		 * and avoids spamming the default logs.
+		 */
+		smbdirect_socket_schedule_cleanup_status(sc,
+							 SMBDIRECT_LOG_INFO,
+							 ret,
+							 SMBDIRECT_SOCKET_DISCONNECTED);
+		if (sc->ib.qp)
+			ib_drain_qp(sc->ib.qp);
+		return 0;
+
+	default:
+		break;
+	}
+
+	/*
+	 * This is an internal error, should be handled above via
+	 * event->event != sc->rdma.expected_event already.
+	 */
+	WARN_ON_ONCE(sc->rdma.expected_event != RDMA_CM_EVENT_DISCONNECTED);
+	smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+	return 0;
+}
+
+void smbdirect_connection_rdma_established(struct smbdirect_socket *sc)
+{
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"rdma established: device: %.*s local: %pISpsfc remote: %pISpsfc\n",
+		IB_DEVICE_NAME_MAX,
+		sc->ib.dev->name,
+		&sc->rdma.cm_id->route.addr.src_addr,
+		&sc->rdma.cm_id->route.addr.dst_addr);
+
+	sc->rdma.cm_id->event_handler = smbdirect_connection_rdma_event_handler;
+	sc->rdma.expected_event = RDMA_CM_EVENT_DISCONNECTED;
+}
+
+void smbdirect_connection_negotiation_done(struct smbdirect_socket *sc)
+{
+	if (unlikely(sc->first_error))
+		return;
+
+	if (sc->status == SMBDIRECT_SOCKET_CONNECTED)
+		/*
+		 * This is the accept case where
+		 * smbdirect_socket_accept() already sets
+		 * SMBDIRECT_SOCKET_CONNECTED
+		 */
+		goto done;
+
+	if (sc->status != SMBDIRECT_SOCKET_NEGOTIATE_RUNNING) {
+		/*
+		 * Something went wrong...
+		 */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"status=%s first_error=%1pe local: %pISpsfc remote: %pISpsfc\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			&sc->rdma.cm_id->route.addr.src_addr,
+			&sc->rdma.cm_id->route.addr.dst_addr);
+		return;
+	}
+
+	/*
+	 * We are done, so we can wake up the waiter.
+	 */
+	WARN_ONCE(sc->status == SMBDIRECT_SOCKET_CONNECTED,
+		  "status=%s first_error=%1pe",
+		  smbdirect_socket_status_string(sc->status),
+		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+	sc->status = SMBDIRECT_SOCKET_CONNECTED;
+
+	/*
+	 * We need to setup the refill and send immediate work
+	 * in order to get a working connection.
+	 */
+done:
+	INIT_WORK(&sc->recv_io.posted.refill_work, smbdirect_connection_recv_io_refill_work);
+	INIT_WORK(&sc->idle.immediate_work, smbdirect_connection_send_immediate_work);
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"negotiated: local: %pISpsfc remote: %pISpsfc\n",
+		&sc->rdma.cm_id->route.addr.src_addr,
+		&sc->rdma.cm_id->route.addr.dst_addr);
+
+	wake_up(&sc->status_wait);
+}
+
+static u32 smbdirect_rdma_rw_send_wrs(struct ib_device *dev,
+				      const struct ib_qp_init_attr *attr)
+{
+	/*
+	 * This could be split out of rdma_rw_init_qp()
+	 * and be a helper function next to rdma_rw_mr_factor()
+	 *
+	 * We can't check unlikely(rdma_rw_force_mr) here,
+	 * but that is most likely 0 anyway.
+	 */
+	u32 factor;
+
+	WARN_ON_ONCE(attr->port_num == 0);
+
+	/*
+	 * Each context needs at least one RDMA READ or WRITE WR.
+	 *
+	 * For some hardware we might need more, eventually we should ask the
+	 * HCA driver for a multiplier here.
+	 */
+	factor = 1;
+
+	/*
+	 * If the device needs MRs to perform RDMA READ or WRITE operations,
+	 * we'll need two additional MRs for the registrations and the
+	 * invalidation.
+	 */
+	if (rdma_protocol_iwarp(dev, attr->port_num) || dev->attrs.max_sgl_rd)
+		factor += 2;	/* inv + reg */
+
+	return factor * attr->cap.max_rdma_ctxs;
+}
+
+int smbdirect_connection_create_qp(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct ib_qp_init_attr qp_attr;
+	struct ib_qp_cap qp_cap;
+	u32 rdma_send_wr;
+	u32 max_send_wr;
+	int ret;
+
+	/*
+	 * Note that {rdma,ib}_create_qp() will call
+	 * rdma_rw_init_qp() if max_rdma_ctxs is not 0.
+	 * It will adjust max_send_wr to the required
+	 * number of additional WRs for the RDMA RW operations.
+	 * It will cap max_send_wr to the device limit.
+	 *
+	 * We use allocate sp->responder_resources * 2 MRs
+	 * and each MR needs WRs for REG and INV, so
+	 * we use '* 4'.
+	 *
+	 * +1 for ib_drain_qp()
+	 */
+	memset(&qp_cap, 0, sizeof(qp_cap));
+	qp_cap.max_send_wr = sp->send_credit_target + sp->responder_resources * 4 + 1;
+	qp_cap.max_recv_wr = sp->recv_credit_max + 1;
+	qp_cap.max_send_sge = SMBDIRECT_SEND_IO_MAX_SGE;
+	qp_cap.max_recv_sge = SMBDIRECT_RECV_IO_MAX_SGE;
+	qp_cap.max_inline_data = 0;
+	qp_cap.max_rdma_ctxs = sc->rw_io.credits.max;
+
+	/*
+	 * Find out the number of max_send_wr
+	 * after rdma_rw_init_qp() adjusted it.
+	 *
+	 * We only do it on a temporary variable,
+	 * as rdma_create_qp() will trigger
+	 * rdma_rw_init_qp() again.
+	 */
+	memset(&qp_attr, 0, sizeof(qp_attr));
+	qp_attr.cap = qp_cap;
+	qp_attr.port_num = sc->rdma.cm_id->port_num;
+	rdma_send_wr = smbdirect_rdma_rw_send_wrs(sc->ib.dev, &qp_attr);
+	max_send_wr = qp_cap.max_send_wr + rdma_send_wr;
+
+	if (qp_cap.max_send_wr > sc->ib.dev->attrs.max_cqe ||
+	    qp_cap.max_send_wr > sc->ib.dev->attrs.max_qp_wr) {
+		pr_err("Possible CQE overrun: max_send_wr %d\n",
+		       qp_cap.max_send_wr);
+		pr_err("device %.*s reporting max_cqe %u max_qp_wr %u\n",
+		       IB_DEVICE_NAME_MAX,
+		       sc->ib.dev->name,
+		       sc->ib.dev->attrs.max_cqe,
+		       sc->ib.dev->attrs.max_qp_wr);
+		pr_err("consider lowering send_credit_target = %d\n",
+		       sp->send_credit_target);
+		return -EINVAL;
+	}
+
+	if (qp_cap.max_rdma_ctxs &&
+	    (max_send_wr >= sc->ib.dev->attrs.max_cqe ||
+	     max_send_wr >= sc->ib.dev->attrs.max_qp_wr)) {
+		pr_err("Possible CQE overrun: rdma_send_wr %d + max_send_wr %d = %d\n",
+		       rdma_send_wr, qp_cap.max_send_wr, max_send_wr);
+		pr_err("device %.*s reporting max_cqe %u max_qp_wr %u\n",
+		       IB_DEVICE_NAME_MAX,
+		       sc->ib.dev->name,
+		       sc->ib.dev->attrs.max_cqe,
+		       sc->ib.dev->attrs.max_qp_wr);
+		pr_err("consider lowering send_credit_target = %d, max_rdma_ctxs = %d\n",
+		       sp->send_credit_target, qp_cap.max_rdma_ctxs);
+		return -EINVAL;
+	}
+
+	if (qp_cap.max_recv_wr > sc->ib.dev->attrs.max_cqe ||
+	    qp_cap.max_recv_wr > sc->ib.dev->attrs.max_qp_wr) {
+		pr_err("Possible CQE overrun: max_recv_wr %d\n",
+		       qp_cap.max_recv_wr);
+		pr_err("device %.*s reporting max_cqe %u max_qp_wr %u\n",
+		       IB_DEVICE_NAME_MAX,
+		       sc->ib.dev->name,
+		       sc->ib.dev->attrs.max_cqe,
+		       sc->ib.dev->attrs.max_qp_wr);
+		pr_err("consider lowering receive_credit_max = %d\n",
+		       sp->recv_credit_max);
+		return -EINVAL;
+	}
+
+	if (qp_cap.max_send_sge > sc->ib.dev->attrs.max_send_sge ||
+	    qp_cap.max_recv_sge > sc->ib.dev->attrs.max_recv_sge) {
+		pr_err("device %.*s max_send_sge/max_recv_sge = %u/%u too small\n",
+		       IB_DEVICE_NAME_MAX,
+		       sc->ib.dev->name,
+		       sc->ib.dev->attrs.max_send_sge,
+		       sc->ib.dev->attrs.max_recv_sge);
+		return -EINVAL;
+	}
+
+	sc->ib.pd = ib_alloc_pd(sc->ib.dev, 0);
+	if (IS_ERR(sc->ib.pd)) {
+		pr_err("Can't create RDMA PD: %1pe\n", sc->ib.pd);
+		ret = PTR_ERR(sc->ib.pd);
+		sc->ib.pd = NULL;
+		return ret;
+	}
+
+	sc->ib.send_cq = ib_alloc_cq_any(sc->ib.dev, sc,
+					 max_send_wr,
+					 sc->ib.poll_ctx);
+	if (IS_ERR(sc->ib.send_cq)) {
+		pr_err("Can't create RDMA send CQ: %1pe\n", sc->ib.send_cq);
+		ret = PTR_ERR(sc->ib.send_cq);
+		sc->ib.send_cq = NULL;
+		goto err;
+	}
+
+	sc->ib.recv_cq = ib_alloc_cq_any(sc->ib.dev, sc,
+					 qp_cap.max_recv_wr,
+					 sc->ib.poll_ctx);
+	if (IS_ERR(sc->ib.recv_cq)) {
+		pr_err("Can't create RDMA recv CQ: %1pe\n", sc->ib.recv_cq);
+		ret = PTR_ERR(sc->ib.recv_cq);
+		sc->ib.recv_cq = NULL;
+		goto err;
+	}
+
+	/*
+	 * We reset completely here!
+	 * As the above use was just temporary
+	 * to calc max_send_wr and rdma_send_wr.
+	 *
+	 * rdma_create_qp() will trigger rdma_rw_init_qp()
+	 * again if max_rdma_ctxs is not 0.
+	 */
+	memset(&qp_attr, 0, sizeof(qp_attr));
+	qp_attr.event_handler = smbdirect_connection_qp_event_handler;
+	qp_attr.qp_context = sc;
+	qp_attr.cap = qp_cap;
+	qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
+	qp_attr.qp_type = IB_QPT_RC;
+	qp_attr.send_cq = sc->ib.send_cq;
+	qp_attr.recv_cq = sc->ib.recv_cq;
+	qp_attr.port_num = ~0;
+
+	ret = rdma_create_qp(sc->rdma.cm_id, sc->ib.pd, &qp_attr);
+	if (ret) {
+		pr_err("Can't create RDMA QP: %1pe\n",
+		       SMBDIRECT_DEBUG_ERR_PTR(ret));
+		goto err;
+	}
+	sc->ib.qp = sc->rdma.cm_id->qp;
+
+	return 0;
+err:
+	smbdirect_connection_destroy_qp(sc);
+	return ret;
+}
+
+void smbdirect_connection_destroy_qp(struct smbdirect_socket *sc)
+{
+	if (sc->ib.qp) {
+		ib_drain_qp(sc->ib.qp);
+		sc->ib.qp = NULL;
+		rdma_destroy_qp(sc->rdma.cm_id);
+	}
+	/*
+	 * These CQs were created with ib_alloc_cq_any(), which arms an internal
+	 * completion handler (ib_cq_poll_work for IB_POLL_WORKQUEUE). They MUST be
+	 * torn down with ib_free_cq(), which cancel_work_sync()es that poll work
+	 * before freeing the CQ. ib_destroy_cq() skips that step, so a completion
+	 * posted late by the (software) provider — e.g. rxe posting an RNR error
+	 * from rxe_receiver after rdma_destroy_qp() — re-queues ib_cq_poll_work on
+	 * an already-freed CQ (KASAN slab-use-after-free in ib_cq_poll_work).
+	 */
+	if (sc->ib.recv_cq) {
+		ib_free_cq(sc->ib.recv_cq);
+		sc->ib.recv_cq = NULL;
+	}
+	if (sc->ib.send_cq) {
+		ib_free_cq(sc->ib.send_cq);
+		sc->ib.send_cq = NULL;
+	}
+	if (sc->ib.pd) {
+		ib_dealloc_pd(sc->ib.pd);
+		sc->ib.pd = NULL;
+	}
+}
+
+int smbdirect_connection_create_mem_pools(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	char name[80];
+	size_t i;
+
+	/*
+	 * We use sizeof(struct smbdirect_negotiate_resp) for the
+	 * payload size as it is larger as
+	 * sizeof(struct smbdirect_data_transfer).
+	 *
+	 * This will fit client and server usage for now.
+	 */
+	snprintf(name, sizeof(name), "smbdirect_send_io_cache_%p", sc);
+	struct kmem_cache_args send_io_args = {
+		.align		= __alignof__(struct smbdirect_send_io),
+	};
+	sc->send_io.mem.cache = kmem_cache_create(name,
+						  sizeof(struct smbdirect_send_io) +
+						  sizeof(struct smbdirect_negotiate_resp),
+						  &send_io_args,
+						  SLAB_HWCACHE_ALIGN);
+	if (!sc->send_io.mem.cache)
+		goto err;
+
+	sc->send_io.mem.pool = mempool_create_slab_pool(sp->send_credit_target,
+							sc->send_io.mem.cache);
+	if (!sc->send_io.mem.pool)
+		goto err;
+
+	/*
+	 * A payload size of sp->max_recv_size should fit
+	 * any message.
+	 *
+	 * For smbdirect_data_transfer messages the whole
+	 * buffer might be exposed to userspace
+	 * (currently on the client side...)
+	 * The documentation says data_offset = 0 would be
+	 * strange but valid.
+	 */
+	snprintf(name, sizeof(name), "smbdirect_recv_io_cache_%p", sc);
+	struct kmem_cache_args recv_io_args = {
+		.align		= __alignof__(struct smbdirect_recv_io),
+		.useroffset	= sizeof(struct smbdirect_recv_io),
+		.usersize	= sp->max_recv_size,
+	};
+	sc->recv_io.mem.cache = kmem_cache_create(name,
+						  sizeof(struct smbdirect_recv_io) +
+						  sp->max_recv_size,
+						  &recv_io_args,
+						  SLAB_HWCACHE_ALIGN);
+	if (!sc->recv_io.mem.cache)
+		goto err;
+
+	sc->recv_io.mem.pool = mempool_create_slab_pool(sp->recv_credit_max,
+							sc->recv_io.mem.cache);
+	if (!sc->recv_io.mem.pool)
+		goto err;
+
+	for (i = 0; i < sp->recv_credit_max; i++) {
+		struct smbdirect_recv_io *recv_io;
+
+		recv_io = mempool_alloc(sc->recv_io.mem.pool,
+					sc->recv_io.mem.gfp_mask);
+		if (!recv_io)
+			goto err;
+		recv_io->socket = sc;
+		recv_io->sge.length = 0;
+		list_add_tail(&recv_io->list, &sc->recv_io.free.list);
+	}
+
+	return 0;
+err:
+	smbdirect_connection_destroy_mem_pools(sc);
+	return -ENOMEM;
+}
+
+void smbdirect_connection_destroy_mem_pools(struct smbdirect_socket *sc)
+{
+	struct smbdirect_recv_io *recv_io, *next_io;
+
+	list_for_each_entry_safe(recv_io, next_io, &sc->recv_io.free.list, list) {
+		list_del(&recv_io->list);
+		mempool_free(recv_io, sc->recv_io.mem.pool);
+	}
+
+	/*
+	 * Note mempool_destroy() and kmem_cache_destroy()
+	 * work fine with a NULL pointer
+	 */
+
+	mempool_destroy(sc->recv_io.mem.pool);
+	sc->recv_io.mem.pool = NULL;
+
+	kmem_cache_destroy(sc->recv_io.mem.cache);
+	sc->recv_io.mem.cache = NULL;
+
+	mempool_destroy(sc->send_io.mem.pool);
+	sc->send_io.mem.pool = NULL;
+
+	kmem_cache_destroy(sc->send_io.mem.cache);
+	sc->send_io.mem.cache = NULL;
+}
+
+struct smbdirect_send_io *smbdirect_connection_alloc_send_io(struct smbdirect_socket *sc)
+{
+	struct smbdirect_send_io *msg;
+
+	msg = mempool_alloc(sc->send_io.mem.pool, sc->send_io.mem.gfp_mask);
+	if (!msg)
+		return ERR_PTR(-ENOMEM);
+	msg->socket = sc;
+	INIT_LIST_HEAD(&msg->sibling_list);
+	msg->num_sge = 0;
+
+	return msg;
+}
+
+void smbdirect_connection_free_send_io(struct smbdirect_send_io *msg)
+{
+	struct smbdirect_socket *sc = msg->socket;
+	size_t i;
+
+	/*
+	 * The list needs to be empty!
+	 * The caller should take care of it.
+	 */
+	WARN_ON_ONCE(!list_empty(&msg->sibling_list));
+
+	/*
+	 * Note we call ib_dma_unmap_page(), even if some sges are mapped using
+	 * ib_dma_map_single().
+	 *
+	 * The difference between _single() and _page() only matters for the
+	 * ib_dma_map_*() case.
+	 *
+	 * For the ib_dma_unmap_*() case it does not matter as both take the
+	 * dma_addr_t and dma_unmap_single_attrs() is just an alias to
+	 * dma_unmap_page_attrs().
+	 */
+	for (i = 0; i < msg->num_sge; i++)
+		ib_dma_unmap_page(sc->ib.dev,
+				  msg->sge[i].addr,
+				  msg->sge[i].length,
+				  DMA_TO_DEVICE);
+
+	mempool_free(msg, sc->send_io.mem.pool);
+}
+
+struct smbdirect_recv_io *smbdirect_connection_get_recv_io(struct smbdirect_socket *sc)
+{
+	struct smbdirect_recv_io *msg = NULL;
+	unsigned long flags;
+
+	spin_lock_irqsave(&sc->recv_io.free.lock, flags);
+	if (likely(!sc->first_error))
+		msg = list_first_entry_or_null(&sc->recv_io.free.list,
+					       struct smbdirect_recv_io,
+					       list);
+	if (likely(msg)) {
+		list_del(&msg->list);
+		sc->statistics.get_receive_buffer++;
+	}
+	spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);
+
+	return msg;
+}
+
+void smbdirect_connection_put_recv_io(struct smbdirect_recv_io *msg)
+{
+	struct smbdirect_socket *sc = msg->socket;
+	unsigned long flags;
+
+	if (likely(msg->sge.length != 0)) {
+		ib_dma_unmap_single(sc->ib.dev,
+				    msg->sge.addr,
+				    msg->sge.length,
+				    DMA_FROM_DEVICE);
+		msg->sge.length = 0;
+	}
+
+	spin_lock_irqsave(&sc->recv_io.free.lock, flags);
+	list_add_tail(&msg->list, &sc->recv_io.free.list);
+	sc->statistics.put_receive_buffer++;
+	spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);
+
+	queue_work(sc->workqueues.refill, &sc->recv_io.posted.refill_work);
+}
+
+void smbdirect_connection_reassembly_append_recv_io(struct smbdirect_socket *sc,
+						    struct smbdirect_recv_io *msg,
+						    u32 data_length)
+{
+	unsigned long flags;
+
+	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
+	list_add_tail(&msg->list, &sc->recv_io.reassembly.list);
+	sc->recv_io.reassembly.queue_length++;
+	/*
+	 * Make sure reassembly_data_length is updated after list and
+	 * reassembly_queue_length are updated. On the dequeue side
+	 * reassembly_data_length is checked without a lock to determine
+	 * if reassembly_queue_length and list is up to date
+	 */
+	virt_wmb();
+	sc->recv_io.reassembly.data_length += data_length;
+	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
+	sc->statistics.enqueue_reassembly_queue++;
+}
+
+struct smbdirect_recv_io *
+smbdirect_connection_reassembly_first_recv_io(struct smbdirect_socket *sc)
+{
+	struct smbdirect_recv_io *msg;
+
+	msg = list_first_entry_or_null(&sc->recv_io.reassembly.list,
+				       struct smbdirect_recv_io,
+				       list);
+
+	return msg;
+}
+
+void smbdirect_connection_negotiate_rdma_resources(struct smbdirect_socket *sc,
+						   u8 peer_initiator_depth,
+						   u8 peer_responder_resources,
+						   const struct rdma_conn_param *param)
+{
+	struct smbdirect_socket_parameters *sp = &sc->parameters;
+
+	if (rdma_protocol_iwarp(sc->ib.dev, sc->rdma.cm_id->port_num) &&
+	    param->private_data_len == 8) {
+		/*
+		 * Legacy clients with only iWarp MPA v1 support
+		 * need a private blob in order to negotiate
+		 * the IRD/ORD values.
+		 */
+		const __be32 *ird_ord_hdr = param->private_data;
+		u32 ird32 = be32_to_cpu(ird_ord_hdr[0]);
+		u32 ord32 = be32_to_cpu(ird_ord_hdr[1]);
+
+		/*
+		 * cifs.ko sends the legacy IRD/ORD negotiation
+		 * event if iWarp MPA v2 was used.
+		 *
+		 * Here we check that the values match and only
+		 * mark the client as legacy if they don't match.
+		 */
+		if ((u32)param->initiator_depth != ird32 ||
+		    (u32)param->responder_resources != ord32) {
+			/*
+			 * There are broken clients (old cifs.ko)
+			 * using little endian and also
+			 * struct rdma_conn_param only uses u8
+			 * for initiator_depth and responder_resources,
+			 * so we truncate the value to U8_MAX.
+			 *
+			 * smb_direct_accept_client() will then
+			 * do the real negotiation in order to
+			 * select the minimum between client and
+			 * server.
+			 */
+			ird32 = min_t(u32, ird32, U8_MAX);
+			ord32 = min_t(u32, ord32, U8_MAX);
+
+			sc->rdma.legacy_iwarp = true;
+			peer_initiator_depth = (u8)ird32;
+			peer_responder_resources = (u8)ord32;
+		}
+	}
+
+	/*
+	 * negotiate the value by using the minimum
+	 * between client and server if the client provided
+	 * non 0 values.
+	 */
+	if (peer_initiator_depth != 0)
+		sp->initiator_depth = min_t(u8, sp->initiator_depth,
+					    peer_initiator_depth);
+	if (peer_responder_resources != 0)
+		sp->responder_resources = min_t(u8, sp->responder_resources,
+						peer_responder_resources);
+}
+
+bool smbdirect_connection_is_connected(struct smbdirect_socket *sc)
+{
+	if (unlikely(!sc || sc->first_error || sc->status != SMBDIRECT_SOCKET_CONNECTED))
+		return false;
+	return true;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_is_connected);
+
+int smbdirect_connection_wait_for_connected(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	union {
+		struct sockaddr sa;
+		struct sockaddr_storage ss;
+	} src_addr, dst_addr;
+	const struct sockaddr *src = NULL;
+	const struct sockaddr *dst = NULL;
+	char _devname[IB_DEVICE_NAME_MAX] = { 0, };
+	const char *devname = NULL;
+	int ret;
+
+	if (sc->rdma.cm_id) {
+		src_addr.ss = sc->rdma.cm_id->route.addr.src_addr;
+		if (src_addr.sa.sa_family != AF_UNSPEC)
+			src = &src_addr.sa;
+		dst_addr.ss = sc->rdma.cm_id->route.addr.dst_addr;
+		if (dst_addr.sa.sa_family != AF_UNSPEC)
+			dst = &dst_addr.sa;
+
+		if (sc->ib.dev) {
+			memcpy(_devname, sc->ib.dev->name, IB_DEVICE_NAME_MAX);
+			devname = _devname;
+		}
+	}
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"waiting for connection: device: %.*s local: %pISpsfc remote: %pISpsfc\n",
+		IB_DEVICE_NAME_MAX, devname, src, dst);
+
+	ret = wait_event_interruptible_timeout(sc->status_wait,
+					       sc->status == SMBDIRECT_SOCKET_CONNECTED ||
+					       sc->first_error,
+					       msecs_to_jiffies(sp->negotiate_timeout_msec));
+	if (sc->rdma.cm_id) {
+		/*
+		 * Maybe src and dev are updated in the meantime.
+		 */
+		src_addr.ss = sc->rdma.cm_id->route.addr.src_addr;
+		if (src_addr.sa.sa_family != AF_UNSPEC)
+			src = &src_addr.sa;
+		dst_addr.ss = sc->rdma.cm_id->route.addr.dst_addr;
+		if (dst_addr.sa.sa_family != AF_UNSPEC)
+			dst = &dst_addr.sa;
+
+		if (sc->ib.dev) {
+			memcpy(_devname, sc->ib.dev->name, IB_DEVICE_NAME_MAX);
+			devname = _devname;
+		}
+	}
+	if (ret == 0)
+		ret = -ETIMEDOUT;
+	if (ret < 0)
+		smbdirect_socket_schedule_cleanup(sc, ret);
+	if (sc->first_error) {
+		int lvl = SMBDIRECT_LOG_ERR;
+
+		ret = sc->first_error;
+		if (ret == -ENODEV)
+			lvl = SMBDIRECT_LOG_INFO;
+
+		smbdirect_log_rdma_event(sc, lvl,
+			"connection failed %1pe device: %.*s local: %pISpsfc remote: %pISpsfc\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret),
+			IB_DEVICE_NAME_MAX, devname, src, dst);
+		return ret;
+	}
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_wait_for_connected);
+
+void smbdirect_connection_idle_timer_work(struct work_struct *work)
+{
+	struct smbdirect_socket *sc =
+		container_of(work, struct smbdirect_socket, idle.timer_work.work);
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+
+	if (sc->idle.keepalive != SMBDIRECT_KEEPALIVE_NONE) {
+		smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_ERR,
+			"%s => timeout sc->idle.keepalive=%s\n",
+			smbdirect_socket_status_string(sc->status),
+			sc->idle.keepalive == SMBDIRECT_KEEPALIVE_SENT ?
+			"SENT" : "PENDING");
+		smbdirect_socket_schedule_cleanup(sc, -ETIMEDOUT);
+		return;
+	}
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
+		return;
+
+	/*
+	 * Now use the keepalive timeout (instead of keepalive interval)
+	 * in order to wait for a response
+	 */
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->keepalive_timeout_msec));
+	smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
+		"schedule send of empty idle message\n");
+	queue_work(sc->workqueues.immediate, &sc->idle.immediate_work);
+}
+
+u16 smbdirect_connection_grant_recv_credits(struct smbdirect_socket *sc)
+{
+	int missing;
+	int available;
+	int new_credits;
+
+	if (atomic_read(&sc->recv_io.credits.count) >= sc->recv_io.credits.target)
+		return 0;
+
+	missing = (int)sc->recv_io.credits.target - atomic_read(&sc->recv_io.credits.count);
+	available = atomic_xchg(&sc->recv_io.credits.available, 0);
+	new_credits = min3((int)U16_MAX, missing, available);
+	if (new_credits <= 0) {
+		/*
+		 * If credits are available, but not granted
+		 * we need to re-add them again.
+		 */
+		if (available)
+			atomic_add(available, &sc->recv_io.credits.available);
+		return 0;
+	}
+
+	if (new_credits < available) {
+		/*
+		 * Readd the remaining available again.
+		 */
+		available -= new_credits;
+		atomic_add(available, &sc->recv_io.credits.available);
+	}
+
+	/*
+	 * Remember we granted the credits
+	 */
+	atomic_add(new_credits, &sc->recv_io.credits.count);
+	return new_credits;
+}
+
+static bool smbdirect_connection_request_keep_alive(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+
+	if (sc->idle.keepalive == SMBDIRECT_KEEPALIVE_PENDING) {
+		sc->idle.keepalive = SMBDIRECT_KEEPALIVE_SENT;
+		/*
+		 * Now use the keepalive timeout (instead of keepalive interval)
+		 * in order to wait for a response
+		 */
+		mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+				 msecs_to_jiffies(sp->keepalive_timeout_msec));
+		return true;
+	}
+
+	return false;
+}
+
+int smbdirect_connection_post_send_wr(struct smbdirect_socket *sc,
+				      struct ib_send_wr *wr)
+{
+	int ret;
+
+	if (unlikely(sc->first_error))
+		return sc->first_error;
+
+	atomic_inc(&sc->send_io.pending.count);
+	ret = ib_post_send(sc->ib.qp, wr, NULL);
+	if (ret) {
+		atomic_dec(&sc->send_io.pending.count);
+		smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+			"ib_post_send() failed %1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		smbdirect_socket_schedule_cleanup(sc, ret);
+	}
+
+	return ret;
+}
+
+static void smbdirect_connection_send_batch_init(struct smbdirect_send_batch *batch,
+						 bool need_invalidate_rkey,
+						 unsigned int remote_key)
+{
+	INIT_LIST_HEAD(&batch->msg_list);
+	batch->wr_cnt = 0;
+	batch->need_invalidate_rkey = need_invalidate_rkey;
+	batch->remote_key = remote_key;
+	batch->credit = 0;
+}
+
+int smbdirect_connection_send_batch_flush(struct smbdirect_socket *sc,
+					  struct smbdirect_send_batch *batch,
+					  bool is_last)
+{
+	struct smbdirect_send_io *first, *last;
+	int ret = 0;
+
+	if (list_empty(&batch->msg_list))
+		goto release_credit;
+
+	first = list_first_entry(&batch->msg_list,
+				 struct smbdirect_send_io,
+				 sibling_list);
+	last = list_last_entry(&batch->msg_list,
+			       struct smbdirect_send_io,
+			       sibling_list);
+
+	if (batch->need_invalidate_rkey) {
+		first->wr.opcode = IB_WR_SEND_WITH_INV;
+		first->wr.ex.invalidate_rkey = batch->remote_key;
+		batch->need_invalidate_rkey = false;
+		batch->remote_key = 0;
+	}
+
+	last->wr.send_flags = IB_SEND_SIGNALED;
+	last->wr.wr_cqe = &last->cqe;
+
+	/*
+	 * Remove last from send_ctx->msg_list
+	 * and splice the rest of send_ctx->msg_list
+	 * to last->sibling_list.
+	 *
+	 * send_ctx->msg_list is a valid empty list
+	 * at the end.
+	 */
+	list_del_init(&last->sibling_list);
+	list_splice_tail_init(&batch->msg_list, &last->sibling_list);
+	batch->wr_cnt = 0;
+
+	ret = smbdirect_connection_post_send_wr(sc, &first->wr);
+	if (ret) {
+		struct smbdirect_send_io *sibling, *next;
+
+		list_for_each_entry_safe(sibling, next, &last->sibling_list, sibling_list) {
+			list_del_init(&sibling->sibling_list);
+			smbdirect_connection_free_send_io(sibling);
+		}
+		smbdirect_connection_free_send_io(last);
+	}
+
+release_credit:
+	if (is_last && !ret && batch->credit) {
+		atomic_add(batch->credit, &sc->send_io.bcredits.count);
+		batch->credit = 0;
+		wake_up(&sc->send_io.bcredits.wait_queue);
+	}
+
+	return ret;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_send_batch_flush);
+
+struct smbdirect_send_batch *
+smbdirect_init_send_batch_storage(struct smbdirect_send_batch_storage *storage,
+				  bool need_invalidate_rkey,
+				  unsigned int remote_key)
+{
+	struct smbdirect_send_batch *batch = (struct smbdirect_send_batch *)storage;
+
+	memset(storage, 0, sizeof(*storage));
+	BUILD_BUG_ON(sizeof(*batch) > sizeof(*storage));
+
+	smbdirect_connection_send_batch_init(batch,
+					     need_invalidate_rkey,
+					     remote_key);
+
+	return batch;
+}
+EXPORT_SYMBOL_GPL(smbdirect_init_send_batch_storage);
+
+static int smbdirect_connection_wait_for_send_bcredit(struct smbdirect_socket *sc,
+						      struct smbdirect_send_batch *batch)
+{
+	int ret;
+
+	if (batch->credit)
+		return 0;
+
+	ret = smbdirect_socket_wait_for_credits(sc,
+						 SMBDIRECT_SOCKET_CONNECTED,
+						 -ENOTCONN,
+						 &sc->send_io.bcredits.wait_queue,
+						 &sc->send_io.bcredits.count,
+						 1);
+	if (ret)
+		return ret;
+
+	batch->credit = 1;
+	return 0;
+}
+
+static int smbdirect_connection_wait_for_send_lcredit(struct smbdirect_socket *sc,
+						      struct smbdirect_send_batch *batch)
+{
+	if (batch && atomic_read(&sc->send_io.lcredits.count) <= 1) {
+		int ret;
+
+		ret = smbdirect_connection_send_batch_flush(sc, batch, false);
+		if (ret)
+			return ret;
+	}
+
+	return smbdirect_socket_wait_for_credits(sc,
+						 SMBDIRECT_SOCKET_CONNECTED,
+						 -ENOTCONN,
+						 &sc->send_io.lcredits.wait_queue,
+						 &sc->send_io.lcredits.count,
+						 1);
+}
+
+static int smbdirect_connection_wait_for_send_credits(struct smbdirect_socket *sc,
+						      struct smbdirect_send_batch *batch)
+{
+	if (batch && (batch->wr_cnt >= 16 || atomic_read(&sc->send_io.credits.count) <= 1)) {
+		int ret;
+
+		ret = smbdirect_connection_send_batch_flush(sc, batch, false);
+		if (ret)
+			return ret;
+	}
+
+	return smbdirect_socket_wait_for_credits(sc,
+						 SMBDIRECT_SOCKET_CONNECTED,
+						 -ENOTCONN,
+						 &sc->send_io.credits.wait_queue,
+						 &sc->send_io.credits.count,
+						 1);
+}
+
+static void smbdirect_connection_send_io_done(struct ib_cq *cq, struct ib_wc *wc);
+
+static int smbdirect_connection_post_send_io(struct smbdirect_socket *sc,
+					     struct smbdirect_send_batch *batch,
+					     struct smbdirect_send_io *msg)
+{
+	int i;
+
+	for (i = 0; i < msg->num_sge; i++)
+		ib_dma_sync_single_for_device(sc->ib.dev,
+					      msg->sge[i].addr, msg->sge[i].length,
+					      DMA_TO_DEVICE);
+
+	msg->cqe.done = smbdirect_connection_send_io_done;
+	msg->wr.wr_cqe = &msg->cqe;
+	msg->wr.opcode = IB_WR_SEND;
+	msg->wr.sg_list = &msg->sge[0];
+	msg->wr.num_sge = msg->num_sge;
+	msg->wr.next = NULL;
+
+	if (batch) {
+		msg->wr.send_flags = 0;
+		if (!list_empty(&batch->msg_list)) {
+			struct smbdirect_send_io *last;
+
+			last = list_last_entry(&batch->msg_list,
+					       struct smbdirect_send_io,
+					       sibling_list);
+			last->wr.next = &msg->wr;
+		}
+		list_add_tail(&msg->sibling_list, &batch->msg_list);
+		batch->wr_cnt++;
+		return 0;
+	}
+
+	msg->wr.send_flags = IB_SEND_SIGNALED;
+	return smbdirect_connection_post_send_wr(sc, &msg->wr);
+}
+
+int smbdirect_connection_send_single_iter(struct smbdirect_socket *sc,
+					  struct smbdirect_send_batch *batch,
+					  struct iov_iter *iter,
+					  unsigned int flags,
+					  u32 remaining_data_length)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_send_batch _batch;
+	struct smbdirect_send_io *msg;
+	struct smbdirect_data_transfer *packet;
+	size_t header_length;
+	u16 new_credits = 0;
+	u32 data_length = 0;
+	int ret;
+
+	if (WARN_ON_ONCE(flags))
+		return -EINVAL; /* no flags support for now */
+
+	if (iter) {
+		if (WARN_ON_ONCE(iov_iter_rw(iter) != ITER_SOURCE))
+			return -EINVAL; /* It's a bug in upper layer to get there */
+
+		header_length = sizeof(struct smbdirect_data_transfer);
+		if (WARN_ON_ONCE(remaining_data_length == 0 ||
+				 iov_iter_count(iter) > remaining_data_length))
+			return -EINVAL;
+	} else {
+		/* If this is a packet without payload, don't send padding */
+		header_length = offsetof(struct smbdirect_data_transfer, padding);
+		if (WARN_ON_ONCE(remaining_data_length))
+			return -EINVAL;
+	}
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
+		smbdirect_log_write(sc, SMBDIRECT_LOG_ERR,
+			"status=%s first_error=%1pe => %1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
+		return -ENOTCONN;
+	}
+
+	if (!batch) {
+		smbdirect_connection_send_batch_init(&_batch, false, 0);
+		batch = &_batch;
+	}
+
+	ret = smbdirect_connection_wait_for_send_bcredit(sc, batch);
+	if (ret)
+		goto bcredit_failed;
+
+	ret = smbdirect_connection_wait_for_send_lcredit(sc, batch);
+	if (ret)
+		goto lcredit_failed;
+
+	ret = smbdirect_connection_wait_for_send_credits(sc, batch);
+	if (ret)
+		goto credit_failed;
+
+	new_credits = smbdirect_connection_grant_recv_credits(sc);
+	if (new_credits == 0 &&
+	    atomic_read(&sc->send_io.credits.count) == 0 &&
+	    atomic_read(&sc->recv_io.credits.count) == 0) {
+		/*
+		 * queue the refill work in order to
+		 * get some new recv credits we can grant to
+		 * the peer.
+		 */
+		queue_work(sc->workqueues.refill, &sc->recv_io.posted.refill_work);
+
+		/*
+		 * wait until either the refill work or the peer
+		 * granted new credits
+		 */
+		ret = wait_event_interruptible(sc->send_io.credits.wait_queue,
+					       atomic_read(&sc->send_io.credits.count) >= 1 ||
+					       atomic_read(&sc->recv_io.credits.available) >= 1 ||
+					       sc->status != SMBDIRECT_SOCKET_CONNECTED);
+		if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
+			ret = -ENOTCONN;
+		if (ret < 0)
+			goto credit_failed;
+
+		new_credits = smbdirect_connection_grant_recv_credits(sc);
+	}
+
+	msg = smbdirect_connection_alloc_send_io(sc);
+	if (IS_ERR(msg)) {
+		ret = PTR_ERR(msg);
+		goto alloc_failed;
+	}
+
+	/* Map the packet to DMA */
+	msg->sge[0].addr = ib_dma_map_single(sc->ib.dev,
+					     msg->packet,
+					     header_length,
+					     DMA_TO_DEVICE);
+	ret = ib_dma_mapping_error(sc->ib.dev, msg->sge[0].addr);
+	if (ret)
+		goto err;
+
+	msg->sge[0].length = header_length;
+	msg->sge[0].lkey = sc->ib.pd->local_dma_lkey;
+	msg->num_sge = 1;
+
+	if (iter) {
+		struct smbdirect_map_sges extract = {
+			.num_sge	= msg->num_sge,
+			.max_sge	= ARRAY_SIZE(msg->sge),
+			.sge		= msg->sge,
+			.device		= sc->ib.dev,
+			.local_dma_lkey	= sc->ib.pd->local_dma_lkey,
+			.direction	= DMA_TO_DEVICE,
+		};
+		size_t payload_len = umin(iov_iter_count(iter),
+					  sp->max_send_size - sizeof(*packet));
+
+		ret = smbdirect_map_sges_from_iter(iter, payload_len, &extract);
+		if (ret < 0)
+			goto err;
+		data_length = ret;
+		remaining_data_length -= data_length;
+		msg->num_sge = extract.num_sge;
+	}
+
+	/* Fill in the packet header */
+	packet = (struct smbdirect_data_transfer *)msg->packet;
+	packet->credits_requested = cpu_to_le16(sp->send_credit_target);
+	packet->credits_granted = cpu_to_le16(new_credits);
+
+	packet->flags = 0;
+	if (smbdirect_connection_request_keep_alive(sc))
+		packet->flags |= cpu_to_le16(SMBDIRECT_FLAG_RESPONSE_REQUESTED);
+
+	packet->reserved = 0;
+	if (!data_length)
+		packet->data_offset = 0;
+	else
+		packet->data_offset = cpu_to_le32(24);
+	packet->data_length = cpu_to_le32(data_length);
+	packet->remaining_data_length = cpu_to_le32(remaining_data_length);
+	packet->padding = 0;
+
+	smbdirect_log_outgoing(sc, SMBDIRECT_LOG_INFO,
+		"DataOut: %s=%u, %s=%u, %s=0x%x, %s=%u, %s=%u, %s=%u\n",
+		"CreditsRequested",
+		le16_to_cpu(packet->credits_requested),
+		"CreditsGranted",
+		le16_to_cpu(packet->credits_granted),
+		"Flags",
+		le16_to_cpu(packet->flags),
+		"RemainingDataLength",
+		le32_to_cpu(packet->remaining_data_length),
+		"DataOffset",
+		le32_to_cpu(packet->data_offset),
+		"DataLength",
+		le32_to_cpu(packet->data_length));
+
+	ret = smbdirect_connection_post_send_io(sc, batch, msg);
+	if (ret)
+		goto err;
+
+	/*
+	 * From here msg is moved to send_ctx
+	 * and we should not free it explicitly.
+	 */
+
+	if (batch == &_batch) {
+		ret = smbdirect_connection_send_batch_flush(sc, batch, true);
+		if (ret)
+			goto flush_failed;
+	}
+
+	return data_length;
+err:
+	smbdirect_connection_free_send_io(msg);
+flush_failed:
+alloc_failed:
+	atomic_inc(&sc->send_io.credits.count);
+credit_failed:
+	atomic_inc(&sc->send_io.lcredits.count);
+lcredit_failed:
+	atomic_add(batch->credit, &sc->send_io.bcredits.count);
+	batch->credit = 0;
+bcredit_failed:
+	return ret;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_send_single_iter);
+
+int smbdirect_connection_send_wait_zero_pending(struct smbdirect_socket *sc)
+{
+	/*
+	 * As an optimization, we don't wait for individual I/O to finish
+	 * before sending the next one.
+	 * Send them all and wait for pending send count to get to 0
+	 * that means all the I/Os have been out and we are good to return
+	 */
+
+	wait_event(sc->send_io.pending.zero_wait_queue,
+		   atomic_read(&sc->send_io.pending.count) == 0 ||
+		   sc->status != SMBDIRECT_SOCKET_CONNECTED);
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
+		smbdirect_log_write(sc, SMBDIRECT_LOG_ERR,
+			"status=%s first_error=%1pe => %1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
+		return -ENOTCONN;
+	}
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_send_wait_zero_pending);
+
+int smbdirect_connection_send_iter(struct smbdirect_socket *sc,
+				   struct iov_iter *iter,
+				   unsigned int flags,
+				   bool need_invalidate,
+				   unsigned int remote_key)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_send_batch batch;
+	int total_count = iov_iter_count(iter);
+	int ret;
+	int error = 0;
+	__be32 hdr;
+
+	if (WARN_ONCE(flags, "unexpected flags=0x%x\n", flags))
+		return -EINVAL; /* no flags support for now */
+
+	if (WARN_ON_ONCE(iov_iter_rw(iter) != ITER_SOURCE))
+		return -EINVAL; /* It's a bug in upper layer to get there */
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
+		smbdirect_log_write(sc, SMBDIRECT_LOG_INFO,
+			"status=%s first_error=%1pe => %1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
+		return -ENOTCONN;
+	}
+
+	/*
+	 * For now we expect the iter to have the full
+	 * message, including a 4 byte length header.
+	 */
+	if (iov_iter_count(iter) <= 4)
+		return -EINVAL;
+	if (!copy_from_iter_full(&hdr, sizeof(hdr), iter))
+		return -EFAULT;
+	if (iov_iter_count(iter) != be32_to_cpu(hdr))
+		return -EINVAL;
+
+	/*
+	 * The size must fit into the negotiated
+	 * fragmented send size.
+	 */
+	if (iov_iter_count(iter) > sp->max_fragmented_send_size)
+		return -EMSGSIZE;
+
+	smbdirect_log_write(sc, SMBDIRECT_LOG_INFO,
+		"Sending (RDMA): length=%zu\n",
+		iov_iter_count(iter));
+
+	smbdirect_connection_send_batch_init(&batch, need_invalidate, remote_key);
+	while (iov_iter_count(iter)) {
+		ret = smbdirect_connection_send_single_iter(sc,
+							    &batch,
+							    iter,
+							    flags,
+							    iov_iter_count(iter));
+		if (unlikely(ret < 0)) {
+			error = ret;
+			break;
+		}
+	}
+
+	ret = smbdirect_connection_send_batch_flush(sc, &batch, true);
+	if (unlikely(ret && !error))
+		error = ret;
+
+	/*
+	 * As an optimization, we don't wait for individual I/O to finish
+	 * before sending the next one.
+	 * Send them all and wait for pending send count to get to 0
+	 * that means all the I/Os have been out and we are good to return
+	 */
+
+	ret = smbdirect_connection_send_wait_zero_pending(sc);
+	if (unlikely(ret && !error))
+		error = ret;
+
+	if (unlikely(error))
+		return error;
+
+	return total_count;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_send_iter);
+
+static void smbdirect_connection_send_io_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_send_io *msg =
+		container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
+	struct smbdirect_socket *sc = msg->socket;
+	struct smbdirect_send_io *sibling, *next;
+	int lcredits = 0;
+
+	smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_INFO,
+		"smbdirect_send_io completed. status='%s (%d)', opcode=%d\n",
+		ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+
+	if (unlikely(!(msg->wr.send_flags & IB_SEND_SIGNALED))) {
+		/*
+		 * This happens when smbdirect_send_io is a sibling
+		 * before the final message, it is signaled on
+		 * error anyway, so we need to skip
+		 * smbdirect_connection_free_send_io here,
+		 * otherwise is will destroy the memory
+		 * of the siblings too, which will cause
+		 * use after free problems for the others
+		 * triggered from ib_drain_qp().
+		 */
+		if (wc->status != IB_WC_SUCCESS)
+			goto skip_free;
+
+		/*
+		 * This should not happen!
+		 * But we better just close the
+		 * connection...
+		 */
+		smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+			"unexpected send completion wc->status=%s (%d) wc->opcode=%d\n",
+			ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	/*
+	 * Free possible siblings and then the main send_io
+	 */
+	list_for_each_entry_safe(sibling, next, &msg->sibling_list, sibling_list) {
+		list_del_init(&sibling->sibling_list);
+		smbdirect_connection_free_send_io(sibling);
+		lcredits += 1;
+	}
+	/* Note this frees wc->wr_cqe, but not wc */
+	smbdirect_connection_free_send_io(msg);
+	lcredits += 1;
+
+	if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_SEND))) {
+skip_free:
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
+				"wc->status=%s (%d) wc->opcode=%d\n",
+				ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+		return;
+	}
+
+	atomic_add(lcredits, &sc->send_io.lcredits.count);
+	wake_up(&sc->send_io.lcredits.wait_queue);
+
+	if (atomic_dec_and_test(&sc->send_io.pending.count))
+		wake_up(&sc->send_io.pending.zero_wait_queue);
+}
+
+static void smbdirect_connection_send_immediate_work(struct work_struct *work)
+{
+	struct smbdirect_socket *sc =
+		container_of(work, struct smbdirect_socket, idle.immediate_work);
+	int ret;
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
+		return;
+
+	smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
+		"send an empty message\n");
+	sc->statistics.send_empty++;
+	ret = smbdirect_connection_send_single_iter(sc, NULL, NULL, 0, 0);
+	if (ret < 0) {
+		smbdirect_log_write(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_send_single_iter ret=%1pe\n",
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+		smbdirect_socket_schedule_cleanup(sc, ret);
+	}
+}
+
+int smbdirect_connection_post_recv_io(struct smbdirect_recv_io *msg)
+{
+	struct smbdirect_socket *sc = msg->socket;
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct ib_recv_wr recv_wr = {
+		.wr_cqe = &msg->cqe,
+		.sg_list = &msg->sge,
+		.num_sge = 1,
+	};
+	int ret;
+
+	if (unlikely(sc->first_error))
+		return sc->first_error;
+
+	msg->sge.addr = ib_dma_map_single(sc->ib.dev,
+					  msg->packet,
+					  sp->max_recv_size,
+					  DMA_FROM_DEVICE);
+	ret = ib_dma_mapping_error(sc->ib.dev, msg->sge.addr);
+	if (ret)
+		return ret;
+
+	msg->sge.length = sp->max_recv_size;
+	msg->sge.lkey = sc->ib.pd->local_dma_lkey;
+
+	ret = ib_post_recv(sc->ib.qp, &recv_wr, NULL);
+	if (ret) {
+		smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
+			"ib_post_recv failed ret=%d (%1pe)\n",
+			ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
+		ib_dma_unmap_single(sc->ib.dev,
+				    msg->sge.addr,
+				    msg->sge.length,
+				    DMA_FROM_DEVICE);
+		msg->sge.length = 0;
+		smbdirect_socket_schedule_cleanup(sc, ret);
+	}
+
+	return ret;
+}
+
+void smbdirect_connection_recv_io_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_recv_io *recv_io =
+		container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
+	struct smbdirect_socket *sc = recv_io->socket;
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_data_transfer *data_transfer;
+	int current_recv_credits;
+	u16 old_recv_credit_target;
+	u16 credits_requested;
+	u16 credits_granted;
+	u16 flags;
+	u32 data_offset;
+	u32 data_length;
+	u32 remaining_data_length;
+
+	if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_RECV))) {
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
+				"wc->status=%s (%d) wc->opcode=%d\n",
+				ib_wc_status_msg(wc->status), wc->status, wc->opcode);
+		goto error;
+	}
+
+	smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_INFO,
+		"recv_io=0x%p type=%d wc status=%s wc opcode %d byte_len=%d pkey_index=%u\n",
+		recv_io, sc->recv_io.expected,
+		ib_wc_status_msg(wc->status), wc->opcode,
+		wc->byte_len, wc->pkey_index);
+
+	/*
+	 * Reset timer to the keepalive interval in
+	 * order to trigger our next keepalive message.
+	 */
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
+	mod_delayed_work(sc->workqueues.idle, &sc->idle.timer_work,
+			 msecs_to_jiffies(sp->keepalive_interval_msec));
+
+	ib_dma_sync_single_for_cpu(sc->ib.dev,
+				   recv_io->sge.addr,
+				   recv_io->sge.length,
+				   DMA_FROM_DEVICE);
+
+	if (unlikely(wc->byte_len <
+	    offsetof(struct smbdirect_data_transfer, padding))) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"wc->byte_len=%u < %zu\n",
+			wc->byte_len,
+			offsetof(struct smbdirect_data_transfer, padding));
+		goto error;
+	}
+
+	data_transfer = (struct smbdirect_data_transfer *)recv_io->packet;
+	credits_requested = le16_to_cpu(data_transfer->credits_requested);
+	credits_granted = le16_to_cpu(data_transfer->credits_granted);
+	flags = le16_to_cpu(data_transfer->flags);
+	remaining_data_length = le32_to_cpu(data_transfer->remaining_data_length);
+	data_offset = le32_to_cpu(data_transfer->data_offset);
+	data_length = le32_to_cpu(data_transfer->data_length);
+
+	smbdirect_log_incoming(sc, SMBDIRECT_LOG_INFO,
+		"DataIn: %s=%u, %s=%u, %s=0x%x, %s=%u, %s=%u, %s=%u\n",
+		"CreditsRequested",
+		credits_requested,
+		"CreditsGranted",
+		credits_granted,
+		"Flags",
+		flags,
+		"RemainingDataLength",
+		remaining_data_length,
+		"DataOffset",
+		data_offset,
+		"DataLength",
+		data_length);
+
+	if (unlikely(credits_requested == 0)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: credits_requested == 0\n");
+		goto error;
+	}
+
+	if (unlikely(data_offset % 8 != 0)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"invalid: data_offset=%u (0x%x) not aligned to 8\n",
+			data_offset, data_offset);
+		goto error;
+	}
+
+	if (unlikely(wc->byte_len < data_offset ||
+	    (u64)wc->byte_len < (u64)data_offset + data_length)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"wc->byte_len=%u < date_offset=%u + data_length=%u\n",
+			wc->byte_len, data_offset, data_length);
+		goto error;
+	}
+
+	if (unlikely(remaining_data_length > sp->max_fragmented_recv_size ||
+	    data_length > sp->max_fragmented_recv_size ||
+	    (u64)remaining_data_length + (u64)data_length > (u64)sp->max_fragmented_recv_size)) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"remaining_data_length=%u + data_length=%u > max_fragmented=%u\n",
+			remaining_data_length, data_length, sp->max_fragmented_recv_size);
+		goto error;
+	}
+
+	if (data_length) {
+		if (sc->recv_io.reassembly.full_packet_received)
+			recv_io->first_segment = true;
+
+		if (remaining_data_length)
+			sc->recv_io.reassembly.full_packet_received = false;
+		else
+			sc->recv_io.reassembly.full_packet_received = true;
+	}
+
+	atomic_dec(&sc->recv_io.posted.count);
+	current_recv_credits = atomic_dec_return(&sc->recv_io.credits.count);
+
+	/*
+	 * We take the value from the peer, which is checked to be higher than 0,
+	 * but we limit it to the max value we support in order to have
+	 * the main logic simpler.
+	 */
+	old_recv_credit_target = sc->recv_io.credits.target;
+	sc->recv_io.credits.target = credits_requested;
+	sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target,
+					   sp->recv_credit_max);
+	if (credits_granted) {
+		atomic_add(credits_granted, &sc->send_io.credits.count);
+		/*
+		 * We have new send credits granted from remote peer
+		 * If any sender is waiting for credits, unblock it
+		 */
+		wake_up(&sc->send_io.credits.wait_queue);
+	}
+
+	/* Send an immediate response right away if requested */
+	if (flags & SMBDIRECT_FLAG_RESPONSE_REQUESTED) {
+		smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
+			"schedule send of immediate response\n");
+		queue_work(sc->workqueues.immediate, &sc->idle.immediate_work);
+	}
+
+	/*
+	 * If this is a packet with data playload place the data in
+	 * reassembly queue and wake up the reading thread
+	 */
+	if (data_length) {
+		if (current_recv_credits <= (sc->recv_io.credits.target / 4) ||
+		    sc->recv_io.credits.target > old_recv_credit_target)
+			queue_work(sc->workqueues.refill, &sc->recv_io.posted.refill_work);
+
+		smbdirect_connection_reassembly_append_recv_io(sc, recv_io, data_length);
+		wake_up(&sc->recv_io.reassembly.wait_queue);
+	} else
+		smbdirect_connection_put_recv_io(recv_io);
+
+	return;
+
+error:
+	/*
+	 * Make sure smbdirect_connection_put_recv_io() does not
+	 * start recv_io.posted.refill_work.
+	 */
+	disable_work(&sc->recv_io.posted.refill_work);
+	smbdirect_connection_put_recv_io(recv_io);
+	smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+}
+
+int smbdirect_connection_recv_io_refill(struct smbdirect_socket *sc)
+{
+	int missing;
+	int posted = 0;
+
+	if (unlikely(sc->first_error))
+		return sc->first_error;
+
+	/*
+	 * Find out how much smbdirect_recv_io buffers we should post.
+	 *
+	 * Note that sc->recv_io.credits.target is the value
+	 * from the peer and it can in theory change over time,
+	 * but it is forced to be at least 1 and at max
+	 * sp->recv_credit_max.
+	 *
+	 * So it can happen that missing will be lower than 0,
+	 * which means the peer has recently lowered its desired
+	 * target, while be already granted a higher number of credits.
+	 *
+	 * Note 'posted' is the number of smbdirect_recv_io buffers
+	 * posted within this function, while sc->recv_io.posted.count
+	 * is the overall value of posted smbdirect_recv_io buffers.
+	 *
+	 * We try to post as much buffers as missing, but
+	 * this is limited if a lot of smbdirect_recv_io buffers
+	 * are still in the sc->recv_io.reassembly.list instead of
+	 * the sc->recv_io.free.list.
+	 *
+	 */
+	missing = (int)sc->recv_io.credits.target - atomic_read(&sc->recv_io.posted.count);
+	while (posted < missing) {
+		struct smbdirect_recv_io *recv_io;
+		int ret;
+
+		/*
+		 * It's ok if smbdirect_connection_get_recv_io()
+		 * returns NULL, it means smbdirect_recv_io structures
+		 * are still be in the reassembly.list.
+		 */
+		recv_io = smbdirect_connection_get_recv_io(sc);
+		if (!recv_io)
+			break;
+
+		recv_io->first_segment = false;
+
+		ret = smbdirect_connection_post_recv_io(recv_io);
+		if (ret) {
+			smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
+				"smbdirect_connection_post_recv_io failed rc=%d (%1pe)\n",
+				ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
+			smbdirect_connection_put_recv_io(recv_io);
+			return ret;
+		}
+
+		atomic_inc(&sc->recv_io.posted.count);
+		posted += 1;
+	}
+
+	/* If nothing was posted we're done */
+	if (posted == 0)
+		return 0;
+
+	atomic_add(posted, &sc->recv_io.credits.available);
+
+	/*
+	 * If the last send credit is waiting for credits
+	 * it can grant we need to wake it up
+	 */
+	if (atomic_read(&sc->send_io.bcredits.count) == 0 &&
+	    atomic_read(&sc->send_io.credits.count) == 0)
+		wake_up(&sc->send_io.credits.wait_queue);
+
+	/*
+	 * If we posted at least one smbdirect_recv_io buffer,
+	 * we need to inform the peer about it and grant
+	 * additional credits.
+	 *
+	 * However there is one case where we don't want to
+	 * do that.
+	 *
+	 * If only a single credit was missing before
+	 * reaching the requested target, we should not
+	 * post an immediate send, as that would cause
+	 * endless ping pong once a keep alive exchange
+	 * is started.
+	 *
+	 * However if sc->recv_io.credits.target is only 1,
+	 * the peer has no credit left and we need to
+	 * grant the credit anyway.
+	 */
+	if (missing == 1 && sc->recv_io.credits.target != 1)
+		return 0;
+
+	return posted;
+}
+
+static void smbdirect_connection_recv_io_refill_work(struct work_struct *work)
+{
+	struct smbdirect_socket *sc =
+		container_of(work, struct smbdirect_socket, recv_io.posted.refill_work);
+	int posted;
+
+	posted = smbdirect_connection_recv_io_refill(sc);
+	if (unlikely(posted < 0)) {
+		smbdirect_socket_schedule_cleanup(sc, posted);
+		return;
+	}
+	if (posted > 0) {
+		smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
+			"schedule send of an empty message\n");
+		queue_work(sc->workqueues.immediate, &sc->idle.immediate_work);
+	}
+}
+
+int smbdirect_connection_recvmsg(struct smbdirect_socket *sc,
+				 struct msghdr *msg,
+				 unsigned int flags)
+{
+	struct smbdirect_recv_io *response;
+	struct smbdirect_data_transfer *data_transfer;
+	size_t size = iov_iter_count(&msg->msg_iter);
+	int to_copy, to_read, data_read, offset;
+	u32 data_length, remaining_data_length, data_offset;
+	int ret;
+
+	if (WARN_ONCE(flags, "unexpected flags=0x%x\n", flags))
+		return -EINVAL; /* no flags support for now */
+
+	if (WARN_ON_ONCE(iov_iter_rw(&msg->msg_iter) != ITER_DEST))
+		return -EINVAL; /* It's a bug in upper layer to get there */
+
+again:
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
+		smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+			"status=%s first_error=%1pe => %1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+			SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
+		return -ENOTCONN;
+	}
+
+	/*
+	 * No need to hold the reassembly queue lock all the time as we are
+	 * the only one reading from the front of the queue. The transport
+	 * may add more entries to the back of the queue at the same time
+	 */
+	smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+		"size=%zd sc->recv_io.reassembly.data_length=%d\n",
+		size, sc->recv_io.reassembly.data_length);
+	if (sc->recv_io.reassembly.data_length >= size) {
+		int queue_length;
+		int queue_removed = 0;
+		unsigned long flags;
+
+		/*
+		 * Need to make sure reassembly_data_length is read before
+		 * reading reassembly_queue_length and calling
+		 * smbdirect_connection_reassembly_first_recv_io. This call is lock free
+		 * as we never read at the end of the queue which are being
+		 * updated in SOFTIRQ as more data is received
+		 */
+		virt_rmb();
+		queue_length = sc->recv_io.reassembly.queue_length;
+		data_read = 0;
+		to_read = size;
+		offset = sc->recv_io.reassembly.first_entry_offset;
+		while (data_read < size) {
+			response = smbdirect_connection_reassembly_first_recv_io(sc);
+			data_transfer = (void *)response->packet;
+			data_length = le32_to_cpu(data_transfer->data_length);
+			remaining_data_length =
+				le32_to_cpu(
+					data_transfer->remaining_data_length);
+			data_offset = le32_to_cpu(data_transfer->data_offset);
+
+			/*
+			 * The upper layer expects RFC1002 length at the
+			 * beginning of the payload. Return it to indicate
+			 * the total length of the packet. This minimize the
+			 * change to upper layer packet processing logic. This
+			 * will be eventually remove when an intermediate
+			 * transport layer is added
+			 */
+			if (response->first_segment && size == 4) {
+				unsigned int rfc1002_len =
+					data_length + remaining_data_length;
+				__be32 rfc1002_hdr = cpu_to_be32(rfc1002_len);
+
+				if (copy_to_iter(&rfc1002_hdr, sizeof(rfc1002_hdr),
+						 &msg->msg_iter) != sizeof(rfc1002_hdr))
+					return -EFAULT;
+				data_read = 4;
+				response->first_segment = false;
+				smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+					"returning rfc1002 length %d\n",
+					rfc1002_len);
+				goto read_rfc1002_done;
+			}
+
+			to_copy = min_t(int, data_length - offset, to_read);
+			if (copy_to_iter((u8 *)data_transfer + data_offset + offset,
+					 to_copy, &msg->msg_iter) != to_copy)
+				return -EFAULT;
+
+			/* move on to the next buffer? */
+			if (to_copy == data_length - offset) {
+				queue_length--;
+				/*
+				 * No need to lock if we are not at the
+				 * end of the queue
+				 */
+				if (queue_length)
+					list_del(&response->list);
+				else {
+					spin_lock_irqsave(
+						&sc->recv_io.reassembly.lock, flags);
+					list_del(&response->list);
+					spin_unlock_irqrestore(
+						&sc->recv_io.reassembly.lock, flags);
+				}
+				queue_removed++;
+				sc->statistics.dequeue_reassembly_queue++;
+				smbdirect_connection_put_recv_io(response);
+				offset = 0;
+				smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+					"smbdirect_connection_put_recv_io offset=0\n");
+			} else
+				offset += to_copy;
+
+			to_read -= to_copy;
+			data_read += to_copy;
+
+			smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+				 "memcpy %d bytes len-ofs=%u => todo=%u done=%u ofs=%u\n",
+				 to_copy, data_length - offset,
+				 to_read, data_read, offset);
+		}
+
+		spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
+		sc->recv_io.reassembly.data_length -= data_read;
+		sc->recv_io.reassembly.queue_length -= queue_removed;
+		spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
+
+		sc->recv_io.reassembly.first_entry_offset = offset;
+		smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+			 "returning data_read=%d reassembly_length=%d first_ofs=%u\n",
+			 data_read, sc->recv_io.reassembly.data_length,
+			 sc->recv_io.reassembly.first_entry_offset);
+read_rfc1002_done:
+		return data_read;
+	}
+
+	smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
+		"wait_event on more data\n");
+	ret = wait_event_interruptible(sc->recv_io.reassembly.wait_queue,
+				       sc->recv_io.reassembly.data_length >= size ||
+				       sc->status != SMBDIRECT_SOCKET_CONNECTED);
+	/* Don't return any data if interrupted */
+	if (ret)
+		return ret;
+
+	goto again;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_recvmsg);
+
+static bool smbdirect_map_sges_single_page(struct smbdirect_map_sges *state,
+					   struct page *page, size_t off, size_t len)
+{
+	struct ib_sge *sge;
+	u64 addr;
+
+	if (state->num_sge >= state->max_sge)
+		return false;
+
+	addr = ib_dma_map_page(state->device, page,
+			       off, len, state->direction);
+	if (ib_dma_mapping_error(state->device, addr))
+		return false;
+
+	sge = &state->sge[state->num_sge++];
+	sge->addr   = addr;
+	sge->length = len;
+	sge->lkey   = state->local_dma_lkey;
+
+	return true;
+}
+
+/*
+ * Extract page fragments from a BVEC-class iterator and add them to an ib_sge
+ * list.  The pages are not pinned.
+ */
+static ssize_t smbdirect_map_sges_from_bvec(struct iov_iter *iter,
+					    struct smbdirect_map_sges *state,
+					    ssize_t maxsize)
+{
+	const struct bio_vec *bv = iter->bvec;
+	unsigned long start = iter->iov_offset;
+	unsigned int i;
+	ssize_t ret = 0;
+
+	for (i = 0; i < iter->nr_segs; i++) {
+		size_t off, len;
+		bool ok;
+
+		len = bv[i].bv_len;
+		if (start >= len) {
+			start -= len;
+			continue;
+		}
+
+		len = min_t(size_t, maxsize, len - start);
+		off = bv[i].bv_offset + start;
+
+		ok = smbdirect_map_sges_single_page(state,
+						    bv[i].bv_page,
+						    off,
+						    len);
+		if (!ok)
+			return -EIO;
+
+		ret += len;
+		maxsize -= len;
+		if (state->num_sge >= state->max_sge || maxsize <= 0)
+			break;
+		start = 0;
+	}
+
+	if (ret > 0)
+		iov_iter_advance(iter, ret);
+	return ret;
+}
+
+/*
+ * Extract fragments from a KVEC-class iterator and add them to an ib_sge list.
+ * This can deal with vmalloc'd buffers as well as kmalloc'd or static buffers.
+ * The pages are not pinned.
+ */
+static ssize_t smbdirect_map_sges_from_kvec(struct iov_iter *iter,
+					    struct smbdirect_map_sges *state,
+					    ssize_t maxsize)
+{
+	const struct kvec *kv = iter->kvec;
+	unsigned long start = iter->iov_offset;
+	unsigned int i;
+	ssize_t ret = 0;
+
+	for (i = 0; i < iter->nr_segs; i++) {
+		struct page *page;
+		unsigned long kaddr;
+		size_t off, len, seg;
+
+		len = kv[i].iov_len;
+		if (start >= len) {
+			start -= len;
+			continue;
+		}
+
+		kaddr = (unsigned long)kv[i].iov_base + start;
+		off = kaddr & ~PAGE_MASK;
+		len = min_t(size_t, maxsize, len - start);
+		kaddr &= PAGE_MASK;
+
+		maxsize -= len;
+		do {
+			bool ok;
+
+			seg = min_t(size_t, len, PAGE_SIZE - off);
+
+			if (is_vmalloc_or_module_addr((void *)kaddr))
+				page = vmalloc_to_page((void *)kaddr);
+			else
+				page = virt_to_page((void *)kaddr);
+
+			ok = smbdirect_map_sges_single_page(state, page, off, seg);
+			if (!ok)
+				return -EIO;
+
+			ret += seg;
+			len -= seg;
+			kaddr += PAGE_SIZE;
+			off = 0;
+		} while (len > 0 && state->num_sge < state->max_sge);
+
+		if (state->num_sge >= state->max_sge || maxsize <= 0)
+			break;
+		start = 0;
+	}
+
+	if (ret > 0)
+		iov_iter_advance(iter, ret);
+	return ret;
+}
+
+/*
+ * Extract folio fragments from a FOLIOQ-class iterator and add them to an
+ * ib_sge list.  The folios are not pinned.
+ */
+static ssize_t smbdirect_map_sges_from_folioq(struct iov_iter *iter,
+					      struct smbdirect_map_sges *state,
+					      ssize_t maxsize)
+{
+	const struct folio_queue *folioq = iter->folioq;
+	unsigned int slot = iter->folioq_slot;
+	ssize_t ret = 0;
+	size_t offset = iter->iov_offset;
+
+	if (WARN_ON_ONCE(!folioq))
+		return -EIO;
+
+	if (slot >= folioq_nr_slots(folioq)) {
+		folioq = folioq->next;
+		if (WARN_ON_ONCE(!folioq))
+			return -EIO;
+		slot = 0;
+	}
+
+	do {
+		struct folio *folio = folioq_folio(folioq, slot);
+		size_t fsize = folioq_folio_size(folioq, slot);
+
+		if (offset < fsize) {
+			size_t part = umin(maxsize, fsize - offset);
+			bool ok;
+
+			ok = smbdirect_map_sges_single_page(state,
+							    folio_page(folio, 0),
+							    offset,
+							    part);
+			if (!ok)
+				return -EIO;
+
+			offset += part;
+			ret += part;
+			maxsize -= part;
+		}
+
+		if (offset >= fsize) {
+			offset = 0;
+			slot++;
+			if (slot >= folioq_nr_slots(folioq)) {
+				if (!folioq->next) {
+					WARN_ON_ONCE(ret < iter->count);
+					break;
+				}
+				folioq = folioq->next;
+				slot = 0;
+			}
+		}
+	} while (state->num_sge < state->max_sge && maxsize > 0);
+
+	iter->folioq = folioq;
+	iter->folioq_slot = slot;
+	iter->iov_offset = offset;
+	iter->count -= ret;
+	return ret;
+}
+
+/*
+ * Extract page fragments from up to the given amount of the source iterator
+ * and build up an ib_sge list that refers to all of those bits.  The ib_sge list
+ * is appended to, up to the maximum number of elements set in the parameter
+ * block.
+ *
+ * The extracted page fragments are not pinned or ref'd in any way; if an
+ * IOVEC/UBUF-type iterator is to be used, it should be converted to a
+ * BVEC-type iterator and the pages pinned, ref'd or otherwise held in some
+ * way.
+ */
+static ssize_t smbdirect_map_sges_from_iter(struct iov_iter *iter, size_t len,
+					    struct smbdirect_map_sges *state)
+{
+	ssize_t ret;
+	size_t before = state->num_sge;
+
+	if (WARN_ON_ONCE(iov_iter_rw(iter) != ITER_SOURCE))
+		return -EIO;
+
+	switch (iov_iter_type(iter)) {
+	case ITER_BVEC:
+		ret = smbdirect_map_sges_from_bvec(iter, state, len);
+		break;
+	case ITER_KVEC:
+		ret = smbdirect_map_sges_from_kvec(iter, state, len);
+		break;
+	case ITER_FOLIOQ:
+		ret = smbdirect_map_sges_from_folioq(iter, state, len);
+		break;
+	default:
+		WARN_ONCE(1, "iov_iter_type[%u]\n", iov_iter_type(iter));
+		return -EIO;
+	}
+
+	if (ret < 0) {
+		while (state->num_sge > before) {
+			struct ib_sge *sge = &state->sge[--state->num_sge];
+
+			ib_dma_unmap_page(state->device,
+					  sge->addr,
+					  sge->length,
+					  state->direction);
+		}
+	}
+
+	return ret;
+}
diff --git a/fs/smb/smbdirect/debug.c b/fs/smb/smbdirect/debug.c
new file mode 100644
index 000000000000..3445843445bf
--- /dev/null
+++ b/fs/smb/smbdirect/debug.c
@@ -0,0 +1,88 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+#include <linux/seq_file.h>
+
+void smbdirect_connection_legacy_debug_proc_show(struct smbdirect_socket *sc,
+						 unsigned int rdma_readwrite_threshold,
+						 struct seq_file *m)
+{
+	const struct smbdirect_socket_parameters *sp;
+
+	if (!sc)
+		return;
+	sp = &sc->parameters;
+
+	seq_puts(m, "\n");
+	seq_printf(m, "SMBDirect protocol version: 0x%x ",
+		   SMBDIRECT_V1);
+	seq_printf(m, "transport status: %s (%u)",
+		   smbdirect_socket_status_string(sc->status),
+		   sc->status);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Conn receive_credit_max: %u ",
+		   sp->recv_credit_max);
+	seq_printf(m, "send_credit_target: %u max_send_size: %u",
+		   sp->send_credit_target,
+		   sp->max_send_size);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Conn max_fragmented_recv_size: %u ",
+		   sp->max_fragmented_recv_size);
+	seq_printf(m, "max_fragmented_send_size: %u max_receive_size:%u",
+		   sp->max_fragmented_send_size,
+		   sp->max_recv_size);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Conn keep_alive_interval: %u ",
+		   sp->keepalive_interval_msec / 1000);
+	seq_printf(m, "max_readwrite_size: %u rdma_readwrite_threshold: %u",
+		   sp->max_read_write_size,
+		   rdma_readwrite_threshold);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Debug count_get_receive_buffer: %llu ",
+		   sc->statistics.get_receive_buffer);
+	seq_printf(m, "count_put_receive_buffer: %llu count_send_empty: %llu",
+		   sc->statistics.put_receive_buffer,
+		   sc->statistics.send_empty);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Read Queue count_enqueue_reassembly_queue: %llu ",
+		   sc->statistics.enqueue_reassembly_queue);
+	seq_printf(m, "count_dequeue_reassembly_queue: %llu ",
+		   sc->statistics.dequeue_reassembly_queue);
+	seq_printf(m, "reassembly_data_length: %u ",
+		   sc->recv_io.reassembly.data_length);
+	seq_printf(m, "reassembly_queue_length: %u",
+		   sc->recv_io.reassembly.queue_length);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Current Credits send_credits: %u ",
+		   atomic_read(&sc->send_io.credits.count));
+	seq_printf(m, "receive_credits: %u receive_credit_target: %u",
+		   atomic_read(&sc->recv_io.credits.count),
+		   sc->recv_io.credits.target);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "Pending send_pending: %u ",
+		   atomic_read(&sc->send_io.pending.count));
+
+	seq_puts(m, "\n");
+	seq_printf(m, "MR responder_resources: %u ",
+		   sp->responder_resources);
+	seq_printf(m, "max_frmr_depth: %u mr_type: 0x%x",
+		   sp->max_frmr_depth,
+		   sc->mr_io.type);
+
+	seq_puts(m, "\n");
+	seq_printf(m, "MR mr_ready_count: %u mr_used_count: %u",
+		   atomic_read(&sc->mr_io.ready.count),
+		   atomic_read(&sc->mr_io.used.count));
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_legacy_debug_proc_show);
diff --git a/fs/smb/smbdirect/devices.c b/fs/smb/smbdirect/devices.c
new file mode 100644
index 000000000000..7adacbdfe12e
--- /dev/null
+++ b/fs/smb/smbdirect/devices.c
@@ -0,0 +1,277 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (C) 2018, LG Electronics.
+ *   Copyright (c) 2025 Stefan Metzmacher
+ */
+
+#include "internal.h"
+
+static u8 smbdirect_ib_device_rdma_capable_node_type(struct ib_device *ib_dev)
+{
+	if (!smbdirect_frwr_is_supported(&ib_dev->attrs))
+		return RDMA_NODE_UNSPECIFIED;
+
+	switch (ib_dev->node_type) {
+	case RDMA_NODE_IB_CA: /* Infiniband, RoCE v1 and v2 */
+	case RDMA_NODE_RNIC:  /* iWarp */
+		return ib_dev->node_type;
+	}
+
+	return RDMA_NODE_UNSPECIFIED;
+}
+
+static int smbdirect_ib_client_add(struct ib_device *ib_dev)
+{
+	u8 node_type = smbdirect_ib_device_rdma_capable_node_type(ib_dev);
+	struct smbdirect_device *sdev;
+	const char *node_str;
+	const char *action;
+	u32 pidx;
+
+	switch (node_type) {
+	case RDMA_NODE_IB_CA:
+		node_str = "IB_CA";
+		action = "added";
+		break;
+	case RDMA_NODE_RNIC:
+		node_str = "RNIC";
+		action = "added";
+		break;
+	case RDMA_NODE_UNSPECIFIED:
+		node_str = "UNSPECIFIED";
+		action = "ignored";
+		break;
+	default:
+		node_str = "UNKNOWN";
+		action = "ignored";
+		node_type = RDMA_NODE_UNSPECIFIED;
+		break;
+	}
+
+	pr_info("ib_dev[%.*s]: %s: %s %s=%u %s=0x%llx %s=0x%llx %s=0x%llx\n",
+		IB_DEVICE_NAME_MAX,
+		ib_dev->name,
+		action,
+		node_str,
+		"max_fast_reg_page_list_len",
+		ib_dev->attrs.max_fast_reg_page_list_len,
+		"device_cap_flags",
+		ib_dev->attrs.device_cap_flags,
+		"kernel_cap_flags",
+		ib_dev->attrs.kernel_cap_flags,
+		"page_size_cap",
+		ib_dev->attrs.page_size_cap);
+
+	if (node_type == RDMA_NODE_UNSPECIFIED)
+		return 0;
+
+	pr_info("ib_dev[%.*s]: %s=%u %s=%u %s=%u %s=%u %s=%u %s=%u %s=%u %s=%u %s=%u\n",
+		IB_DEVICE_NAME_MAX,
+		ib_dev->name,
+		"num_ports",
+		rdma_end_port(ib_dev),
+		"max_qp_rd_atom",
+		ib_dev->attrs.max_qp_rd_atom,
+		"max_qp_init_rd_atom",
+		ib_dev->attrs.max_qp_init_rd_atom,
+		"max_sgl_rd",
+		ib_dev->attrs.max_sgl_rd,
+		"max_sge_rd",
+		ib_dev->attrs.max_sge_rd,
+		"max_cqe",
+		ib_dev->attrs.max_cqe,
+		"max_qp_wr",
+		ib_dev->attrs.max_qp_wr,
+		"max_send_sge",
+		ib_dev->attrs.max_send_sge,
+		"max_recv_sge",
+		ib_dev->attrs.max_recv_sge);
+
+	rdma_for_each_port(ib_dev, pidx) {
+		const struct ib_port_immutable *ib_pi =
+			ib_port_immutable_read(ib_dev, pidx);
+		u32 core_cap_flags = ib_pi ? ib_pi->core_cap_flags : 0;
+
+		pr_info("ib_dev[%.*s]PORT[%u]: %s=%u %s=%u %s=%u %s=%u %s=%u %s=0x%x\n",
+			IB_DEVICE_NAME_MAX,
+			ib_dev->name,
+			pidx,
+			"iwarp",
+			rdma_protocol_iwarp(ib_dev, pidx),
+			"ib",
+			rdma_protocol_ib(ib_dev, pidx),
+			"roce",
+			rdma_protocol_roce(ib_dev, pidx),
+			"v1",
+			rdma_protocol_roce_eth_encap(ib_dev, pidx),
+			"v2",
+			rdma_protocol_roce_udp_encap(ib_dev, pidx),
+			"core_cap_flags",
+			core_cap_flags);
+	}
+
+	sdev = kzalloc_obj(*sdev);
+	if (!sdev)
+		return -ENOMEM;
+	sdev->ib_dev = ib_dev;
+	snprintf(sdev->ib_name, ARRAY_SIZE(sdev->ib_name), "%.*s",
+		 IB_DEVICE_NAME_MAX, ib_dev->name);
+
+	write_lock(&smbdirect_globals.devices.lock);
+	list_add(&sdev->list, &smbdirect_globals.devices.list);
+	write_unlock(&smbdirect_globals.devices.lock);
+
+	return 0;
+}
+
+static void smbdirect_ib_client_remove(struct ib_device *ib_dev, void *client_data)
+{
+	struct smbdirect_device *sdev, *tmp;
+
+	write_lock(&smbdirect_globals.devices.lock);
+	list_for_each_entry_safe(sdev, tmp, &smbdirect_globals.devices.list, list) {
+		if (sdev->ib_dev == ib_dev) {
+			list_del(&sdev->list);
+			pr_info("ib_dev[%.*s] removed\n",
+				IB_DEVICE_NAME_MAX, sdev->ib_name);
+			kfree(sdev);
+			break;
+		}
+	}
+	write_unlock(&smbdirect_globals.devices.lock);
+}
+
+static void smbdirect_ib_client_rename(struct ib_device *ib_dev, void *client_data)
+{
+	struct smbdirect_device *sdev;
+
+	write_lock(&smbdirect_globals.devices.lock);
+	list_for_each_entry(sdev, &smbdirect_globals.devices.list, list) {
+		if (sdev->ib_dev == ib_dev) {
+			pr_info("ib_dev[%.*s] renamed to [%.*s]\n",
+				IB_DEVICE_NAME_MAX, sdev->ib_name,
+				IB_DEVICE_NAME_MAX, ib_dev->name);
+			snprintf(sdev->ib_name, ARRAY_SIZE(sdev->ib_name), "%.*s",
+				 IB_DEVICE_NAME_MAX, ib_dev->name);
+			break;
+		}
+	}
+	write_unlock(&smbdirect_globals.devices.lock);
+}
+
+static struct ib_client smbdirect_ib_client = {
+	.name	= "smbdirect_ib_client",
+	.add	= smbdirect_ib_client_add,
+	.remove	= smbdirect_ib_client_remove,
+	.rename	= smbdirect_ib_client_rename,
+};
+
+static u8 smbdirect_netdev_find_rdma_capable_node_type(struct net_device *netdev)
+{
+	struct smbdirect_device *sdev;
+	u8 node_type = RDMA_NODE_UNSPECIFIED;
+
+	read_lock(&smbdirect_globals.devices.lock);
+	list_for_each_entry(sdev, &smbdirect_globals.devices.list, list) {
+		u32 pi;
+
+		rdma_for_each_port(sdev->ib_dev, pi) {
+			struct net_device *ndev;
+
+			ndev = ib_device_get_netdev(sdev->ib_dev, pi);
+			if (!ndev)
+				continue;
+
+			if (ndev == netdev) {
+				dev_put(ndev);
+				node_type = sdev->ib_dev->node_type;
+				goto out;
+			}
+			dev_put(ndev);
+		}
+	}
+out:
+	read_unlock(&smbdirect_globals.devices.lock);
+
+	if (node_type == RDMA_NODE_UNSPECIFIED) {
+		struct ib_device *ibdev;
+
+		ibdev = ib_device_get_by_netdev(netdev, RDMA_DRIVER_UNKNOWN);
+		if (ibdev) {
+			node_type = smbdirect_ib_device_rdma_capable_node_type(ibdev);
+			ib_device_put(ibdev);
+		}
+	}
+
+	return node_type;
+}
+
+/*
+ * Returns RDMA_NODE_UNSPECIFIED when the netdev has
+ * no support for smbdirect capable rdma.
+ *
+ * Otherwise RDMA_NODE_RNIC is returned for iwarp devices
+ * and RDMA_NODE_IB_CA or Infiniband and RoCE (v1 and v2)
+ */
+u8 smbdirect_netdev_rdma_capable_node_type(struct net_device *netdev)
+{
+	struct net_device *lower_dev;
+	struct list_head *iter;
+	u8 node_type = RDMA_NODE_UNSPECIFIED;
+
+	node_type = smbdirect_netdev_find_rdma_capable_node_type(netdev);
+	if (node_type != RDMA_NODE_UNSPECIFIED)
+		return node_type;
+
+	/* check if netdev is bridge or VLAN */
+	if (netif_is_bridge_master(netdev) || netdev->priv_flags & IFF_802_1Q_VLAN)
+		netdev_for_each_lower_dev(netdev, lower_dev, iter) {
+			node_type = smbdirect_netdev_find_rdma_capable_node_type(lower_dev);
+			if (node_type != RDMA_NODE_UNSPECIFIED)
+				return node_type;
+		}
+
+	/* check if netdev is IPoIB safely without layer violation */
+	if (netdev->type == ARPHRD_INFINIBAND)
+		return RDMA_NODE_IB_CA;
+
+	return RDMA_NODE_UNSPECIFIED;
+}
+EXPORT_SYMBOL_GPL(smbdirect_netdev_rdma_capable_node_type);
+
+__init int smbdirect_devices_init(void)
+{
+	int ret;
+
+	rwlock_init(&smbdirect_globals.devices.lock);
+	INIT_LIST_HEAD(&smbdirect_globals.devices.list);
+
+	ret = ib_register_client(&smbdirect_ib_client);
+	if (ret) {
+		pr_crit("failed to ib_register_client: %d %1pe\n",
+			ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	return 0;
+}
+
+__exit void smbdirect_devices_exit(void)
+{
+	struct smbdirect_device *sdev, *tmp;
+
+	/*
+	 * On exist we just cleanup so that
+	 * smbdirect_ib_client_remove() won't
+	 * print removals of devices.
+	 */
+	write_lock(&smbdirect_globals.devices.lock);
+	list_for_each_entry_safe(sdev, tmp, &smbdirect_globals.devices.list, list) {
+		list_del(&sdev->list);
+		kfree(sdev);
+	}
+	write_unlock(&smbdirect_globals.devices.lock);
+
+	ib_unregister_client(&smbdirect_ib_client);
+}
diff --git a/fs/smb/smbdirect/internal.h b/fs/smb/smbdirect/internal.h
new file mode 100644
index 000000000000..e9959e6dc13a
--- /dev/null
+++ b/fs/smb/smbdirect/internal.h
@@ -0,0 +1,141 @@
+/* SPDX-License-Identifier: GPL-2.0-or-later */
+/*
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#ifndef __FS_SMB_COMMON_SMBDIRECT_INTERNAL_H__
+#define __FS_SMB_COMMON_SMBDIRECT_INTERNAL_H__
+
+#define DEFAULT_SYMBOL_NAMESPACE "SMBDIRECT"
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include <linux/smbdirect.h>
+#include "pdu.h"
+
+#include <linux/mutex.h>
+
+struct smbdirect_module_state {
+	struct mutex mutex;
+
+	struct {
+		struct workqueue_struct *accept;
+		struct workqueue_struct *connect;
+		struct workqueue_struct *idle;
+		struct workqueue_struct *refill;
+		struct workqueue_struct *immediate;
+		struct workqueue_struct *cleanup;
+	} workqueues;
+
+	struct {
+		rwlock_t lock;
+		struct list_head list;
+	} devices;
+};
+
+extern struct smbdirect_module_state smbdirect_globals;
+
+#include "socket.h"
+
+struct smbdirect_device {
+	struct list_head list;
+	struct ib_device *ib_dev;
+	/*
+	 * copy of ib_dev->name,
+	 * in order to print renames
+	 */
+	char ib_name[IB_DEVICE_NAME_MAX];
+};
+
+int smbdirect_socket_init_new(struct net *net, struct smbdirect_socket *sc);
+
+int smbdirect_socket_init_accepting(struct rdma_cm_id *id, struct smbdirect_socket *sc);
+
+void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
+					 const char *macro_name,
+					 unsigned int lvl,
+					 const char *func,
+					 unsigned int line,
+					 int error,
+					 enum smbdirect_socket_status *force_status);
+#define smbdirect_socket_schedule_cleanup(__sc, __error) \
+	__smbdirect_socket_schedule_cleanup(__sc, \
+		"smbdirect_socket_schedule_cleanup", SMBDIRECT_LOG_ERR, \
+		__func__, __LINE__, __error, NULL)
+#define smbdirect_socket_schedule_cleanup_lvl(__sc, __lvl, __error) \
+	__smbdirect_socket_schedule_cleanup(__sc, \
+		"smbdirect_socket_schedule_cleanup_lvl", __lvl, \
+		__func__, __LINE__, __error, NULL)
+#define smbdirect_socket_schedule_cleanup_status(__sc, __lvl, __error, __status) do { \
+	enum smbdirect_socket_status __force_status = __status; \
+	__smbdirect_socket_schedule_cleanup(__sc, \
+		"smbdirect_socket_schedule_cleanup_status", __lvl, \
+		__func__, __LINE__, __error, &__force_status); \
+} while (0)
+
+void smbdirect_socket_destroy_sync(struct smbdirect_socket *sc);
+
+int smbdirect_socket_wait_for_credits(struct smbdirect_socket *sc,
+				      enum smbdirect_socket_status expected_status,
+				      int unexpected_errno,
+				      wait_queue_head_t *waitq,
+				      atomic_t *total_credits,
+				      int needed);
+
+void smbdirect_connection_rdma_established(struct smbdirect_socket *sc);
+
+void smbdirect_connection_negotiation_done(struct smbdirect_socket *sc);
+
+int smbdirect_connection_create_qp(struct smbdirect_socket *sc);
+
+void smbdirect_connection_destroy_qp(struct smbdirect_socket *sc);
+
+int smbdirect_connection_create_mem_pools(struct smbdirect_socket *sc);
+
+void smbdirect_connection_destroy_mem_pools(struct smbdirect_socket *sc);
+
+struct smbdirect_send_io *smbdirect_connection_alloc_send_io(struct smbdirect_socket *sc);
+
+void smbdirect_connection_free_send_io(struct smbdirect_send_io *msg);
+
+struct smbdirect_recv_io *smbdirect_connection_get_recv_io(struct smbdirect_socket *sc);
+
+void smbdirect_connection_put_recv_io(struct smbdirect_recv_io *msg);
+
+void smbdirect_connection_reassembly_append_recv_io(struct smbdirect_socket *sc,
+						    struct smbdirect_recv_io *msg,
+						    u32 data_length);
+
+struct smbdirect_recv_io *
+smbdirect_connection_reassembly_first_recv_io(struct smbdirect_socket *sc);
+
+void smbdirect_connection_negotiate_rdma_resources(struct smbdirect_socket *sc,
+						   u8 peer_initiator_depth,
+						   u8 peer_responder_resources,
+						   const struct rdma_conn_param *param);
+
+void smbdirect_connection_idle_timer_work(struct work_struct *work);
+
+u16 smbdirect_connection_grant_recv_credits(struct smbdirect_socket *sc);
+
+int smbdirect_connection_post_send_wr(struct smbdirect_socket *sc,
+				      struct ib_send_wr *wr);
+
+int smbdirect_connection_post_recv_io(struct smbdirect_recv_io *msg);
+
+void smbdirect_connection_recv_io_done(struct ib_cq *cq, struct ib_wc *wc);
+
+int smbdirect_connection_recv_io_refill(struct smbdirect_socket *sc);
+
+int smbdirect_connection_create_mr_list(struct smbdirect_socket *sc);
+
+void smbdirect_connection_destroy_mr_list(struct smbdirect_socket *sc);
+
+int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
+				     const struct rdma_conn_param *param);
+
+void smbdirect_accept_negotiate_finish(struct smbdirect_socket *sc, u32 ntstatus);
+
+__init int smbdirect_devices_init(void);
+__exit void smbdirect_devices_exit(void);
+
+#endif /* __FS_SMB_COMMON_SMBDIRECT_INTERNAL_H__ */
diff --git a/fs/smb/smbdirect/listen.c b/fs/smb/smbdirect/listen.c
new file mode 100644
index 000000000000..2f78bcaedbf8
--- /dev/null
+++ b/fs/smb/smbdirect/listen.c
@@ -0,0 +1,308 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (C) 2018, LG Electronics.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+
+static int smbdirect_listen_rdma_event_handler(struct rdma_cm_id *id,
+					       struct rdma_cm_event *event);
+
+int smbdirect_socket_listen(struct smbdirect_socket *sc, int backlog)
+{
+	int ret;
+
+	if (backlog < 0)
+		return -EINVAL;
+	if (!backlog)
+		backlog = 1; /* use 1 as default for now */
+
+	if (sc->first_error)
+		return -EINVAL;
+
+	if (sc->status != SMBDIRECT_SOCKET_CREATED)
+		return -EINVAL;
+
+	if (WARN_ON_ONCE(!sc->rdma.cm_id))
+		return -EINVAL;
+
+	if (sc->rdma.cm_id->device)
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"try to listen on addr: %pISpsfc dev: %.*s\n",
+			&sc->rdma.cm_id->route.addr.src_addr,
+			IB_DEVICE_NAME_MAX,
+			sc->rdma.cm_id->device->name);
+	else
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"try to listen on addr: %pISpsfc\n",
+			&sc->rdma.cm_id->route.addr.src_addr);
+
+	/* already checked above */
+	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED);
+	sc->status = SMBDIRECT_SOCKET_LISTENING;
+	sc->rdma.expected_event = RDMA_CM_EVENT_CONNECT_REQUEST;
+	rdma_lock_handler(sc->rdma.cm_id);
+	sc->rdma.cm_id->event_handler = smbdirect_listen_rdma_event_handler;
+	rdma_unlock_handler(sc->rdma.cm_id);
+
+	ret = rdma_listen(sc->rdma.cm_id, backlog);
+	if (ret) {
+		sc->first_error = ret;
+		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
+		if (sc->rdma.cm_id->device)
+			smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+				"listening failed %1pe on addr: %pISpsfc dev: %.*s\n",
+				SMBDIRECT_DEBUG_ERR_PTR(ret),
+				&sc->rdma.cm_id->route.addr.src_addr,
+				IB_DEVICE_NAME_MAX,
+				sc->rdma.cm_id->device->name);
+		else
+			smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+				"listening failed %1pe on addr: %pISpsfc\n",
+				SMBDIRECT_DEBUG_ERR_PTR(ret),
+				&sc->rdma.cm_id->route.addr.src_addr);
+		return ret;
+	}
+
+	/*
+	 * This is a value > 0, checked above,
+	 * so we are able to use sc->listen.backlog == -1,
+	 * as indication that the socket was never
+	 * a listener.
+	 */
+	sc->listen.backlog = backlog;
+
+	if (sc->rdma.cm_id->device)
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"listening on addr: %pISpsfc dev: %.*s\n",
+			&sc->rdma.cm_id->route.addr.src_addr,
+			IB_DEVICE_NAME_MAX,
+			sc->rdma.cm_id->device->name);
+	else
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"listening on addr: %pISpsfc\n",
+			&sc->rdma.cm_id->route.addr.src_addr);
+
+	/*
+	 * The rest happens async via smbdirect_listen_rdma_event_handler()
+	 */
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_listen);
+
+static int smbdirect_new_rdma_event_handler(struct rdma_cm_id *new_id,
+					    struct rdma_cm_event *event)
+{
+	int ret = -ESTALE;
+
+	/*
+	 * This should be replaced before any real work
+	 * starts! So it should never be called!
+	 */
+
+	if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+		ret = -ENETDOWN;
+	if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
+		ret = event->status;
+	WARN_ONCE(1,
+		  "%s should not be called! event=%s status=%d => ret=%1pe\n",
+		  __func__,
+		  rdma_event_msg(event->event),
+		  event->status,
+		  SMBDIRECT_DEBUG_ERR_PTR(ret));
+	return -ESTALE;
+}
+
+static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
+					    struct rdma_cm_id *new_id,
+					    const struct rdma_cm_event *event);
+
+static int smbdirect_listen_rdma_event_handler(struct rdma_cm_id *new_id,
+					       struct rdma_cm_event *event)
+{
+	struct smbdirect_socket *lsc = new_id->context;
+	int ret;
+
+	if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST) {
+		new_id->context = NULL;
+		new_id->event_handler = smbdirect_new_rdma_event_handler;
+	} else
+		new_id = NULL;
+
+	/*
+	 * cma_cm_event_handler() has
+	 * lockdep_assert_held(&id_priv->handler_mutex);
+	 *
+	 * Mutexes are not allowed in interrupts,
+	 * and we rely on not being in an interrupt here,
+	 * as we might sleep.
+	 */
+	WARN_ON_ONCE(in_interrupt());
+
+	if (event->status || event->event != lsc->rdma.expected_event) {
+		ret = -ECONNABORTED;
+
+		if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+			ret = -ENETDOWN;
+		if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
+			ret = event->status;
+
+		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
+			"%s (first_error=%1pe, expected=%s) => event=%s status=%d => ret=%1pe\n",
+			smbdirect_socket_status_string(lsc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(lsc->first_error),
+			rdma_event_msg(lsc->rdma.expected_event),
+			rdma_event_msg(event->event),
+			event->status,
+			SMBDIRECT_DEBUG_ERR_PTR(ret));
+
+		/*
+		 * In case of error return it and let the caller
+		 * destroy new_id
+		 */
+		smbdirect_socket_schedule_cleanup(lsc, ret);
+		return new_id ? ret : 0;
+	}
+
+	smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_INFO,
+		"%s (first_error=%1pe) event=%s\n",
+		smbdirect_socket_status_string(lsc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(lsc->first_error),
+		rdma_event_msg(event->event));
+
+	/*
+	 * In case of error return it and let the caller
+	 * destroy new_id
+	 */
+	if (lsc->first_error)
+		return new_id ? lsc->first_error : 0;
+
+	switch (event->event) {
+	case RDMA_CM_EVENT_CONNECT_REQUEST:
+		WARN_ON_ONCE(lsc->status != SMBDIRECT_SOCKET_LISTENING);
+
+		/*
+		 * In case of error return it and let the caller
+		 * destroy new_id
+		 */
+		ret = smbdirect_listen_connect_request(lsc, new_id, event);
+		if (ret)
+			return ret;
+		return 0;
+
+	default:
+		break;
+	}
+
+	/*
+	 * This is an internal error
+	 */
+	WARN_ON_ONCE(lsc->rdma.expected_event != RDMA_CM_EVENT_CONNECT_REQUEST);
+	smbdirect_socket_schedule_cleanup(lsc, -EINVAL);
+	return 0;
+}
+
+static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
+					    struct rdma_cm_id *new_id,
+					    const struct rdma_cm_event *event)
+{
+	const struct smbdirect_socket_parameters *lsp = &lsc->parameters;
+	struct smbdirect_socket *nsc;
+	unsigned long flags;
+	size_t backlog = max_t(size_t, 1, lsc->listen.backlog);
+	size_t psockets;
+	size_t rsockets;
+	int ret;
+
+	if (!smbdirect_frwr_is_supported(&new_id->device->attrs)) {
+		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
+			"Fast Registration Work Requests (FRWR) is not supported device %.*s\n",
+			IB_DEVICE_NAME_MAX,
+			new_id->device->name);
+		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
+			"Device capability flags = %llx max_fast_reg_page_list_len = %u\n",
+			new_id->device->attrs.device_cap_flags,
+			new_id->device->attrs.max_fast_reg_page_list_len);
+		return -EPROTONOSUPPORT;
+	}
+
+	if (lsp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IB &&
+	    !rdma_ib_or_roce(new_id->device, new_id->port_num)) {
+		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
+			"Not IB: device: %.*s IW:%u local: %pISpsfc remote: %pISpsfc\n",
+			IB_DEVICE_NAME_MAX,
+			new_id->device->name,
+			rdma_protocol_iwarp(new_id->device, new_id->port_num),
+			&new_id->route.addr.src_addr,
+			&new_id->route.addr.dst_addr);
+		return -EPROTONOSUPPORT;
+	}
+	if (lsp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IW &&
+	    !rdma_protocol_iwarp(new_id->device, new_id->port_num)) {
+		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
+			"Not IW: device: %.*s IB:%u local: %pISpsfc remote: %pISpsfc\n",
+			IB_DEVICE_NAME_MAX,
+			new_id->device->name,
+			rdma_ib_or_roce(new_id->device, new_id->port_num),
+			&new_id->route.addr.src_addr,
+			&new_id->route.addr.dst_addr);
+		return -EPROTONOSUPPORT;
+	}
+
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	psockets = list_count_nodes(&lsc->listen.pending);
+	rsockets = list_count_nodes(&lsc->listen.ready);
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+
+	if (psockets > backlog ||
+	    rsockets > backlog ||
+	    (psockets + rsockets) > backlog) {
+		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
+			"Backlog[%d][%zu] full pending[%zu] ready[%zu]\n",
+			lsc->listen.backlog, backlog, psockets, rsockets);
+		return -EBUSY;
+	}
+
+	ret = smbdirect_socket_create_accepting(new_id, &nsc);
+	if (ret)
+		goto socket_init_failed;
+
+	nsc->logging = lsc->logging;
+	ret = smbdirect_socket_set_initial_parameters(nsc, &lsc->parameters);
+	if (ret)
+		goto set_params_failed;
+	ret = smbdirect_socket_set_kernel_settings(nsc,
+						   lsc->ib.poll_ctx,
+						   lsc->send_io.mem.gfp_mask);
+	if (ret)
+		goto set_settings_failed;
+
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	list_add_tail(&nsc->accept.list, &lsc->listen.pending);
+	nsc->accept.listener = lsc;
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+
+	ret = smbdirect_accept_connect_request(nsc, &event->param.conn);
+	if (ret)
+		goto accept_connect_failed;
+
+	return 0;
+
+accept_connect_failed:
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	list_del_init(&nsc->accept.list);
+	nsc->accept.listener = NULL;
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+set_settings_failed:
+set_params_failed:
+	/*
+	 * The caller will destroy new_id
+	 */
+	nsc->ib.dev = NULL;
+	nsc->rdma.cm_id = NULL;
+	smbdirect_socket_release(nsc);
+socket_init_failed:
+	return ret;
+}
diff --git a/fs/smb/smbdirect/main.c b/fs/smb/smbdirect/main.c
new file mode 100644
index 000000000000..606732fefb69
--- /dev/null
+++ b/fs/smb/smbdirect/main.c
@@ -0,0 +1,121 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+#include <linux/module.h>
+
+struct smbdirect_module_state smbdirect_globals = {
+	.mutex = __MUTEX_INITIALIZER(smbdirect_globals.mutex),
+};
+
+static __init int smbdirect_module_init(void)
+{
+	int ret = -ENOMEM;
+
+	pr_notice("subsystem loading...\n");
+	mutex_lock(&smbdirect_globals.mutex);
+
+	smbdirect_globals.workqueues.accept = alloc_workqueue("smbdirect-accept",
+							      WQ_SYSFS |
+							      WQ_PERCPU |
+							      WQ_POWER_EFFICIENT,
+							      0);
+	if (smbdirect_globals.workqueues.accept == NULL)
+		goto alloc_accept_wq_failed;
+
+	smbdirect_globals.workqueues.connect = alloc_workqueue("smbdirect-connect",
+							       WQ_SYSFS |
+							       WQ_PERCPU |
+							       WQ_POWER_EFFICIENT,
+							       0);
+	if (smbdirect_globals.workqueues.connect == NULL)
+		goto alloc_connect_wq_failed;
+
+	smbdirect_globals.workqueues.idle = alloc_workqueue("smbdirect-idle",
+							    WQ_SYSFS |
+							    WQ_PERCPU |
+							    WQ_POWER_EFFICIENT,
+							    0);
+	if (smbdirect_globals.workqueues.idle == NULL)
+		goto alloc_idle_wq_failed;
+
+	smbdirect_globals.workqueues.refill = alloc_workqueue("smbdirect-refill",
+							      WQ_HIGHPRI |
+							      WQ_SYSFS |
+							      WQ_PERCPU |
+							      WQ_POWER_EFFICIENT,
+							      0);
+	if (smbdirect_globals.workqueues.refill == NULL)
+		goto alloc_refill_wq_failed;
+
+	smbdirect_globals.workqueues.immediate = alloc_workqueue("smbdirect-immediate",
+								 WQ_HIGHPRI |
+								 WQ_SYSFS |
+								 WQ_PERCPU |
+								 WQ_POWER_EFFICIENT,
+								 0);
+	if (smbdirect_globals.workqueues.immediate == NULL)
+		goto alloc_immediate_wq_failed;
+
+	smbdirect_globals.workqueues.cleanup = alloc_workqueue("smbdirect-cleanup",
+							       WQ_MEM_RECLAIM |
+							       WQ_HIGHPRI |
+							       WQ_SYSFS |
+							       WQ_PERCPU |
+							       WQ_POWER_EFFICIENT,
+							       0);
+	if (smbdirect_globals.workqueues.cleanup == NULL)
+		goto alloc_cleanup_wq_failed;
+
+	ret = smbdirect_devices_init();
+	if (ret)
+		goto devices_init_failed;
+
+	mutex_unlock(&smbdirect_globals.mutex);
+	pr_notice("subsystem loaded\n");
+	return 0;
+
+devices_init_failed:
+	destroy_workqueue(smbdirect_globals.workqueues.cleanup);
+alloc_cleanup_wq_failed:
+	destroy_workqueue(smbdirect_globals.workqueues.immediate);
+alloc_immediate_wq_failed:
+	destroy_workqueue(smbdirect_globals.workqueues.refill);
+alloc_refill_wq_failed:
+	destroy_workqueue(smbdirect_globals.workqueues.idle);
+alloc_idle_wq_failed:
+	destroy_workqueue(smbdirect_globals.workqueues.connect);
+alloc_connect_wq_failed:
+	destroy_workqueue(smbdirect_globals.workqueues.accept);
+alloc_accept_wq_failed:
+	mutex_unlock(&smbdirect_globals.mutex);
+	pr_crit("failed to loaded: %d (%1pe)\n",
+		ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
+	return ret;
+}
+
+static __exit void smbdirect_module_exit(void)
+{
+	pr_notice("subsystem unloading...\n");
+	mutex_lock(&smbdirect_globals.mutex);
+
+	smbdirect_devices_exit();
+
+	destroy_workqueue(smbdirect_globals.workqueues.accept);
+	destroy_workqueue(smbdirect_globals.workqueues.connect);
+	destroy_workqueue(smbdirect_globals.workqueues.idle);
+	destroy_workqueue(smbdirect_globals.workqueues.refill);
+	destroy_workqueue(smbdirect_globals.workqueues.immediate);
+	destroy_workqueue(smbdirect_globals.workqueues.cleanup);
+
+	mutex_unlock(&smbdirect_globals.mutex);
+	pr_notice("subsystem unloaded\n");
+}
+
+module_init(smbdirect_module_init);
+module_exit(smbdirect_module_exit);
+
+MODULE_DESCRIPTION("smbdirect subsystem");
+MODULE_LICENSE("GPL");
diff --git a/fs/smb/smbdirect/mr.c b/fs/smb/smbdirect/mr.c
new file mode 100644
index 000000000000..15c6363a2f97
--- /dev/null
+++ b/fs/smb/smbdirect/mr.c
@@ -0,0 +1,496 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+
+/*
+ * Allocate MRs used for RDMA read/write
+ * The number of MRs will not exceed hardware capability in responder_resources
+ * All MRs are kept in mr_list. The MR can be recovered after it's used
+ * Recovery is done in smbd_mr_recovery_work. The content of list entry changes
+ * as MRs are used and recovered for I/O, but the list links will not change
+ */
+int smbdirect_connection_create_mr_list(struct smbdirect_socket *sc)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_mr_io *mr;
+	int ret;
+	u32 i;
+
+	if (sp->responder_resources == 0) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"responder_resources negotiated as 0\n");
+		return -EINVAL;
+	}
+
+	/* Allocate more MRs (2x) than hardware responder_resources */
+	for (i = 0; i < sp->responder_resources * 2; i++) {
+		mr = kzalloc_obj(*mr);
+		if (!mr) {
+			ret = -ENOMEM;
+			goto kzalloc_mr_failed;
+		}
+
+		kref_init(&mr->kref);
+		mutex_init(&mr->mutex);
+
+		mr->mr = ib_alloc_mr(sc->ib.pd,
+				     sc->mr_io.type,
+				     sp->max_frmr_depth);
+		if (IS_ERR(mr->mr)) {
+			ret = PTR_ERR(mr->mr);
+			smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+				"ib_alloc_mr failed ret=%d (%1pe) type=0x%x max_frmr_depth=%u\n",
+				ret, SMBDIRECT_DEBUG_ERR_PTR(ret),
+				sc->mr_io.type, sp->max_frmr_depth);
+			goto ib_alloc_mr_failed;
+		}
+		mr->sgt.sgl = kzalloc_objs(struct scatterlist, sp->max_frmr_depth);
+		if (!mr->sgt.sgl) {
+			ret = -ENOMEM;
+			smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+				"failed to allocate sgl, max_frmr_depth=%u\n",
+				sp->max_frmr_depth);
+			goto kcalloc_sgl_failed;
+		}
+		mr->state = SMBDIRECT_MR_READY;
+		mr->socket = sc;
+
+		list_add_tail(&mr->list, &sc->mr_io.all.list);
+		atomic_inc(&sc->mr_io.ready.count);
+	}
+
+	return 0;
+
+kcalloc_sgl_failed:
+	ib_dereg_mr(mr->mr);
+ib_alloc_mr_failed:
+	mutex_destroy(&mr->mutex);
+	kfree(mr);
+kzalloc_mr_failed:
+	smbdirect_connection_destroy_mr_list(sc);
+	return ret;
+}
+
+static void smbdirect_mr_io_disable_locked(struct smbdirect_mr_io *mr)
+{
+	struct smbdirect_socket *sc = mr->socket;
+
+	lockdep_assert_held(&mr->mutex);
+
+	if (mr->state == SMBDIRECT_MR_DISABLED)
+		return;
+
+	if (mr->mr)
+		ib_dereg_mr(mr->mr);
+	if (mr->sgt.nents)
+		ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
+	kfree(mr->sgt.sgl);
+
+	mr->mr = NULL;
+	mr->sgt.sgl = NULL;
+	mr->sgt.nents = 0;
+
+	mr->state = SMBDIRECT_MR_DISABLED;
+}
+
+static void smbdirect_mr_io_free_locked(struct kref *kref)
+{
+	struct smbdirect_mr_io *mr =
+		container_of(kref, struct smbdirect_mr_io, kref);
+
+	lockdep_assert_held(&mr->mutex);
+
+	/*
+	 * smbdirect_mr_io_disable_locked() should already be called!
+	 */
+	if (WARN_ON_ONCE(mr->state != SMBDIRECT_MR_DISABLED))
+		smbdirect_mr_io_disable_locked(mr);
+
+	mutex_unlock(&mr->mutex);
+	mutex_destroy(&mr->mutex);
+	kfree(mr);
+}
+
+void smbdirect_connection_destroy_mr_list(struct smbdirect_socket *sc)
+{
+	struct smbdirect_mr_io *mr, *tmp;
+	LIST_HEAD(all_list);
+	unsigned long flags;
+
+	spin_lock_irqsave(&sc->mr_io.all.lock, flags);
+	list_splice_tail_init(&sc->mr_io.all.list, &all_list);
+	spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
+
+	list_for_each_entry_safe(mr, tmp, &all_list, list) {
+		mutex_lock(&mr->mutex);
+
+		smbdirect_mr_io_disable_locked(mr);
+		list_del(&mr->list);
+		mr->socket = NULL;
+
+		/*
+		 * No kref_put_mutex() as it's already locked.
+		 *
+		 * If smbdirect_mr_io_free_locked() is called
+		 * and the mutex is unlocked and mr is gone,
+		 * in that case kref_put() returned 1.
+		 *
+		 * If kref_put() returned 0 we know that
+		 * smbdirect_mr_io_free_locked() didn't
+		 * run. Not by us nor by anyone else, as we
+		 * still hold the mutex, so we need to unlock.
+		 *
+		 * If the mr is still registered it will
+		 * be dangling (detached from the connection
+		 * waiting for smbd_deregister_mr() to be
+		 * called in order to free the memory.
+		 */
+		if (!kref_put(&mr->kref, smbdirect_mr_io_free_locked))
+			mutex_unlock(&mr->mutex);
+	}
+}
+
+/*
+ * Get a MR from mr_list. This function waits until there is at least one MR
+ * available in the list. There may be several CPUs issuing I/O trying to get MR
+ * at the same time, mr_list_lock is used to protect this situation.
+ */
+static struct smbdirect_mr_io *
+smbdirect_connection_get_mr_io(struct smbdirect_socket *sc)
+{
+	struct smbdirect_mr_io *mr;
+	unsigned long flags;
+	int ret;
+
+again:
+	ret = wait_event_interruptible(sc->mr_io.ready.wait_queue,
+				       atomic_read(&sc->mr_io.ready.count) ||
+				       sc->status != SMBDIRECT_SOCKET_CONNECTED);
+	if (ret) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"wait_event_interruptible ret=%d (%1pe)\n",
+			ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return NULL;
+	}
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"sc->status=%s sc->first_error=%1pe\n",
+			smbdirect_socket_status_string(sc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+		return NULL;
+	}
+
+	spin_lock_irqsave(&sc->mr_io.all.lock, flags);
+	list_for_each_entry(mr, &sc->mr_io.all.list, list) {
+		if (mr->state == SMBDIRECT_MR_READY) {
+			mr->state = SMBDIRECT_MR_REGISTERED;
+			kref_get(&mr->kref);
+			spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
+			atomic_dec(&sc->mr_io.ready.count);
+			atomic_inc(&sc->mr_io.used.count);
+			return mr;
+		}
+	}
+
+	spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
+	/*
+	 * It is possible that we could fail to get MR because other processes may
+	 * try to acquire a MR at the same time. If this is the case, retry it.
+	 */
+	goto again;
+}
+
+static void smbdirect_connection_mr_io_register_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_mr_io *mr =
+		container_of(wc->wr_cqe, struct smbdirect_mr_io, cqe);
+	struct smbdirect_socket *sc = mr->socket;
+
+	if (wc->status != IB_WC_SUCCESS) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"wc->status=%s opcode=%d\n",
+			ib_wc_status_msg(wc->status), wc->opcode);
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+	}
+}
+
+static void smbdirect_connection_mr_io_local_inv_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	struct smbdirect_mr_io *mr =
+		container_of(wc->wr_cqe, struct smbdirect_mr_io, cqe);
+	struct smbdirect_socket *sc = mr->socket;
+
+	mr->state = SMBDIRECT_MR_INVALIDATED;
+	if (wc->status != IB_WC_SUCCESS) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"invalidate failed status=%s\n",
+			ib_wc_status_msg(wc->status));
+		smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
+	}
+	complete(&mr->invalidate_done);
+}
+
+/*
+ * Transcribe the pages from an iterator into an MR scatterlist.
+ */
+static int smbdirect_iter_to_sgt(struct iov_iter *iter,
+				 struct sg_table *sgt,
+				 unsigned int max_sg)
+{
+	int ret;
+
+	memset(sgt->sgl, 0, max_sg * sizeof(struct scatterlist));
+
+	ret = extract_iter_to_sg(iter, iov_iter_count(iter), sgt, max_sg, 0);
+	WARN_ON(ret < 0);
+	if (sgt->nents > 0)
+		sg_mark_end(&sgt->sgl[sgt->nents - 1]);
+
+	return ret;
+}
+
+/*
+ * Register memory for RDMA read/write
+ * iter: the buffer to register memory with
+ * writing: true if this is a RDMA write (SMB read), false for RDMA read
+ * need_invalidate: true if this MR needs to be locally invalidated after I/O
+ * return value: the MR registered, NULL if failed.
+ */
+struct smbdirect_mr_io *
+smbdirect_connection_register_mr_io(struct smbdirect_socket *sc,
+				    struct iov_iter *iter,
+				    bool writing,
+				    bool need_invalidate)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	struct smbdirect_mr_io *mr;
+	int ret, num_pages, num_mapped;
+	struct ib_reg_wr *reg_wr;
+
+	num_pages = iov_iter_npages(iter, sp->max_frmr_depth + 1);
+	if (num_pages > sp->max_frmr_depth) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"num_pages=%d max_frmr_depth=%d\n",
+			num_pages, sp->max_frmr_depth);
+		WARN_ON_ONCE(1);
+		return NULL;
+	}
+
+	mr = smbdirect_connection_get_mr_io(sc);
+	if (!mr) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"smbdirect_connection_get_mr_io returning NULL\n");
+		return NULL;
+	}
+
+	mutex_lock(&mr->mutex);
+
+	mr->dir = writing ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
+	mr->need_invalidate = need_invalidate;
+	mr->sgt.nents = 0;
+	mr->sgt.orig_nents = 0;
+
+	smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_INFO,
+		"num_pages=%u count=%zu depth=%u\n",
+		num_pages, iov_iter_count(iter), sp->max_frmr_depth);
+	smbdirect_iter_to_sgt(iter, &mr->sgt, sp->max_frmr_depth);
+
+	num_mapped = ib_dma_map_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
+	if (!num_mapped) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"ib_dma_map_sg num_pages=%u dir=%x num_mapped=%d\n",
+			num_pages, mr->dir, num_mapped);
+		ret = -EIO;
+		goto dma_map_error;
+	}
+
+	ret = ib_map_mr_sg(mr->mr, mr->sgt.sgl, num_mapped, NULL, PAGE_SIZE);
+	if (ret != num_mapped) {
+		smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+			"ib_map_mr_sg failed ret = %d num_mapped = %u\n",
+			ret, num_mapped);
+		if (ret >= 0)
+			ret = -EIO;
+		goto map_mr_error;
+	}
+
+	ib_update_fast_reg_key(mr->mr, ib_inc_rkey(mr->mr->rkey));
+	reg_wr = &mr->wr;
+	reg_wr->wr.opcode = IB_WR_REG_MR;
+	mr->cqe.done = smbdirect_connection_mr_io_register_done;
+	reg_wr->wr.wr_cqe = &mr->cqe;
+	reg_wr->wr.num_sge = 0;
+	reg_wr->wr.send_flags = IB_SEND_SIGNALED;
+	reg_wr->mr = mr->mr;
+	reg_wr->key = mr->mr->rkey;
+	reg_wr->access = writing ?
+			IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
+			IB_ACCESS_REMOTE_READ;
+
+	/*
+	 * There is no need for waiting for complemtion on ib_post_send
+	 * on IB_WR_REG_MR. Hardware enforces a barrier and order of execution
+	 * on the next ib_post_send when we actually send I/O to remote peer
+	 */
+	ret = ib_post_send(sc->ib.qp, &reg_wr->wr, NULL);
+	if (!ret) {
+		/*
+		 * smbdirect_connection_get_mr_io() gave us a reference
+		 * via kref_get(&mr->kref), we keep that and let
+		 * the caller use smbdirect_connection_deregister_mr_io()
+		 * to remove it again.
+		 */
+		mutex_unlock(&mr->mutex);
+		return mr;
+	}
+
+	smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+		"ib_post_send failed ret=%d (%1pe) reg_wr->key=0x%x\n",
+		ret, SMBDIRECT_DEBUG_ERR_PTR(ret), reg_wr->key);
+
+map_mr_error:
+	ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
+
+dma_map_error:
+	mr->sgt.nents = 0;
+	mr->state = SMBDIRECT_MR_ERROR;
+	atomic_dec(&sc->mr_io.used.count);
+
+	smbdirect_socket_schedule_cleanup(sc, ret);
+
+	/*
+	 * smbdirect_connection_get_mr_io() gave us a reference
+	 * via kref_get(&mr->kref), we need to remove it again
+	 * on error.
+	 *
+	 * No kref_put_mutex() as it's already locked.
+	 *
+	 * If smbdirect_mr_io_free_locked() is called
+	 * and the mutex is unlocked and mr is gone,
+	 * in that case kref_put() returned 1.
+	 *
+	 * If kref_put() returned 0 we know that
+	 * smbdirect_mr_io_free_locked() didn't
+	 * run. Not by us nor by anyone else, as we
+	 * still hold the mutex, so we need to unlock.
+	 */
+	if (!kref_put(&mr->kref, smbdirect_mr_io_free_locked))
+		mutex_unlock(&mr->mutex);
+	return NULL;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_register_mr_io);
+
+void smbdirect_mr_io_fill_buffer_descriptor(struct smbdirect_mr_io *mr,
+					    struct smbdirect_buffer_descriptor_v1 *v1)
+{
+	mutex_lock(&mr->mutex);
+	if (mr->state == SMBDIRECT_MR_REGISTERED) {
+		v1->offset = cpu_to_le64(mr->mr->iova);
+		v1->token = cpu_to_le32(mr->mr->rkey);
+		v1->length = cpu_to_le32(mr->mr->length);
+	} else {
+		v1->offset = cpu_to_le64(U64_MAX);
+		v1->token = cpu_to_le32(U32_MAX);
+		v1->length = cpu_to_le32(U32_MAX);
+	}
+	mutex_unlock(&mr->mutex);
+}
+EXPORT_SYMBOL_GPL(smbdirect_mr_io_fill_buffer_descriptor);
+
+/*
+ * Deregister a MR after I/O is done
+ * This function may wait if remote invalidation is not used
+ * and we have to locally invalidate the buffer to prevent data is being
+ * modified by remote peer after upper layer consumes it
+ */
+void smbdirect_connection_deregister_mr_io(struct smbdirect_mr_io *mr)
+{
+	struct smbdirect_socket *sc = mr->socket;
+	int ret = 0;
+
+lock_again:
+	mutex_lock(&mr->mutex);
+	if (mr->state == SMBDIRECT_MR_DISABLED)
+		goto put_kref;
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
+		smbdirect_mr_io_disable_locked(mr);
+		goto put_kref;
+	}
+
+	if (mr->need_invalidate) {
+		struct ib_send_wr *wr = &mr->inv_wr;
+
+		/* Need to finish local invalidation before returning */
+		wr->opcode = IB_WR_LOCAL_INV;
+		mr->cqe.done = smbdirect_connection_mr_io_local_inv_done;
+		wr->wr_cqe = &mr->cqe;
+		wr->num_sge = 0;
+		wr->ex.invalidate_rkey = mr->mr->rkey;
+		wr->send_flags = IB_SEND_SIGNALED;
+
+		init_completion(&mr->invalidate_done);
+		ret = ib_post_send(sc->ib.qp, wr, NULL);
+		if (ret) {
+			smbdirect_log_rdma_mr(sc, SMBDIRECT_LOG_ERR,
+				"ib_post_send failed ret=%d (%1pe)\n",
+				ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
+			smbdirect_mr_io_disable_locked(mr);
+			smbdirect_socket_schedule_cleanup(sc, ret);
+			goto done;
+		}
+
+		/*
+		 * We still hold the reference to mr
+		 * so we can unlock while waiting.
+		 */
+		mutex_unlock(&mr->mutex);
+		wait_for_completion(&mr->invalidate_done);
+		mr->need_invalidate = false;
+		goto lock_again;
+	} else
+		/*
+		 * For remote invalidation, just set it to SMBDIRECT_MR_INVALIDATED
+		 * and defer to mr_recovery_work to recover the MR for next use
+		 */
+		mr->state = SMBDIRECT_MR_INVALIDATED;
+
+	if (mr->sgt.nents) {
+		ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
+		mr->sgt.nents = 0;
+	}
+
+	WARN_ONCE(mr->state != SMBDIRECT_MR_INVALIDATED,
+		  "mr->state[%u] != SMBDIRECT_MR_INVALIDATED[%u]\n",
+		  mr->state, SMBDIRECT_MR_INVALIDATED);
+	mr->state = SMBDIRECT_MR_READY;
+	if (atomic_inc_return(&sc->mr_io.ready.count) == 1)
+		wake_up(&sc->mr_io.ready.wait_queue);
+
+done:
+	atomic_dec(&sc->mr_io.used.count);
+
+put_kref:
+	/*
+	 * No kref_put_mutex() as it's already locked.
+	 *
+	 * If smbdirect_mr_io_free_locked() is called
+	 * and the mutex is unlocked and mr is gone,
+	 * in that case kref_put() returned 1.
+	 *
+	 * If kref_put() returned 0 we know that
+	 * smbdirect_mr_io_free_locked() didn't
+	 * run. Not by us nor by anyone else, as we
+	 * still hold the mutex, so we need to unlock
+	 * and keep the mr in SMBDIRECT_MR_READY or
+	 * SMBDIRECT_MR_ERROR state.
+	 */
+	if (!kref_put(&mr->kref, smbdirect_mr_io_free_locked))
+		mutex_unlock(&mr->mutex);
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_deregister_mr_io);
diff --git a/fs/smb/smbdirect/pdu.h b/fs/smb/smbdirect/pdu.h
new file mode 100644
index 000000000000..7693ba337873
--- /dev/null
+++ b/fs/smb/smbdirect/pdu.h
@@ -0,0 +1,59 @@
+/* SPDX-License-Identifier: GPL-2.0-or-later */
+/*
+ *   Copyright (c) 2017 Stefan Metzmacher
+ */
+
+#ifndef __FS_SMB_COMMON_SMBDIRECT_SMBDIRECT_PDU_H__
+#define __FS_SMB_COMMON_SMBDIRECT_SMBDIRECT_PDU_H__
+
+#define SMBDIRECT_V1 0x0100
+
+/* SMBD minimum receive size and fragmented sized defined in [MS-SMBD] */
+#define SMBDIRECT_MIN_RECEIVE_SIZE		128
+#define SMBDIRECT_MIN_FRAGMENTED_SIZE		131072
+
+/* SMBD negotiation request packet [MS-SMBD] 2.2.1 */
+struct smbdirect_negotiate_req {
+	__le16 min_version;
+	__le16 max_version;
+	__le16 reserved;
+	__le16 credits_requested;
+	__le32 preferred_send_size;
+	__le32 max_receive_size;
+	__le32 max_fragmented_size;
+} __packed;
+
+/* SMBD negotiation response packet [MS-SMBD] 2.2.2 */
+struct smbdirect_negotiate_resp {
+	__le16 min_version;
+	__le16 max_version;
+	__le16 negotiated_version;
+	__le16 reserved;
+	__le16 credits_requested;
+	__le16 credits_granted;
+	__le32 status;
+	__le32 max_readwrite_size;
+	__le32 preferred_send_size;
+	__le32 max_receive_size;
+	__le32 max_fragmented_size;
+} __packed;
+
+#define SMBDIRECT_DATA_MIN_HDR_SIZE 0x14
+#define SMBDIRECT_DATA_OFFSET       0x18
+
+#define SMBDIRECT_FLAG_RESPONSE_REQUESTED 0x0001
+
+/* SMBD data transfer packet with payload [MS-SMBD] 2.2.3 */
+struct smbdirect_data_transfer {
+	__le16 credits_requested;
+	__le16 credits_granted;
+	__le16 flags;
+	__le16 reserved;
+	__le32 remaining_data_length;
+	__le32 data_offset;
+	__le32 data_length;
+	__le32 padding;
+	__u8 buffer[];
+} __packed;
+
+#endif /* __FS_SMB_COMMON_SMBDIRECT_SMBDIRECT_PDU_H__ */
diff --git a/fs/smb/smbdirect/rw.c b/fs/smb/smbdirect/rw.c
new file mode 100644
index 000000000000..6fe38042cfb9
--- /dev/null
+++ b/fs/smb/smbdirect/rw.c
@@ -0,0 +1,255 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (C) 2018, LG Electronics.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+
+static int smbdirect_connection_wait_for_rw_credits(struct smbdirect_socket *sc,
+						    int credits)
+{
+	return smbdirect_socket_wait_for_credits(sc,
+						 SMBDIRECT_SOCKET_CONNECTED,
+						 -ENOTCONN,
+						 &sc->rw_io.credits.wait_queue,
+						 &sc->rw_io.credits.count,
+						 credits);
+}
+
+static int smbdirect_connection_calc_rw_credits(struct smbdirect_socket *sc,
+						const void *buf,
+						size_t len)
+{
+	return DIV_ROUND_UP(smbdirect_get_buf_page_count(buf, len),
+			    sc->rw_io.credits.num_pages);
+}
+
+static int smbdirect_connection_rdma_get_sg_list(void *buf,
+						 size_t size,
+						 struct scatterlist *sg_list,
+						 size_t nentries)
+{
+	bool high = is_vmalloc_addr(buf);
+	struct page *page;
+	size_t offset, len;
+	int i = 0;
+
+	if (size == 0 || nentries < smbdirect_get_buf_page_count(buf, size))
+		return -EINVAL;
+
+	offset = offset_in_page(buf);
+	buf -= offset;
+	while (size > 0) {
+		len = min_t(size_t, PAGE_SIZE - offset, size);
+		if (high)
+			page = vmalloc_to_page(buf);
+		else
+			page = kmap_to_page(buf);
+
+		if (!sg_list)
+			return -EINVAL;
+		sg_set_page(sg_list, page, len, offset);
+		sg_list = sg_next(sg_list);
+
+		buf += PAGE_SIZE;
+		size -= len;
+		offset = 0;
+		i++;
+	}
+
+	return i;
+}
+
+static void smbdirect_connection_rw_io_free(struct smbdirect_rw_io *msg,
+					    enum dma_data_direction dir)
+{
+	struct smbdirect_socket *sc = msg->socket;
+
+	rdma_rw_ctx_destroy(&msg->rdma_ctx,
+			    sc->ib.qp,
+			    sc->ib.qp->port,
+			    msg->sgt.sgl,
+			    msg->sgt.nents,
+			    dir);
+	sg_free_table_chained(&msg->sgt, SG_CHUNK_SIZE);
+	kfree(msg);
+}
+
+static void smbdirect_connection_rdma_rw_done(struct ib_cq *cq, struct ib_wc *wc,
+					      enum dma_data_direction dir)
+{
+	struct smbdirect_rw_io *msg =
+		container_of(wc->wr_cqe, struct smbdirect_rw_io, cqe);
+	struct smbdirect_socket *sc = msg->socket;
+
+	if (wc->status != IB_WC_SUCCESS) {
+		msg->error = -EIO;
+		pr_err("read/write error. opcode = %d, status = %s(%d)\n",
+		       wc->opcode, ib_wc_status_msg(wc->status), wc->status);
+		if (wc->status != IB_WC_WR_FLUSH_ERR)
+			smbdirect_socket_schedule_cleanup(sc, msg->error);
+	}
+
+	complete(msg->completion);
+}
+
+static void smbdirect_connection_rdma_read_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	smbdirect_connection_rdma_rw_done(cq, wc, DMA_FROM_DEVICE);
+}
+
+static void smbdirect_connection_rdma_write_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+	smbdirect_connection_rdma_rw_done(cq, wc, DMA_TO_DEVICE);
+}
+
+int smbdirect_connection_rdma_xmit(struct smbdirect_socket *sc,
+				   void *buf, size_t buf_len,
+				   struct smbdirect_buffer_descriptor_v1 *desc,
+				   size_t desc_len,
+				   bool is_read)
+{
+	const struct smbdirect_socket_parameters *sp = &sc->parameters;
+	enum dma_data_direction direction = is_read ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
+	struct smbdirect_rw_io *msg, *next_msg;
+	size_t i;
+	int ret;
+	DECLARE_COMPLETION_ONSTACK(completion);
+	struct ib_send_wr *first_wr;
+	LIST_HEAD(msg_list);
+	u8 *desc_buf;
+	int credits_needed;
+	size_t desc_buf_len, desc_num = 0;
+
+	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
+		return -ENOTCONN;
+
+	if (buf_len > sp->max_read_write_size)
+		return -EINVAL;
+
+	/* calculate needed credits */
+	credits_needed = 0;
+	desc_buf = buf;
+	for (i = 0; i < desc_len / sizeof(*desc); i++) {
+		if (!buf_len)
+			break;
+
+		desc_buf_len = le32_to_cpu(desc[i].length);
+		if (!desc_buf_len)
+			return -EINVAL;
+
+		if (desc_buf_len > buf_len) {
+			desc_buf_len = buf_len;
+			desc[i].length = cpu_to_le32(desc_buf_len);
+			buf_len = 0;
+		}
+
+		credits_needed += smbdirect_connection_calc_rw_credits(sc,
+								       desc_buf,
+								       desc_buf_len);
+		desc_buf += desc_buf_len;
+		buf_len -= desc_buf_len;
+		desc_num++;
+	}
+
+	smbdirect_log_rdma_rw(sc, SMBDIRECT_LOG_INFO,
+		"RDMA %s, len %zu, needed credits %d\n",
+		str_read_write(is_read), buf_len, credits_needed);
+
+	ret = smbdirect_connection_wait_for_rw_credits(sc, credits_needed);
+	if (ret < 0)
+		return ret;
+
+	/* build rdma_rw_ctx for each descriptor */
+	desc_buf = buf;
+	for (i = 0; i < desc_num; i++) {
+		size_t page_count;
+
+		msg = kzalloc_flex(*msg, sg_list, SG_CHUNK_SIZE,
+				   sc->rw_io.mem.gfp_mask);
+		if (!msg) {
+			ret = -ENOMEM;
+			goto out;
+		}
+
+		desc_buf_len = le32_to_cpu(desc[i].length);
+		page_count = smbdirect_get_buf_page_count(desc_buf, desc_buf_len);
+
+		msg->socket = sc;
+		msg->cqe.done = is_read ?
+			smbdirect_connection_rdma_read_done :
+			smbdirect_connection_rdma_write_done;
+		msg->completion = &completion;
+
+		msg->sgt.sgl = &msg->sg_list[0];
+		ret = sg_alloc_table_chained(&msg->sgt,
+					     page_count,
+					     msg->sg_list,
+					     SG_CHUNK_SIZE);
+		if (ret) {
+			ret = -ENOMEM;
+			goto free_msg;
+		}
+
+		ret = smbdirect_connection_rdma_get_sg_list(desc_buf,
+							    desc_buf_len,
+							    msg->sgt.sgl,
+							    msg->sgt.orig_nents);
+		if (ret < 0)
+			goto free_table;
+
+		ret = rdma_rw_ctx_init(&msg->rdma_ctx,
+				       sc->ib.qp,
+				       sc->ib.qp->port,
+				       msg->sgt.sgl,
+				       page_count,
+				       0,
+				       le64_to_cpu(desc[i].offset),
+				       le32_to_cpu(desc[i].token),
+				       direction);
+		if (ret < 0) {
+			pr_err("failed to init rdma_rw_ctx: %d\n", ret);
+			goto free_table;
+		}
+
+		list_add_tail(&msg->list, &msg_list);
+		desc_buf += desc_buf_len;
+	}
+
+	/* concatenate work requests of rdma_rw_ctxs */
+	first_wr = NULL;
+	list_for_each_entry_reverse(msg, &msg_list, list) {
+		first_wr = rdma_rw_ctx_wrs(&msg->rdma_ctx,
+					   sc->ib.qp,
+					   sc->ib.qp->port,
+					   &msg->cqe,
+					   first_wr);
+	}
+
+	ret = ib_post_send(sc->ib.qp, first_wr, NULL);
+	if (ret) {
+		pr_err("failed to post send wr for RDMA R/W: %d\n", ret);
+		goto out;
+	}
+
+	msg = list_last_entry(&msg_list, struct smbdirect_rw_io, list);
+	wait_for_completion(&completion);
+	ret = msg->error;
+out:
+	list_for_each_entry_safe(msg, next_msg, &msg_list, list) {
+		list_del(&msg->list);
+		smbdirect_connection_rw_io_free(msg, direction);
+	}
+	atomic_add(credits_needed, &sc->rw_io.credits.count);
+	wake_up(&sc->rw_io.credits.wait_queue);
+	return ret;
+
+free_table:
+	sg_free_table_chained(&msg->sgt, SG_CHUNK_SIZE);
+free_msg:
+	kfree(msg);
+	goto out;
+}
+EXPORT_SYMBOL_GPL(smbdirect_connection_rdma_xmit);
diff --git a/fs/smb/smbdirect/socket.c b/fs/smb/smbdirect/socket.c
new file mode 100644
index 000000000000..bb02df6158b9
--- /dev/null
+++ b/fs/smb/smbdirect/socket.c
@@ -0,0 +1,791 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ *   Copyright (C) 2017, Microsoft Corporation.
+ *   Copyright (c) 2025, Stefan Metzmacher
+ */
+
+#include "internal.h"
+
+bool smbdirect_frwr_is_supported(const struct ib_device_attr *attrs)
+{
+	/*
+	 * Test if FRWR (Fast Registration Work Requests) is supported on the
+	 * device This implementation requires FRWR on RDMA read/write return
+	 * value: true if it is supported
+	 */
+
+	if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
+		return false;
+	if (attrs->max_fast_reg_page_list_len == 0)
+		return false;
+	return true;
+}
+EXPORT_SYMBOL_GPL(smbdirect_frwr_is_supported);
+
+static void smbdirect_socket_cleanup_work(struct work_struct *work);
+
+static int smbdirect_socket_rdma_event_handler(struct rdma_cm_id *id,
+					       struct rdma_cm_event *event)
+{
+	struct smbdirect_socket *sc = id->context;
+	int ret = -ESTALE;
+
+	/*
+	 * This should be replaced before any real work
+	 * starts! So it should never be called!
+	 */
+
+	if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
+		ret = -ENETDOWN;
+	if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
+		ret = event->status;
+	pr_err("%s (first_error=%1pe, expected=%s) => event=%s status=%d => ret=%1pe\n",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
+		rdma_event_msg(sc->rdma.expected_event),
+		rdma_event_msg(event->event),
+		event->status,
+		SMBDIRECT_DEBUG_ERR_PTR(ret));
+	WARN_ONCE(1, "%s should not be called!\n", __func__);
+	sc->rdma.cm_id = NULL;
+	return -ESTALE;
+}
+
+int smbdirect_socket_init_new(struct net *net, struct smbdirect_socket *sc)
+{
+	struct rdma_cm_id *id;
+	int ret;
+
+	smbdirect_socket_init(sc);
+
+	id = rdma_create_id(net,
+			    smbdirect_socket_rdma_event_handler,
+			    sc,
+			    RDMA_PS_TCP,
+			    IB_QPT_RC);
+	if (IS_ERR(id)) {
+		pr_err("%s: rdma_create_id() failed %1pe\n", __func__, id);
+		return PTR_ERR(id);
+	}
+
+	ret = rdma_set_afonly(id, 1);
+	if (ret) {
+		rdma_destroy_id(id);
+		pr_err("%s: rdma_set_afonly() failed %1pe\n",
+		       __func__, SMBDIRECT_DEBUG_ERR_PTR(ret));
+		return ret;
+	}
+
+	sc->rdma.cm_id = id;
+
+	INIT_WORK(&sc->disconnect_work, smbdirect_socket_cleanup_work);
+
+	return 0;
+}
+
+int smbdirect_socket_create_kern(struct net *net, struct smbdirect_socket **_sc)
+{
+	struct smbdirect_socket *sc;
+	int ret;
+
+	ret = -ENOMEM;
+	sc = kzalloc_obj(*sc);
+	if (!sc)
+		goto alloc_failed;
+
+	ret = smbdirect_socket_init_new(net, sc);
+	if (ret)
+		goto init_failed;
+
+	kref_init(&sc->refs.destroy);
+
+	*_sc = sc;
+	return 0;
+
+init_failed:
+	kfree(sc);
+alloc_failed:
+	return ret;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_create_kern);
+
+int smbdirect_socket_init_accepting(struct rdma_cm_id *id, struct smbdirect_socket *sc)
+{
+	smbdirect_socket_init(sc);
+
+	sc->rdma.cm_id = id;
+	sc->rdma.cm_id->context = sc;
+	sc->rdma.cm_id->event_handler = smbdirect_socket_rdma_event_handler;
+
+	sc->ib.dev = sc->rdma.cm_id->device;
+
+	INIT_WORK(&sc->disconnect_work, smbdirect_socket_cleanup_work);
+
+	return 0;
+}
+
+int smbdirect_socket_create_accepting(struct rdma_cm_id *id, struct smbdirect_socket **_sc)
+{
+	struct smbdirect_socket *sc;
+	int ret;
+
+	ret = -ENOMEM;
+	sc = kzalloc_obj(*sc);
+	if (!sc)
+		goto alloc_failed;
+
+	ret = smbdirect_socket_init_accepting(id, sc);
+	if (ret)
+		goto init_failed;
+
+	kref_init(&sc->refs.destroy);
+
+	*_sc = sc;
+	return 0;
+
+init_failed:
+	kfree(sc);
+alloc_failed:
+	return ret;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_create_accepting);
+
+int smbdirect_socket_set_initial_parameters(struct smbdirect_socket *sc,
+					    const struct smbdirect_socket_parameters *sp)
+{
+	/*
+	 * This is only allowed before connect or accept
+	 */
+	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED,
+		  "status=%s first_error=%1pe",
+		  smbdirect_socket_status_string(sc->status),
+		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+	if (sc->status != SMBDIRECT_SOCKET_CREATED)
+		return -EINVAL;
+
+	if (sp->flags & ~SMBDIRECT_FLAG_PORT_RANGE_MASK)
+		return -EINVAL;
+
+	if (sp->initiator_depth > U8_MAX)
+		return -EINVAL;
+	if (sp->responder_resources > U8_MAX)
+		return -EINVAL;
+
+	if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IB &&
+	    sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IW)
+		return -EINVAL;
+	else if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IB)
+		rdma_restrict_node_type(sc->rdma.cm_id, RDMA_NODE_IB_CA);
+	else if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IW)
+		rdma_restrict_node_type(sc->rdma.cm_id, RDMA_NODE_RNIC);
+
+	/*
+	 * Make a copy of the callers parameters
+	 * from here we only work on the copy
+	 *
+	 * TODO: do we want consistency checking?
+	 */
+	sc->parameters = *sp;
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_set_initial_parameters);
+
+const struct smbdirect_socket_parameters *
+smbdirect_socket_get_current_parameters(struct smbdirect_socket *sc)
+{
+	return &sc->parameters;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_get_current_parameters);
+
+int smbdirect_socket_set_kernel_settings(struct smbdirect_socket *sc,
+					 enum ib_poll_context poll_ctx,
+					 gfp_t gfp_mask)
+{
+	/*
+	 * This is only allowed before connect or accept
+	 */
+	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED,
+		  "status=%s first_error=%1pe",
+		  smbdirect_socket_status_string(sc->status),
+		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+	if (sc->status != SMBDIRECT_SOCKET_CREATED)
+		return -EINVAL;
+
+	sc->ib.poll_ctx = poll_ctx;
+
+	sc->send_io.mem.gfp_mask = gfp_mask;
+	sc->recv_io.mem.gfp_mask = gfp_mask;
+	sc->rw_io.mem.gfp_mask = gfp_mask;
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_set_kernel_settings);
+
+void smbdirect_socket_set_logging(struct smbdirect_socket *sc,
+				  void *private_ptr,
+				  bool (*needed)(struct smbdirect_socket *sc,
+						 void *private_ptr,
+						 unsigned int lvl,
+						 unsigned int cls),
+				  void (*vaprintf)(struct smbdirect_socket *sc,
+						   const char *func,
+						   unsigned int line,
+						   void *private_ptr,
+						   unsigned int lvl,
+						   unsigned int cls,
+						   struct va_format *vaf))
+{
+	sc->logging.private_ptr = private_ptr;
+	sc->logging.needed = needed;
+	sc->logging.vaprintf = vaprintf;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_set_logging);
+
+static void smbdirect_socket_wake_up_all(struct smbdirect_socket *sc)
+{
+	/*
+	 * Wake up all waiters in all wait queues
+	 * in order to notice the broken connection.
+	 */
+	wake_up_all(&sc->status_wait);
+	wake_up_all(&sc->listen.wait_queue);
+	wake_up_all(&sc->send_io.bcredits.wait_queue);
+	wake_up_all(&sc->send_io.lcredits.wait_queue);
+	wake_up_all(&sc->send_io.credits.wait_queue);
+	wake_up_all(&sc->send_io.pending.zero_wait_queue);
+	wake_up_all(&sc->recv_io.reassembly.wait_queue);
+	wake_up_all(&sc->rw_io.credits.wait_queue);
+	wake_up_all(&sc->mr_io.ready.wait_queue);
+}
+
+void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
+					 const char *macro_name,
+					 unsigned int lvl,
+					 const char *func,
+					 unsigned int line,
+					 int error,
+					 enum smbdirect_socket_status *force_status)
+{
+	struct smbdirect_socket *psc, *tsc;
+	unsigned long flags;
+	bool was_first = false;
+
+	if (!sc->first_error) {
+		___smbdirect_log_generic(sc, func, line,
+			lvl,
+			SMBDIRECT_LOG_RDMA_EVENT,
+			"%s(%1pe%s%s) called from %s in line=%u status=%s\n",
+			macro_name,
+			SMBDIRECT_DEBUG_ERR_PTR(error),
+			force_status ? ", " : "",
+			force_status ? smbdirect_socket_status_string(*force_status) : "",
+			func, line,
+			smbdirect_socket_status_string(sc->status));
+		if (error)
+			sc->first_error = error;
+		else
+			sc->first_error = -ECONNABORTED;
+		was_first = true;
+	}
+
+	/*
+	 * make sure other work (than disconnect_work)
+	 * is not queued again but here we don't block and avoid
+	 * disable[_delayed]_work_sync()
+	 */
+	disable_work(&sc->connect.work);
+	disable_work(&sc->recv_io.posted.refill_work);
+	disable_work(&sc->idle.immediate_work);
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
+	disable_delayed_work(&sc->idle.timer_work);
+
+	/*
+	 * In case we were a listener we need to
+	 * disconnect all pending and ready sockets
+	 *
+	 * First we move ready sockets to pending again.
+	 *
+	 * Only a socket that was a listener (listen.backlog != -1) owns a
+	 * populated listen.ready/pending list.  Guarding on that also keeps
+	 * lockdep quiet: without it, the listener holds sc->listen.lock while
+	 * the loop recurses into each child psc, which takes psc->listen.lock.
+	 * Those are always different instances of the same lock class -- a
+	 * child never listens, so the nesting is strictly listener -> child
+	 * and cannot really deadlock -- but lockdep only sees the class and
+	 * reports "possible recursive locking".  A child has empty listen
+	 * lists and nothing to do here, so skipping it loses nothing, and a
+	 * pending child stays on its listener's list for the free path
+	 * (smbdirect_socket_destroy) to reap.
+	 */
+	if (sc->listen.backlog != -1) { /* was a listener */
+		spin_lock_irqsave(&sc->listen.lock, flags);
+		list_splice_init(&sc->listen.ready, &sc->listen.pending);
+		list_for_each_entry_safe(psc, tsc, &sc->listen.pending, accept.list)
+			smbdirect_socket_schedule_cleanup(psc, sc->first_error);
+		spin_unlock_irqrestore(&sc->listen.lock, flags);
+	}
+
+	switch (sc->status) {
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
+	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
+	case SMBDIRECT_SOCKET_ERROR:
+	case SMBDIRECT_SOCKET_DISCONNECTING:
+	case SMBDIRECT_SOCKET_DISCONNECTED:
+	case SMBDIRECT_SOCKET_DESTROYED:
+		/*
+		 * Keep the current error status
+		 */
+		break;
+
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
+		sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED;
+		break;
+
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
+		sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED;
+		break;
+
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
+		sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED;
+		break;
+
+	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
+	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
+		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
+		break;
+
+	case SMBDIRECT_SOCKET_CREATED:
+	case SMBDIRECT_SOCKET_LISTENING:
+		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
+		break;
+
+	case SMBDIRECT_SOCKET_CONNECTED:
+		sc->status = SMBDIRECT_SOCKET_ERROR;
+		break;
+	}
+
+	if (force_status && (was_first || *force_status > sc->status))
+		sc->status = *force_status;
+
+	/*
+	 * Wake up all waiters in all wait queues
+	 * in order to notice the broken connection.
+	 */
+	smbdirect_socket_wake_up_all(sc);
+
+	queue_work(sc->workqueues.cleanup, &sc->disconnect_work);
+}
+
+static void smbdirect_socket_cleanup_work(struct work_struct *work)
+{
+	struct smbdirect_socket *sc =
+		container_of(work, struct smbdirect_socket, disconnect_work);
+	struct smbdirect_socket *psc, *tsc;
+	unsigned long flags;
+
+	/*
+	 * This should not never be called in an interrupt!
+	 */
+	WARN_ON_ONCE(in_interrupt());
+
+	if (!sc->first_error) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
+			"%s called with first_error==0\n",
+			smbdirect_socket_status_string(sc->status));
+
+		sc->first_error = -ECONNABORTED;
+	}
+
+	/*
+	 * make sure this and other work is not queued again
+	 * but here we don't block and avoid
+	 * disable[_delayed]_work_sync()
+	 */
+	disable_work(&sc->disconnect_work);
+	disable_work(&sc->connect.work);
+	disable_work(&sc->recv_io.posted.refill_work);
+	disable_work(&sc->idle.immediate_work);
+	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
+	disable_delayed_work(&sc->idle.timer_work);
+
+	/*
+	 * In case we were a listener we need to
+	 * disconnect all pending and ready sockets
+	 *
+	 * First we move ready sockets to pending again.
+	 *
+	 * Guarded on listen.backlog != -1 for the same reason as in
+	 * __smbdirect_socket_schedule_cleanup(): only a listener owns a
+	 * populated listen list, and skipping the block for a child avoids
+	 * nesting psc->listen.lock under a listener's listen.lock (different
+	 * instances of one class -- harmless, but lockdep cannot tell).
+	 */
+	if (sc->listen.backlog != -1) { /* was a listener */
+		spin_lock_irqsave(&sc->listen.lock, flags);
+		list_splice_init(&sc->listen.ready, &sc->listen.pending);
+		list_for_each_entry_safe(psc, tsc, &sc->listen.pending, accept.list)
+			smbdirect_socket_schedule_cleanup(psc, sc->first_error);
+		spin_unlock_irqrestore(&sc->listen.lock, flags);
+	}
+
+	switch (sc->status) {
+	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
+	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
+	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
+	case SMBDIRECT_SOCKET_CONNECTED:
+	case SMBDIRECT_SOCKET_ERROR:
+		sc->status = SMBDIRECT_SOCKET_DISCONNECTING;
+		/*
+		 * Make sure we hold the callback lock
+		 * im order to coordinate with the
+		 * rdma_event handlers, typically
+		 * smbdirect_connection_rdma_event_handler(),
+		 * and smbdirect_socket_destroy().
+		 *
+		 * So that the order of ib_drain_qp()
+		 * and rdma_disconnect() is controlled
+		 * by the mutex.
+		 */
+		rdma_lock_handler(sc->rdma.cm_id);
+		rdma_disconnect(sc->rdma.cm_id);
+		rdma_unlock_handler(sc->rdma.cm_id);
+		break;
+
+	case SMBDIRECT_SOCKET_CREATED:
+	case SMBDIRECT_SOCKET_LISTENING:
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
+		/*
+		 * rdma_{accept,connect}() never reached
+		 * RDMA_CM_EVENT_ESTABLISHED
+		 */
+		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
+		break;
+
+	case SMBDIRECT_SOCKET_DISCONNECTING:
+	case SMBDIRECT_SOCKET_DISCONNECTED:
+	case SMBDIRECT_SOCKET_DESTROYED:
+		break;
+	}
+
+	/*
+	 * Wake up all waiters in all wait queues
+	 * in order to notice the broken connection.
+	 */
+	smbdirect_socket_wake_up_all(sc);
+}
+
+static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
+{
+	struct smbdirect_socket *psc, *tsc;
+	size_t psockets;
+	struct smbdirect_recv_io *recv_io;
+	struct smbdirect_recv_io *recv_tmp;
+	LIST_HEAD(all_list);
+	LIST_HEAD(pending_list);
+	unsigned long flags;
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"status=%s first_error=%1pe",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+
+	/*
+	 * This should not never be called in an interrupt!
+	 */
+	WARN_ON_ONCE(in_interrupt());
+
+	if (sc->status == SMBDIRECT_SOCKET_DESTROYED)
+		return;
+
+	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_DISCONNECTED,
+		  "status=%s first_error=%1pe",
+		  smbdirect_socket_status_string(sc->status),
+		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+
+	/*
+	 * The listener should clear this before we reach this
+	 */
+	WARN_ONCE(sc->accept.listener,
+		  "status=%s first_error=%1pe",
+		  smbdirect_socket_status_string(sc->status),
+		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+
+	/*
+	 * Wake up all waiters in all wait queues
+	 * in order to notice the broken connection.
+	 *
+	 * Most likely this was already called via
+	 * smbdirect_socket_cleanup_work(), but call it again...
+	 */
+	smbdirect_socket_wake_up_all(sc);
+
+	disable_work_sync(&sc->disconnect_work);
+	disable_work_sync(&sc->connect.work);
+	disable_work_sync(&sc->recv_io.posted.refill_work);
+	disable_work_sync(&sc->idle.immediate_work);
+	disable_delayed_work_sync(&sc->idle.timer_work);
+
+	if (sc->rdma.cm_id)
+		rdma_lock_handler(sc->rdma.cm_id);
+
+	if (sc->ib.qp) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"drain qp\n");
+		ib_drain_qp(sc->ib.qp);
+	}
+
+	/*
+	 * In case we were a listener we need to
+	 * disconnect all pending and ready sockets
+	 *
+	 * We move ready sockets to pending again.
+	 *
+	 * Capture them here -- rdma_lock_handler(sc->rdma.cm_id) is held above,
+	 * so a concurrent CM CONNECT_REQUEST cannot add more; sc->listen.lock
+	 * below only protects the list splice itself -- but DEFER releasing
+	 * them until the listener's cm_id is destroyed:
+	 *
+	 * - smbdirect_socket_release() -> smbdirect_socket_destroy() takes the
+	 *   child's own rdma_lock_handler() lock (&id_priv->handler_mutex).
+	 *   The listener's and the child's cm_id are always different
+	 *   instances, so the nesting cannot really deadlock, but lockdep only
+	 *   sees one lock class and reports "possible recursive locking".
+	 *
+	 * - rdma_destroy_id() of a child before the listener's own
+	 *   rdma_destroy_id() below lets _cma_cancel_listens() walk the freed
+	 *   child id_priv (KASAN slab-use-after-free in __mutex_lock()).
+	 *
+	 * The children are independent sockets whose teardown does not need
+	 * the listener's handler lock.
+	 */
+	spin_lock_irqsave(&sc->listen.lock, flags);
+	list_splice_tail_init(&sc->listen.ready, &pending_list);
+	list_splice_tail_init(&sc->listen.pending, &pending_list);
+	spin_unlock_irqrestore(&sc->listen.lock, flags);
+
+	/* It's not possible for upper layer to get to reassembly */
+	if (sc->listen.backlog == -1) /* was not a listener */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"drain the reassembly queue\n");
+	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
+	list_splice_tail_init(&sc->recv_io.reassembly.list, &all_list);
+	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
+	list_for_each_entry_safe(recv_io, recv_tmp, &all_list, list)
+		smbdirect_connection_put_recv_io(recv_io);
+	sc->recv_io.reassembly.data_length = 0;
+
+	if (sc->listen.backlog == -1) /* was not a listener */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"freeing mr list\n");
+	smbdirect_connection_destroy_mr_list(sc);
+
+	if (sc->listen.backlog == -1) /* was not a listener */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"destroying qp\n");
+	smbdirect_connection_destroy_qp(sc);
+	if (sc->rdma.cm_id) {
+		rdma_unlock_handler(sc->rdma.cm_id);
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"destroying cm_id\n");
+		rdma_destroy_id(sc->rdma.cm_id);
+		sc->rdma.cm_id = NULL;
+	}
+
+	/*
+	 * The listener's rdma_lock_handler() lock is dropped and its cm_id is
+	 * destroyed, so it is safe to release the child sockets captured
+	 * above: each release recurses into smbdirect_socket_destroy() and
+	 * takes that child's own handler_mutex without nesting it under the
+	 * listener's, and _cma_cancel_listens() can no longer reach them.
+	 */
+	psockets = list_count_nodes(&pending_list);
+	if (sc->listen.backlog != -1) /* was a listener */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"release %zu pending sockets\n", psockets);
+	list_for_each_entry_safe(psc, tsc, &pending_list, accept.list) {
+		list_del_init(&psc->accept.list);
+		psc->accept.listener = NULL;
+		smbdirect_socket_release(psc);
+	}
+	if (sc->listen.backlog != -1) /* was a listener */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"released %zu pending sockets\n", psockets);
+
+	if (sc->listen.backlog == -1) /* was not a listener */
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"destroying mem pools\n");
+	smbdirect_connection_destroy_mem_pools(sc);
+
+	sc->status = SMBDIRECT_SOCKET_DESTROYED;
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"rdma session destroyed\n");
+}
+
+void smbdirect_socket_destroy_sync(struct smbdirect_socket *sc)
+{
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"status=%s first_error=%1pe",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+
+	/*
+	 * This should not never be called in an interrupt!
+	 */
+	WARN_ON_ONCE(in_interrupt());
+
+	/*
+	 * First we try to disable the work
+	 * without disable_work_sync() in a
+	 * non blocking way, if it's already
+	 * running it will be handles by
+	 * disable_work_sync() below.
+	 *
+	 * Here we just want to make sure queue_work() in
+	 * smbdirect_socket_schedule_cleanup_lvl()
+	 * is a no-op.
+	 */
+	disable_work(&sc->disconnect_work);
+
+	if (!sc->first_error)
+		/*
+		 * SMBDIRECT_LOG_INFO is enough here
+		 * as this is the typical case where
+		 * we terminate the connection ourself.
+		 */
+		smbdirect_socket_schedule_cleanup_lvl(sc,
+						      SMBDIRECT_LOG_INFO,
+						      -ESHUTDOWN);
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"cancelling and disable disconnect_work\n");
+	disable_work_sync(&sc->disconnect_work);
+
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"destroying rdma session\n");
+	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTING)
+		smbdirect_socket_cleanup_work(&sc->disconnect_work);
+	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTED) {
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"wait for transport being disconnected\n");
+		wait_event(sc->status_wait, sc->status == SMBDIRECT_SOCKET_DISCONNECTED);
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+			"waited for transport being disconnected\n");
+	}
+
+	/*
+	 * Once we reached SMBDIRECT_SOCKET_DISCONNECTED,
+	 * we should call smbdirect_socket_destroy()
+	 */
+	smbdirect_socket_destroy(sc);
+	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
+		"status=%s first_error=%1pe",
+		smbdirect_socket_status_string(sc->status),
+		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
+}
+
+int smbdirect_socket_bind(struct smbdirect_socket *sc, struct sockaddr *addr)
+{
+	int ret;
+
+	if (sc->status != SMBDIRECT_SOCKET_CREATED)
+		return -EINVAL;
+
+	ret = rdma_bind_addr(sc->rdma.cm_id, addr);
+	if (ret)
+		return ret;
+
+	return 0;
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_bind);
+
+void smbdirect_socket_shutdown(struct smbdirect_socket *sc)
+{
+	smbdirect_socket_schedule_cleanup_lvl(sc, SMBDIRECT_LOG_INFO, -ESHUTDOWN);
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_shutdown);
+
+static void smbdirect_socket_release_disconnect(struct kref *kref)
+{
+	struct smbdirect_socket *sc =
+		container_of(kref, struct smbdirect_socket, refs.disconnect);
+
+	/*
+	 * For now do a sync disconnect/destroy
+	 */
+	smbdirect_socket_destroy_sync(sc);
+}
+
+static void smbdirect_socket_release_destroy(struct kref *kref)
+{
+	struct smbdirect_socket *sc =
+		container_of(kref, struct smbdirect_socket, refs.destroy);
+
+	/*
+	 * Do a sync disconnect/destroy...
+	 * hopefully a no-op, as it should be already
+	 * in DESTROYED state, before we free the memory.
+	 */
+	smbdirect_socket_destroy_sync(sc);
+	kfree(sc);
+}
+
+void smbdirect_socket_release(struct smbdirect_socket *sc)
+{
+	/*
+	 * We expect only 1 disconnect reference
+	 * and if it is already 0, it's a use after free!
+	 */
+	WARN_ON_ONCE(kref_read(&sc->refs.disconnect) != 1);
+	WARN_ON(!kref_put(&sc->refs.disconnect, smbdirect_socket_release_disconnect));
+
+	/*
+	 * This may not trigger smbdirect_socket_release_destroy(),
+	 * if struct smbdirect_socket is embedded in another structure
+	 * indicated by REFCOUNT_MAX.
+	 */
+	kref_put(&sc->refs.destroy, smbdirect_socket_release_destroy);
+}
+EXPORT_SYMBOL_GPL(smbdirect_socket_release);
+
+int smbdirect_socket_wait_for_credits(struct smbdirect_socket *sc,
+				      enum smbdirect_socket_status expected_status,
+				      int unexpected_errno,
+				      wait_queue_head_t *waitq,
+				      atomic_t *total_credits,
+				      int needed)
+{
+	int ret;
+
+	if (WARN_ON_ONCE(needed < 0))
+		return -EINVAL;
+
+	do {
+		if (atomic_sub_return(needed, total_credits) >= 0)
+			return 0;
+
+		atomic_add(needed, total_credits);
+		ret = wait_event_interruptible(*waitq,
+					       atomic_read(total_credits) >= needed ||
+					       sc->status != expected_status);
+
+		if (sc->status != expected_status)
+			return unexpected_errno;
+		else if (ret < 0)
+			return ret;
+	} while (true);
+}
diff --git a/fs/smb/smbdirect/socket.h b/fs/smb/smbdirect/socket.h
new file mode 100644
index 000000000000..c09eddd8ad16
--- /dev/null
+++ b/fs/smb/smbdirect/socket.h
@@ -0,0 +1,832 @@
+/* SPDX-License-Identifier: GPL-2.0-or-later */
+/*
+ *   Copyright (c) 2025 Stefan Metzmacher
+ */
+
+#ifndef __FS_SMB_COMMON_SMBDIRECT_SMBDIRECT_SOCKET_H__
+#define __FS_SMB_COMMON_SMBDIRECT_SMBDIRECT_SOCKET_H__
+
+#include <linux/wait.h>
+#include <linux/workqueue.h>
+#include <linux/kref.h>
+#include <linux/mempool.h>
+#include <linux/spinlock.h>
+#include <linux/mutex.h>
+#include <linux/completion.h>
+#include <rdma/rw.h>
+
+enum smbdirect_socket_status {
+	SMBDIRECT_SOCKET_CREATED,
+	SMBDIRECT_SOCKET_LISTENING,
+	SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED,
+	SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING,
+	SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED,
+	SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED,
+	SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING,
+	SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED,
+	SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED,
+	SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING,
+	SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED,
+	SMBDIRECT_SOCKET_NEGOTIATE_NEEDED,
+	SMBDIRECT_SOCKET_NEGOTIATE_RUNNING,
+	SMBDIRECT_SOCKET_NEGOTIATE_FAILED,
+	SMBDIRECT_SOCKET_CONNECTED,
+	SMBDIRECT_SOCKET_ERROR,
+	SMBDIRECT_SOCKET_DISCONNECTING,
+	SMBDIRECT_SOCKET_DISCONNECTED,
+	SMBDIRECT_SOCKET_DESTROYED
+};
+
+static __always_inline
+const char *smbdirect_socket_status_string(enum smbdirect_socket_status status)
+{
+	switch (status) {
+	case SMBDIRECT_SOCKET_CREATED:
+		return "CREATED";
+	case SMBDIRECT_SOCKET_LISTENING:
+		return "LISTENING";
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
+		return "RESOLVE_ADDR_NEEDED";
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
+		return "RESOLVE_ADDR_RUNNING";
+	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
+		return "RESOLVE_ADDR_FAILED";
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
+		return "RESOLVE_ROUTE_NEEDED";
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
+		return "RESOLVE_ROUTE_RUNNING";
+	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
+		return "RESOLVE_ROUTE_FAILED";
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
+		return "RDMA_CONNECT_NEEDED";
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
+		return "RDMA_CONNECT_RUNNING";
+	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
+		return "RDMA_CONNECT_FAILED";
+	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
+		return "NEGOTIATE_NEEDED";
+	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
+		return "NEGOTIATE_RUNNING";
+	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
+		return "NEGOTIATE_FAILED";
+	case SMBDIRECT_SOCKET_CONNECTED:
+		return "CONNECTED";
+	case SMBDIRECT_SOCKET_ERROR:
+		return "ERROR";
+	case SMBDIRECT_SOCKET_DISCONNECTING:
+		return "DISCONNECTING";
+	case SMBDIRECT_SOCKET_DISCONNECTED:
+		return "DISCONNECTED";
+	case SMBDIRECT_SOCKET_DESTROYED:
+		return "DESTROYED";
+	}
+
+	return "<unknown>";
+}
+
+/*
+ * This can be used with %1pe to print errors as strings or '0'
+ * And it avoids warnings like: warn: passing zero to 'ERR_PTR'
+ * from smatch -p=kernel --pedantic
+ */
+static __always_inline
+const void * __must_check SMBDIRECT_DEBUG_ERR_PTR(long error)
+{
+	if (error == 0)
+		return NULL;
+	return ERR_PTR(error);
+}
+
+enum smbdirect_keepalive_status {
+	SMBDIRECT_KEEPALIVE_NONE,
+	SMBDIRECT_KEEPALIVE_PENDING,
+	SMBDIRECT_KEEPALIVE_SENT
+};
+
+struct smbdirect_socket {
+	enum smbdirect_socket_status status;
+	wait_queue_head_t status_wait;
+	int first_error;
+
+	/*
+	 * This points to the workqueues to
+	 * be used for this socket.
+	 */
+	struct {
+		struct workqueue_struct *accept;
+		struct workqueue_struct *connect;
+		struct workqueue_struct *idle;
+		struct workqueue_struct *refill;
+		struct workqueue_struct *immediate;
+		struct workqueue_struct *cleanup;
+	} workqueues;
+
+	struct work_struct disconnect_work;
+
+	/*
+	 * The reference counts.
+	 */
+	struct {
+		/*
+		 * This holds the references by the
+		 * frontend, typically the smb layer.
+		 *
+		 * It is typically 1 and a disconnect
+		 * will happen if it reaches 0.
+		 */
+		struct kref disconnect;
+
+		/*
+		 * This holds the reference by the
+		 * backend, the code that manages
+		 * the lifetime of the whole
+		 * struct smbdirect_socket,
+		 * if this reaches 0 it can will
+		 * be freed.
+		 *
+		 * Can be REFCOUNT_MAX is part
+		 * of another structure.
+		 *
+		 * This is equal or higher than
+		 * the disconnect refcount.
+		 */
+		struct kref destroy;
+	} refs;
+
+	/* RDMA related */
+	struct {
+		struct rdma_cm_id *cm_id;
+		/*
+		 * The expected event in our current
+		 * cm_id->event_handler, all other events
+		 * are treated as an error.
+		 */
+		enum rdma_cm_event_type expected_event;
+		/*
+		 * This is for iWarp MPA v1
+		 */
+		bool legacy_iwarp;
+	} rdma;
+
+	/* IB verbs related */
+	struct {
+		struct ib_pd *pd;
+		enum ib_poll_context poll_ctx;
+		struct ib_cq *send_cq;
+		struct ib_cq *recv_cq;
+
+		/*
+		 * shortcuts for rdma.cm_id->{qp,device};
+		 */
+		struct ib_qp *qp;
+		struct ib_device *dev;
+	} ib;
+
+	struct smbdirect_socket_parameters parameters;
+
+	/*
+	 * The state for connect/negotiation
+	 */
+	struct {
+		spinlock_t lock;
+		struct work_struct work;
+	} connect;
+
+	/*
+	 * The state for keepalive and timeout handling
+	 */
+	struct {
+		enum smbdirect_keepalive_status keepalive;
+		struct work_struct immediate_work;
+		struct delayed_work timer_work;
+	} idle;
+
+	/*
+	 * The state for listen sockets
+	 */
+	struct {
+		spinlock_t lock;
+		struct list_head pending;
+		struct list_head ready;
+		wait_queue_head_t wait_queue;
+		/*
+		 * This starts as -1 and a value != -1
+		 * means this socket was in LISTENING state
+		 * before. Note the valid backlog can
+		 * only be > 0.
+		 */
+		int backlog;
+	} listen;
+
+	/*
+	 * The state for sockets waiting
+	 * for accept, either still waiting
+	 * for the negotiation to finish
+	 * or already ready with a usable
+	 * connection.
+	 */
+	struct {
+		struct smbdirect_socket *listener;
+		struct list_head list;
+	} accept;
+
+	/*
+	 * The state for posted send buffers
+	 */
+	struct {
+		/*
+		 * Memory pools for preallocating
+		 * smbdirect_send_io buffers
+		 */
+		struct {
+			struct kmem_cache *cache;
+			mempool_t *pool;
+			gfp_t gfp_mask;
+		} mem;
+
+		/*
+		 * This is a coordination for smbdirect_send_batch.
+		 *
+		 * There's only one possible credit, which means
+		 * only one instance is running at a time.
+		 */
+		struct {
+			atomic_t count;
+			wait_queue_head_t wait_queue;
+		} bcredits;
+
+		/*
+		 * The local credit state for ib_post_send()
+		 */
+		struct {
+			atomic_t count;
+			wait_queue_head_t wait_queue;
+		} lcredits;
+
+		/*
+		 * The remote credit state for the send side
+		 */
+		struct {
+			atomic_t count;
+			wait_queue_head_t wait_queue;
+		} credits;
+
+		/*
+		 * The state about posted/pending sends
+		 */
+		struct {
+			atomic_t count;
+			/*
+			 * woken when count reached zero
+			 */
+			wait_queue_head_t zero_wait_queue;
+		} pending;
+	} send_io;
+
+	/*
+	 * The state for posted receive buffers
+	 */
+	struct {
+		/*
+		 * The type of PDU we are expecting
+		 */
+		enum {
+			SMBDIRECT_EXPECT_NEGOTIATE_REQ = 1,
+			SMBDIRECT_EXPECT_NEGOTIATE_REP = 2,
+			SMBDIRECT_EXPECT_DATA_TRANSFER = 3,
+		} expected;
+
+		/*
+		 * Memory pools for preallocating
+		 * smbdirect_recv_io buffers
+		 */
+		struct {
+			struct kmem_cache *cache;
+			mempool_t *pool;
+			gfp_t gfp_mask;
+		} mem;
+
+		/*
+		 * The list of free smbdirect_recv_io
+		 * structures
+		 */
+		struct {
+			struct list_head list;
+			spinlock_t lock;
+		} free;
+
+		/*
+		 * The state for posted recv_io messages
+		 * and the refill work struct.
+		 */
+		struct {
+			atomic_t count;
+			struct work_struct refill_work;
+		} posted;
+
+		/*
+		 * The credit state for the recv side
+		 */
+		struct {
+			u16 target;
+			atomic_t available;
+			atomic_t count;
+		} credits;
+
+		/*
+		 * The list of arrived non-empty smbdirect_recv_io
+		 * structures
+		 *
+		 * This represents the reassembly queue.
+		 */
+		struct {
+			struct list_head list;
+			spinlock_t lock;
+			wait_queue_head_t wait_queue;
+			/* total data length of reassembly queue */
+			int data_length;
+			int queue_length;
+			/* the offset to first buffer in reassembly queue */
+			int first_entry_offset;
+			/*
+			 * Indicate if we have received a full packet on the
+			 * connection This is used to identify the first SMBD
+			 * packet of a assembled payload (SMB packet) in
+			 * reassembly queue so we can return a RFC1002 length to
+			 * upper layer to indicate the length of the SMB packet
+			 * received
+			 */
+			bool full_packet_received;
+		} reassembly;
+	} recv_io;
+
+	/*
+	 * The state for Memory registrations on the client
+	 */
+	struct {
+		enum ib_mr_type type;
+
+		/*
+		 * The list of free smbdirect_mr_io
+		 * structures
+		 */
+		struct {
+			struct list_head list;
+			spinlock_t lock;
+		} all;
+
+		/*
+		 * The number of available MRs ready for memory registration
+		 */
+		struct {
+			atomic_t count;
+			wait_queue_head_t wait_queue;
+		} ready;
+
+		/*
+		 * The number of used MRs
+		 */
+		struct {
+			atomic_t count;
+		} used;
+	} mr_io;
+
+	/*
+	 * The state for RDMA read/write requests on the server
+	 */
+	struct {
+		/*
+		 * Memory hints for
+		 * smbdirect_rw_io structs
+		 */
+		struct {
+			gfp_t gfp_mask;
+		} mem;
+
+		/*
+		 * The credit state for the send side
+		 */
+		struct {
+			/*
+			 * The maximum number of rw credits
+			 */
+			size_t max;
+			/*
+			 * The number of pages per credit
+			 */
+			size_t num_pages;
+			atomic_t count;
+			wait_queue_head_t wait_queue;
+		} credits;
+	} rw_io;
+
+	/*
+	 * For debug purposes
+	 */
+	struct {
+		u64 get_receive_buffer;
+		u64 put_receive_buffer;
+		u64 enqueue_reassembly_queue;
+		u64 dequeue_reassembly_queue;
+		u64 send_empty;
+	} statistics;
+
+	struct {
+		void *private_ptr;
+		bool (*needed)(struct smbdirect_socket *sc,
+			       void *private_ptr,
+			       unsigned int lvl,
+			       unsigned int cls);
+		void (*vaprintf)(struct smbdirect_socket *sc,
+				 const char *func,
+				 unsigned int line,
+				 void *private_ptr,
+				 unsigned int lvl,
+				 unsigned int cls,
+				 struct va_format *vaf);
+	} logging;
+};
+
+static void __smbdirect_socket_disabled_work(struct work_struct *work)
+{
+	/*
+	 * Should never be called as disable_[delayed_]work_sync() was used.
+	 */
+	WARN_ON_ONCE(1);
+}
+
+static bool __smbdirect_log_needed(struct smbdirect_socket *sc,
+				   void *private_ptr,
+				   unsigned int lvl,
+				   unsigned int cls)
+{
+	/*
+	 * Should never be called, the caller should
+	 * set it's own functions.
+	 */
+	WARN_ON_ONCE(1);
+	return false;
+}
+
+static void __smbdirect_log_vaprintf(struct smbdirect_socket *sc,
+				     const char *func,
+				     unsigned int line,
+				     void *private_ptr,
+				     unsigned int lvl,
+				     unsigned int cls,
+				     struct va_format *vaf)
+{
+	/*
+	 * Should never be called, the caller should
+	 * set it's own functions.
+	 */
+	WARN_ON_ONCE(1);
+}
+
+__printf(6, 7)
+static void __smbdirect_log_printf(struct smbdirect_socket *sc,
+				   const char *func,
+				   unsigned int line,
+				   unsigned int lvl,
+				   unsigned int cls,
+				   const char *fmt,
+				   ...);
+__maybe_unused
+static void __smbdirect_log_printf(struct smbdirect_socket *sc,
+				   const char *func,
+				   unsigned int line,
+				   unsigned int lvl,
+				   unsigned int cls,
+				   const char *fmt,
+				   ...)
+{
+	struct va_format vaf;
+	va_list args;
+
+	va_start(args, fmt);
+
+	vaf.fmt = fmt;
+	vaf.va = &args;
+
+	sc->logging.vaprintf(sc,
+			     func,
+			     line,
+			     sc->logging.private_ptr,
+			     lvl,
+			     cls,
+			     &vaf);
+	va_end(args);
+}
+
+#define ___smbdirect_log_generic(sc, func, line, lvl, cls, fmt, args...) do {	\
+	if (sc->logging.needed(sc, sc->logging.private_ptr, lvl, cls)) {	\
+		__smbdirect_log_printf(sc, func, line, lvl, cls, fmt, ##args);	\
+	}									\
+} while (0)
+#define __smbdirect_log_generic(sc, lvl, cls, fmt, args...) \
+	___smbdirect_log_generic(sc, __func__, __LINE__, lvl, cls, fmt, ##args)
+
+#define smbdirect_log_outgoing(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_OUTGOING, fmt, ##args)
+#define smbdirect_log_incoming(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_INCOMING, fmt, ##args)
+#define smbdirect_log_read(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_READ, fmt, ##args)
+#define smbdirect_log_write(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_WRITE, fmt, ##args)
+#define smbdirect_log_rdma_send(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_RDMA_SEND, fmt, ##args)
+#define smbdirect_log_rdma_recv(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_RDMA_RECV, fmt, ##args)
+#define smbdirect_log_keep_alive(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_KEEP_ALIVE, fmt, ##args)
+#define smbdirect_log_rdma_event(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_RDMA_EVENT, fmt, ##args)
+#define smbdirect_log_rdma_mr(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_RDMA_MR, fmt, ##args)
+#define smbdirect_log_rdma_rw(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_RDMA_RW, fmt, ##args)
+#define smbdirect_log_negotiate(sc, lvl, fmt, args...) \
+		__smbdirect_log_generic(sc, lvl, SMBDIRECT_LOG_NEGOTIATE, fmt, ##args)
+
+static __always_inline void smbdirect_socket_init(struct smbdirect_socket *sc)
+{
+	/*
+	 * This also sets status = SMBDIRECT_SOCKET_CREATED
+	 */
+	BUILD_BUG_ON(SMBDIRECT_SOCKET_CREATED != 0);
+	memset(sc, 0, sizeof(*sc));
+
+	init_waitqueue_head(&sc->status_wait);
+
+	sc->workqueues.accept = smbdirect_globals.workqueues.accept;
+	sc->workqueues.connect = smbdirect_globals.workqueues.connect;
+	sc->workqueues.idle = smbdirect_globals.workqueues.idle;
+	sc->workqueues.refill = smbdirect_globals.workqueues.refill;
+	sc->workqueues.immediate = smbdirect_globals.workqueues.immediate;
+	sc->workqueues.cleanup = smbdirect_globals.workqueues.cleanup;
+
+	INIT_WORK(&sc->disconnect_work, __smbdirect_socket_disabled_work);
+	disable_work_sync(&sc->disconnect_work);
+
+	kref_init(&sc->refs.disconnect);
+	sc->refs.destroy = (struct kref) KREF_INIT(REFCOUNT_MAX);
+
+	sc->rdma.expected_event = RDMA_CM_EVENT_INTERNAL;
+
+	sc->ib.poll_ctx = IB_POLL_UNBOUND_WORKQUEUE;
+
+	spin_lock_init(&sc->connect.lock);
+	INIT_WORK(&sc->connect.work, __smbdirect_socket_disabled_work);
+	disable_work_sync(&sc->connect.work);
+
+	INIT_WORK(&sc->idle.immediate_work, __smbdirect_socket_disabled_work);
+	disable_work_sync(&sc->idle.immediate_work);
+	INIT_DELAYED_WORK(&sc->idle.timer_work, __smbdirect_socket_disabled_work);
+	disable_delayed_work_sync(&sc->idle.timer_work);
+
+	spin_lock_init(&sc->listen.lock);
+	INIT_LIST_HEAD(&sc->listen.pending);
+	INIT_LIST_HEAD(&sc->listen.ready);
+	sc->listen.backlog = -1; /* not a listener */
+	init_waitqueue_head(&sc->listen.wait_queue);
+
+	INIT_LIST_HEAD(&sc->accept.list);
+
+	sc->send_io.mem.gfp_mask = GFP_KERNEL;
+
+	atomic_set(&sc->send_io.bcredits.count, 0);
+	init_waitqueue_head(&sc->send_io.bcredits.wait_queue);
+
+	atomic_set(&sc->send_io.lcredits.count, 0);
+	init_waitqueue_head(&sc->send_io.lcredits.wait_queue);
+
+	atomic_set(&sc->send_io.credits.count, 0);
+	init_waitqueue_head(&sc->send_io.credits.wait_queue);
+
+	atomic_set(&sc->send_io.pending.count, 0);
+	init_waitqueue_head(&sc->send_io.pending.zero_wait_queue);
+
+	sc->recv_io.mem.gfp_mask = GFP_KERNEL;
+
+	INIT_LIST_HEAD(&sc->recv_io.free.list);
+	spin_lock_init(&sc->recv_io.free.lock);
+
+	atomic_set(&sc->recv_io.posted.count, 0);
+	INIT_WORK(&sc->recv_io.posted.refill_work, __smbdirect_socket_disabled_work);
+	disable_work_sync(&sc->recv_io.posted.refill_work);
+
+	atomic_set(&sc->recv_io.credits.available, 0);
+	atomic_set(&sc->recv_io.credits.count, 0);
+
+	INIT_LIST_HEAD(&sc->recv_io.reassembly.list);
+	spin_lock_init(&sc->recv_io.reassembly.lock);
+	init_waitqueue_head(&sc->recv_io.reassembly.wait_queue);
+
+	sc->rw_io.mem.gfp_mask = GFP_KERNEL;
+	atomic_set(&sc->rw_io.credits.count, 0);
+	init_waitqueue_head(&sc->rw_io.credits.wait_queue);
+
+	spin_lock_init(&sc->mr_io.all.lock);
+	INIT_LIST_HEAD(&sc->mr_io.all.list);
+	atomic_set(&sc->mr_io.ready.count, 0);
+	init_waitqueue_head(&sc->mr_io.ready.wait_queue);
+	atomic_set(&sc->mr_io.used.count, 0);
+
+	sc->logging.private_ptr = NULL;
+	sc->logging.needed = __smbdirect_log_needed;
+	sc->logging.vaprintf = __smbdirect_log_vaprintf;
+}
+
+#define __SMBDIRECT_CHECK_STATUS_FAILED(__sc, __expected_status, __error_cmd, __unexpected_cmd) ({ \
+	bool __failed = false; \
+	if (unlikely((__sc)->first_error)) { \
+		__failed = true; \
+		__error_cmd \
+	} else if (unlikely((__sc)->status != (__expected_status))) { \
+		__failed = true; \
+		__unexpected_cmd \
+	} \
+	__failed; \
+})
+
+#define __SMBDIRECT_CHECK_STATUS_WARN(__sc, __expected_status, __unexpected_cmd) \
+	__SMBDIRECT_CHECK_STATUS_FAILED(__sc, __expected_status, \
+	{ \
+		const struct sockaddr_storage *__src = NULL; \
+		const struct sockaddr_storage *__dst = NULL; \
+		if ((__sc)->rdma.cm_id) { \
+			__src = &(__sc)->rdma.cm_id->route.addr.src_addr; \
+			__dst = &(__sc)->rdma.cm_id->route.addr.dst_addr; \
+		} \
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO, \
+			"expected[%s] != %s first_error=%1pe local=%pISpsfc remote=%pISpsfc\n", \
+			smbdirect_socket_status_string(__expected_status), \
+			smbdirect_socket_status_string((__sc)->status), \
+			SMBDIRECT_DEBUG_ERR_PTR((__sc)->first_error), \
+			__src, __dst); \
+	}, \
+	{ \
+		const struct sockaddr_storage *__src = NULL; \
+		const struct sockaddr_storage *__dst = NULL; \
+		if ((__sc)->rdma.cm_id) { \
+			__src = &(__sc)->rdma.cm_id->route.addr.src_addr; \
+			__dst = &(__sc)->rdma.cm_id->route.addr.dst_addr; \
+		} \
+		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR, \
+			"expected[%s] != %s first_error=%1pe local=%pISpsfc remote=%pISpsfc\n", \
+			smbdirect_socket_status_string(__expected_status), \
+			smbdirect_socket_status_string((__sc)->status), \
+			SMBDIRECT_DEBUG_ERR_PTR((__sc)->first_error), \
+			__src, __dst); \
+		WARN_ONCE(1, \
+			"expected[%s] != %s first_error=%1pe local=%pISpsfc remote=%pISpsfc\n", \
+			smbdirect_socket_status_string(__expected_status), \
+			smbdirect_socket_status_string((__sc)->status), \
+			SMBDIRECT_DEBUG_ERR_PTR((__sc)->first_error), \
+			__src, __dst); \
+		__unexpected_cmd \
+	})
+
+#define SMBDIRECT_CHECK_STATUS_WARN(__sc, __expected_status) \
+	__SMBDIRECT_CHECK_STATUS_WARN(__sc, __expected_status, /* nothing */)
+
+#ifndef __SMBDIRECT_SOCKET_DISCONNECT
+#define __SMBDIRECT_SOCKET_DISCONNECT(__sc) \
+		smbdirect_socket_schedule_cleanup(__sc, -ECONNABORTED)
+#endif /* ! __SMBDIRECT_SOCKET_DISCONNECT */
+
+#define SMBDIRECT_CHECK_STATUS_DISCONNECT(__sc, __expected_status) \
+	__SMBDIRECT_CHECK_STATUS_WARN(__sc, __expected_status, \
+		__SMBDIRECT_SOCKET_DISCONNECT(__sc);)
+
+struct smbdirect_send_io {
+	struct smbdirect_socket *socket;
+	struct ib_cqe cqe;
+
+	/*
+	 * The SGE entries for this work request
+	 *
+	 * The first points to the packet header
+	 */
+#define SMBDIRECT_SEND_IO_MAX_SGE 6
+	size_t num_sge;
+	struct ib_sge sge[SMBDIRECT_SEND_IO_MAX_SGE];
+
+	/*
+	 * Link to the list of sibling smbdirect_send_io
+	 * messages.
+	 */
+	struct list_head sibling_list;
+	struct ib_send_wr wr;
+
+	/* SMBD packet header follows this structure */
+	u8 packet[];
+};
+
+struct smbdirect_send_batch {
+	/*
+	 * List of smbdirect_send_io messages
+	 */
+	struct list_head msg_list;
+	/*
+	 * Number of list entries
+	 */
+	size_t wr_cnt;
+
+	/*
+	 * Possible remote key invalidation state
+	 */
+	bool need_invalidate_rkey;
+	u32 remote_key;
+
+	int credit;
+};
+
+struct smbdirect_recv_io {
+	struct smbdirect_socket *socket;
+	struct ib_cqe cqe;
+
+	/*
+	 * For now we only use a single SGE
+	 * as we have just one large buffer
+	 * per posted recv.
+	 */
+#define SMBDIRECT_RECV_IO_MAX_SGE 1
+	struct ib_sge sge;
+
+	/* Link to free or reassembly list */
+	struct list_head list;
+
+	/* Indicate if this is the 1st packet of a payload */
+	bool first_segment;
+
+	/* SMBD packet header and payload follows this structure */
+	u8 packet[];
+};
+
+enum smbdirect_mr_state {
+	SMBDIRECT_MR_READY,
+	SMBDIRECT_MR_REGISTERED,
+	SMBDIRECT_MR_INVALIDATED,
+	SMBDIRECT_MR_ERROR,
+	SMBDIRECT_MR_DISABLED
+};
+
+struct smbdirect_mr_io {
+	struct smbdirect_socket *socket;
+	struct ib_cqe cqe;
+
+	/*
+	 * We can have up to two references:
+	 * 1. by the connection
+	 * 2. by the registration
+	 */
+	struct kref kref;
+	struct mutex mutex;
+
+	struct list_head list;
+
+	enum smbdirect_mr_state state;
+	struct ib_mr *mr;
+	struct sg_table sgt;
+	enum dma_data_direction dir;
+	union {
+		struct ib_reg_wr wr;
+		struct ib_send_wr inv_wr;
+	};
+
+	bool need_invalidate;
+	struct completion invalidate_done;
+};
+
+struct smbdirect_rw_io {
+	struct smbdirect_socket *socket;
+	struct ib_cqe cqe;
+
+	struct list_head list;
+
+	int error;
+	struct completion *completion;
+
+	struct rdma_rw_ctx rdma_ctx;
+	struct sg_table sgt;
+	struct scatterlist sg_list[];
+};
+
+static inline size_t smbdirect_get_buf_page_count(const void *buf, size_t size)
+{
+	return DIV_ROUND_UP((uintptr_t)buf + size, PAGE_SIZE) -
+		(uintptr_t)buf / PAGE_SIZE;
+}
+
+/*
+ * Maximum number of retries on data transfer operations
+ */
+#define SMBDIRECT_RDMA_CM_RETRY 6
+/*
+ * No need to retry on Receiver Not Ready since SMB_DIRECT manages credits
+ */
+#define SMBDIRECT_RDMA_CM_RNR_RETRY 0
+
+#endif /* __FS_SMB_COMMON_SMBDIRECT_SMBDIRECT_SOCKET_H__ */