[PATCH 7.2 02/10] smb: smbdirect: dont wait for RDMA_CM_EVENT_DISCONNECTED in smbdirect_socket_destroy_sync()

[PATCH 7.2 02/10] smb: smbdirect: dont wait for RDMA_CM_EVENT_DISCONNECTED in smbdirect_socket_destroy_sync()

[Greg Kroah-Hartman]

7.2-stable review patch.  If anyone has any objections, please let me know.

------------------

From: Stefan Metzmacher <metze@samba.org>

[ Upstream commit 4ca76b89917ad5b317e07ae4f0edc16f7c2a1248 ]

smbdirect_socket_destroy_sync() waited for RDMA_CM_EVENT_DISCONNECTED
after smbdirect_socket_cleanup_work() called rdma_disconnect(). That
can take very long, e.g. if the peer just disappeared, until the
rdma cm gives up, or forever if the event already happened, but the
status was overwritten afterwards.

This is not needed: smbdirect_socket_destroy() drains the qp under
rdma_lock_handler(), destroys it and calls rdma_destroy_id(), which
only waits for a currently running event handler and makes sure no
further events are delivered. The rest of the disconnect protocol
(DREQ/DREP and timewait for IB, abrupt close for iWarp) is handled
by the rdma core asynchronously. drivers/nvme/host/rdma.c also just
calls rdma_disconnect() and ib_drain_qp() before rdma_destroy_id().

So smbdirect_socket_destroy() now also accepts
SMBDIRECT_SOCKET_DISCONNECTING and changes the status to
SMBDIRECT_SOCKET_DISCONNECTED itself under rdma_lock_handler().

We could still get RDMA_CM_EVENT_DISCONNECTED in the small
windows between rdma_unlock_handler() and rdma_destroy_id(),
but in that case smbdirect_connection_rdma_event_handler()
is basically a no-op.

Fixes: 422a2436697d ("smb: smbdirect: introduce smbdirect_socket_destroy[_sync]()")
Cc: Namjae Jeon <linkinjeon@kernel.org>
Cc: Paulo Alcantara <pc@manguebit.org>
Cc: Tom Talpey <tom@talpey.com>
Cc: linux-cifs@vger.kernel.org
Cc: samba-technical@lists.samba.org
Cc: linux-rdma@vger.kernel.org
Assisted-by: Claude:claude-opus-5-5
Signed-off-by: Stefan Metzmacher <metze@samba.org>
Signed-off-by: Namjae Jeon <linkinjeon@kernel.org>
Signed-off-by: Sasha Levin <sashal@kernel.org>
---
 fs/smb/smbdirect/connection.c | 12 ++++----
 fs/smb/smbdirect/socket.c     | 56 ++++++++++++++++++++++++++++-------
 2 files changed, 52 insertions(+), 16 deletions(-)

diff --git a/fs/smb/smbdirect/connection.c b/fs/smb/smbdirect/connection.c
index fe99537207176..32327f96a03ab 100644
--- a/fs/smb/smbdirect/connection.c
+++ b/fs/smb/smbdirect/connection.c
@@ -83,9 +83,9 @@ static int smbdirect_connection_rdma_event_handler(struct rdma_cm_id *id,
 		 * 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.
+		 * As otherwise we'd set SMBDIRECT_SOCKET_DISCONNECTING
+		 * and call rdma_disconnect(), but never ever get
+		 * RDMA_CM_EVENT_DISCONNECTED.
 		 */
 		if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
 			smbdirect_socket_schedule_cleanup_status(sc,
@@ -113,9 +113,9 @@ static int smbdirect_connection_rdma_event_handler(struct rdma_cm_id *id,
 		 * 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.
+		 * As otherwise we'd set SMBDIRECT_SOCKET_DISCONNECTING
+		 * and call rdma_disconnect(), but never ever get
+		 * RDMA_CM_EVENT_DISCONNECTED.
 		 *
 		 * This is also a normal disconnect so
 		 * SMBDIRECT_LOG_INFO should be good enough
diff --git a/fs/smb/smbdirect/socket.c b/fs/smb/smbdirect/socket.c
index bb02df6158b97..c36cb7cc0088b 100644
--- a/fs/smb/smbdirect/socket.c
+++ b/fs/smb/smbdirect/socket.c
@@ -511,7 +511,13 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
 	if (sc->status == SMBDIRECT_SOCKET_DESTROYED)
 		return;
 
-	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_DISCONNECTED,
+	/*
+	 * smbdirect_socket_destroy_sync() doesn't wait
+	 * for RDMA_CM_EVENT_DISCONNECTED, so we may still
+	 * be in SMBDIRECT_SOCKET_DISCONNECTING
+	 * (or already reached SMBDIRECT_SOCKET_DISCONNECTED)
+	 */
+	WARN_ONCE(sc->status < SMBDIRECT_SOCKET_DISCONNECTING,
 		  "status=%s first_error=%1pe",
 		  smbdirect_socket_status_string(sc->status),
 		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
@@ -542,6 +548,32 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
 	if (sc->rdma.cm_id)
 		rdma_lock_handler(sc->rdma.cm_id);
 
+	/*
+	 * We hold the handler lock, so the rdma event
+	 * handlers can't change the status anymore.
+	 *
+	 * If RDMA_CM_EVENT_DISCONNECTED didn't arrive yet,
+	 * we just stop waiting for it here.
+	 *
+	 * We already disabled disconnect_work above
+	 * and before we call rdma_unlock_handler below
+	 * we call smbdirect_connection_destroy_qp which
+	 * sets sc->ib.qp = NULL.
+	 *
+	 * Between rdma_unlock_handler() and
+	 * rdma_destroy_id() below there's a small
+	 * windows where RDMA_CM_EVENT_DISCONNECTED
+	 * could still arrive.
+	 *
+	 * But smbdirect_connection_rdma_event_handler
+	 * will be a noop when calling smbdirect_socket_schedule_cleanup*
+	 * ib_drain_qp() also won't be called.
+	 */
+	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTED) {
+		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
+		smbdirect_socket_wake_up_all(sc);
+	}
+
 	if (sc->ib.qp) {
 		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
 			"drain qp\n");
@@ -679,17 +711,21 @@ void smbdirect_socket_destroy_sync(struct smbdirect_socket *sc)
 		"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()
+	 * We don't wait for RDMA_CM_EVENT_DISCONNECTED,
+	 * rdma_disconnect() was already called by
+	 * smbdirect_socket_cleanup_work() if needed
+	 * and smbdirect_socket_destroy() drains the qp,
+	 * destroys it and calls rdma_destroy_id(), which
+	 * only waits for a currently running event handler
+	 * and makes sure no further events are delivered.
+	 * The rest of the disconnect protocol is handled
+	 * by the rdma core asynchronously.
+	 *
+	 * Waiting for RDMA_CM_EVENT_DISCONNECTED could
+	 * take very long or forever, e.g. if the peer
+	 * just disappeared.
 	 */
 	smbdirect_socket_destroy(sc);
 	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
-- 
2.53.0

[PATCH 7.2 03/10] smb: smbdirect: release failed pending sockets of a listener

[Greg Kroah-Hartman]

7.2-stable review patch.  If anyone has any objections, please let me know.

------------------

From: Stefan Metzmacher <metze@samba.org>

[ Upstream commit 328c7e307e8411899e2f90beae83da9e4c45d700 ]

A child of a listener only left the listen.pending or listen.ready
list via smbdirect_socket_accept() or when the listener was
destroyed. So every connection that failed before it was accepted
(negotiate timeout, peer disconnect, invalid negotiate request, ...)
stayed on the list and counted against the backlog until the listener
was destroyed. After about backlog (10 for ksmbd) such connections
smbdirect_listen_connect_request() rejected every new connection with
-EBUSY, so an unauthenticated peer was able to permanently stop
ksmbd from accepting SMB-Direct connections.

Now a failed child moves itself to a new listen.orphaned list at the
end of smbdirect_socket_cleanup_work(), after the disconnect was
started, and queues listen.purge_orphaned_work, which releases it
with smbdirect_socket_release(), without holding the listener's
rdma_lock_handler() lock, in the same way smbdirect_socket_destroy()
does for the remaining children. Orphaned sockets still count against
the backlog until they are released.

sc->accept.listener is now only changed under listener->listen.lock
and whoever clears it is responsible for releasing the socket:
smbdirect_socket_accept(), purge_orphaned_work or the destroy of the
listener.

In order to make that work:

- a struct smbdirect_socket that is a listener is now freed via
  kfree_rcu(), so that a child can dereference sc->accept.listener
  under rcu_read_lock(), even if the listener is released concurrently.

- smbdirect_listen_connect_request() adds nsc to the listen.pending
  list and sets nsc->accept.listener before it calls
  smbdirect_accept_connect_request(), and always returns 0 so the
  rdma_cm core keeps the connection id and nsc owns it. On a non-zero
  return it schedules nsc's teardown with
  smbdirect_socket_schedule_cleanup(), which orphans it off the
  listener so that purge_orphaned_work releases it; nsc destroys its
  own connection id with rdma_destroy_id() in that teardown. The
  teardown has to be deferred because the rdma_cm core holds nsc's
  id_priv->handler_mutex across the handler, so nsc->rdma.cm_id must
  not be destroyed synchronously from here. That same handler_mutex
  (and the listener's) also guarantees nsc can't go away under us,
  so setting the listener before the call is safe and lets a failure
  orphan nsc.

- smbdirect_accept_negotiate_recv_work() no longer moves a socket to
  the ready list if it already failed or no longer belongs to the
  listener. Every accepting socket has a listener, so if
  sc->accept.listener is already cleared, it no longer sends a
  negotiate response, as the socket is about to be released.

- smbdirect_socket_destroy() of the listener disables
  purge_orphaned_work and also releases the orphaned sockets.

Fixes: dc691b91ad16 ("smb: smbdirect: introduce smbdirect_socket_{listen,accept}()")
Reported-by: Lee Seong Hyeon <ihopenre@gmail.com>
Closes: https://lore.kernel.org/linux-cifs/CANTrAmxL5sWU8Sn29JAGZvSq5PBB2OA+VKA0MHsZJMesagtfzA@mail.gmail.com/
Reported-by: Sashiko <sashiko-bot+sashiko@kernel.org>
Closes: https://sashiko.dev/#/patchset/cover.1791224972.git.metze%40samba.org
Closes: https://sashiko.dev/#/patchset/cover.1791229120.git.metze%40samba.org
Cc: Namjae Jeon <linkinjeon@kernel.org>
Cc: Paulo Alcantara <pc@manguebit.org>
Cc: Tom Talpey <tom@talpey.com>
Cc: linux-cifs@vger.kernel.org
Cc: samba-technical@lists.samba.org
Assisted-by: Claude:claude-opus-5-5
Signed-off-by: Stefan Metzmacher <metze@samba.org>
Signed-off-by: Namjae Jeon <linkinjeon@kernel.org>
Signed-off-by: Sasha Levin <sashal@kernel.org>
---
 fs/smb/smbdirect/accept.c   | 141 +++++++++++++++++++++++-------------
 fs/smb/smbdirect/internal.h |   2 +
 fs/smb/smbdirect/listen.c   | 137 ++++++++++++++++++++++++++++++++---
 fs/smb/smbdirect/socket.c   |  57 ++++++++++++++-
 fs/smb/smbdirect/socket.h   |  42 +++++++++++
 5 files changed, 314 insertions(+), 65 deletions(-)

diff --git a/fs/smb/smbdirect/accept.c b/fs/smb/smbdirect/accept.c
index 039c9bfbd8acd..66db978bd8f73 100644
--- a/fs/smb/smbdirect/accept.c
+++ b/fs/smb/smbdirect/accept.c
@@ -19,15 +19,33 @@ 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;
+	struct smbdirect_recv_io *recv_io = NULL;
 	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;
+	/*
+	 * Install the real event handler (and the event we expect next)
+	 * before anything that can fail. On an early error the caller
+	 * (smbdirect_listen_connect_request()) only schedules our deferred
+	 * cleanup and returns 0 to the rdma_cm core, which keeps the cm_id, so
+	 * the cm_id must already carry smbdirect_accept_rdma_event_handler():
+	 * it turns any unexpected event (e.g. RDMA_CM_EVENT_DEVICE_REMOVAL)
+	 * into a clean smbdirect_socket_schedule_cleanup(). Otherwise the
+	 * placeholder smbdirect_socket_rdma_event_handler() installed by
+	 * smbdirect_socket_init_accepting() stays in place and returns -ESTALE,
+	 * which makes the rdma_cm core destroy the cm_id behind the back of our
+	 * pending teardown.
+	 */
+	sc->rdma.expected_event = RDMA_CM_EVENT_ESTABLISHED;
+	sc->rdma.cm_id->event_handler = smbdirect_accept_rdma_event_handler;
+
+	if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_CREATED)) {
+		ret = -EINVAL;
+		goto cleanup;
+	}
 
 	/*
 	 * First set what the we as server are able to support
@@ -47,7 +65,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
 			"smbdirect_accept_init_params() failed %1pe\n",
 			SMBDIRECT_DEBUG_ERR_PTR(ret));
-		goto init_params_failed;
+		goto cleanup;
 	}
 
 	ret = smbdirect_connection_create_qp(sc);
@@ -55,7 +73,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
 			"smbdirect_connection_create_qp() failed %1pe\n",
 			SMBDIRECT_DEBUG_ERR_PTR(ret));
-		goto create_qp_failed;
+		goto cleanup;
 	}
 
 	ret = smbdirect_connection_create_mem_pools(sc);
@@ -63,7 +81,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 		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;
+		goto cleanup;
 	}
 
 	recv_io = smbdirect_connection_get_recv_io(sc);
@@ -72,7 +90,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 		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;
+		goto cleanup;
 	}
 	recv_io->cqe.done = smbdirect_accept_negotiate_recv_done;
 
@@ -86,7 +104,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 		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;
+		goto cleanup;
 	}
 	/*
 	 * From here recv_io is known to the RDMA QP and needs ib_drain_qp and
@@ -122,14 +140,12 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 	/* 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;
+		goto cleanup;
 	}
 
 	/*
@@ -143,31 +159,25 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
 
 	return 0;
 
-rdma_accept_failed:
+cleanup:
 	/*
-	 * 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).
+	 * No RDMA state is torn down here, and in particular sc is not
+	 * released: the caller (smbdirect_listen_connect_request()) owns that
+	 * decision and schedules sc's teardown via
+	 * smbdirect_socket_schedule_cleanup() on a non-zero return. All of the
+	 * RDMA state - the QP, the memory pools and the cm_id itself - is then
+	 * torn down asynchronously by smbdirect_socket_cleanup_work() ->
+	 * smbdirect_socket_release() -> smbdirect_socket_destroy(), which
+	 * drains the QP (returning any posted recv_io to the pool) before
+	 * destroying the memory pools and destroys the cm_id with
+	 * rdma_destroy_id(), all in the correct order.
+	 *
+	 * The only resource not reachable by that teardown is a recv_io that
+	 * was taken from the pool but not yet posted to the QP (recv_io is set
+	 * to NULL after a successful post), so hand it back here.
 	 */
-	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;
 }
 
@@ -301,6 +311,7 @@ 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_socket *lsc;
 	struct smbdirect_recv_io *recv_io;
 	struct smbdirect_negotiate_req *nreq;
 	unsigned long flags;
@@ -463,29 +474,55 @@ static void smbdirect_accept_negotiate_recv_work(struct work_struct *work)
 	 */
 	sp->max_fragmented_send_size = max_fragmented_size;
 
-	if (sc->accept.listener) {
-		struct smbdirect_socket *lsc = sc->accept.listener;
-		unsigned long flags;
+	/*
+	 * Every accepting socket was created by
+	 * smbdirect_listen_connect_request() and has a
+	 * listener. If it's already cleared, the listener
+	 * (or its purge_orphaned_work) is responsible for
+	 * releasing us, so we must not send a negotiate
+	 * response.
+	 *
+	 * The memory of the listener is freed via kfree_rcu(),
+	 * see smbdirect_listen_orphan_socket().
+	 */
+	rcu_read_lock();
+	lsc = READ_ONCE(sc->accept.listener);
+	if (!lsc) {
+		rcu_read_unlock();
+		return;
+	}
 
-		spin_lock_irqsave(&lsc->listen.lock, flags);
-		list_del(&sc->accept.list);
-		list_add_tail(&sc->accept.list, &lsc->listen.ready);
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	/*
+	 * The listener (or its purge_orphaned_work)
+	 * may have cleared sc->accept.listener in the
+	 * meantime and is responsible for releasing us.
+	 *
+	 * If we already failed, we're either
+	 * already on the orphaned list or
+	 * smbdirect_socket_cleanup_work() will
+	 * move us there.
+	 *
+	 * In both cases we must not move us
+	 * to the ready list.
+	 */
+	if (sc->accept.listener == lsc && !READ_ONCE(sc->first_error)) {
+		list_move_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;
 	}
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+	rcu_read_unlock();
 
-	ntstatus = le32_to_cpu(STATUS_SUCCESS);
+	/*
+	 * 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;
 
 not_supported:
 	smbdirect_accept_negotiate_finish(sc, ntstatus);
@@ -838,7 +875,7 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
 				       struct smbdirect_socket,
 				       accept.list);
 	if (nsc) {
-		nsc->accept.listener = NULL;
+		WRITE_ONCE(nsc->accept.listener, NULL);
 		list_del_init_careful(&nsc->accept.list);
 		arg->is_empty = list_empty_careful(&lsc->listen.ready);
 	}
diff --git a/fs/smb/smbdirect/internal.h b/fs/smb/smbdirect/internal.h
index e9959e6dc13ae..69d0c9f899dec 100644
--- a/fs/smb/smbdirect/internal.h
+++ b/fs/smb/smbdirect/internal.h
@@ -74,6 +74,8 @@ void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
 
 void smbdirect_socket_destroy_sync(struct smbdirect_socket *sc);
 
+void smbdirect_listen_orphan_socket(struct smbdirect_socket *sc);
+
 int smbdirect_socket_wait_for_credits(struct smbdirect_socket *sc,
 				      enum smbdirect_socket_status expected_status,
 				      int unexpected_errno,
diff --git a/fs/smb/smbdirect/listen.c b/fs/smb/smbdirect/listen.c
index 2f78bcaedbf82..3c91f00d95ce4 100644
--- a/fs/smb/smbdirect/listen.c
+++ b/fs/smb/smbdirect/listen.c
@@ -10,6 +10,78 @@
 static int smbdirect_listen_rdma_event_handler(struct rdma_cm_id *id,
 					       struct rdma_cm_event *event);
 
+/*
+ * This is called by a socket that failed while it
+ * is still on the pending or ready list of its
+ * listener, typically at the end of
+ * smbdirect_socket_cleanup_work(), when the
+ * disconnect was already started.
+ *
+ * It moves itself to the orphaned list and
+ * lets smbdirect_listen_purge_orphaned_work()
+ * release it. Otherwise it would stay on the
+ * pending or ready list until the listener is
+ * destroyed and fill up the backlog, so that
+ * no new connections would be accepted.
+ *
+ * It's fine to call this more than once.
+ */
+void smbdirect_listen_orphan_socket(struct smbdirect_socket *sc)
+{
+	struct smbdirect_socket *lsc;
+	unsigned long flags;
+
+	/*
+	 * The memory of the listener is freed via
+	 * kfree_rcu(), so it's safe to dereference it
+	 * under rcu_read_lock(), even if sc->accept.listener
+	 * is cleared and the listener is released concurrently.
+	 */
+	rcu_read_lock();
+	lsc = READ_ONCE(sc->accept.listener);
+	if (!lsc) {
+		rcu_read_unlock();
+		return;
+	}
+
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	if (sc->accept.listener == lsc) {
+		list_move_tail(&sc->accept.list, &lsc->listen.orphaned);
+		queue_work(lsc->workqueues.cleanup, &lsc->listen.purge_orphaned_work);
+	}
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+	rcu_read_unlock();
+}
+
+static void smbdirect_listen_purge_orphaned_work(struct work_struct *work)
+{
+	struct smbdirect_socket *lsc =
+		container_of(work, struct smbdirect_socket, listen.purge_orphaned_work);
+	struct smbdirect_socket *psc, *tsc;
+	LIST_HEAD(orphaned_list);
+	unsigned long flags;
+
+	/*
+	 * Clearing accept.listener under listen.lock
+	 * makes us responsible for releasing them.
+	 */
+	spin_lock_irqsave(&lsc->listen.lock, flags);
+	list_splice_tail_init(&lsc->listen.orphaned, &orphaned_list);
+	list_for_each_entry(psc, &orphaned_list, accept.list)
+		WRITE_ONCE(psc->accept.listener, NULL);
+	spin_unlock_irqrestore(&lsc->listen.lock, flags);
+
+	/*
+	 * We don't hold the listener's rdma_lock_handler()
+	 * lock here, see smbdirect_socket_destroy()
+	 * for why that's important.
+	 */
+	list_for_each_entry_safe(psc, tsc, &orphaned_list, accept.list) {
+		list_del_init(&psc->accept.list);
+		smbdirect_socket_release(psc);
+	}
+}
+
 int smbdirect_socket_listen(struct smbdirect_socket *sc, int backlog)
 {
 	int ret;
@@ -47,6 +119,9 @@ int smbdirect_socket_listen(struct smbdirect_socket *sc, int backlog)
 	sc->rdma.cm_id->event_handler = smbdirect_listen_rdma_event_handler;
 	rdma_unlock_handler(sc->rdma.cm_id);
 
+	INIT_WORK(&sc->listen.purge_orphaned_work,
+		  smbdirect_listen_purge_orphaned_work);
+
 	ret = rdma_listen(sc->rdma.cm_id, backlog);
 	if (ret) {
 		sc->first_error = ret;
@@ -214,6 +289,7 @@ static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
 	size_t backlog = max_t(size_t, 1, lsc->listen.backlog);
 	size_t psockets;
 	size_t rsockets;
+	size_t osockets;
 	int ret;
 
 	if (!smbdirect_frwr_is_supported(&new_id->device->attrs)) {
@@ -254,14 +330,21 @@ static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
 	spin_lock_irqsave(&lsc->listen.lock, flags);
 	psockets = list_count_nodes(&lsc->listen.pending);
 	rsockets = list_count_nodes(&lsc->listen.ready);
+	/*
+	 * Orphaned sockets still count against
+	 * the backlog until they are released by
+	 * smbdirect_listen_purge_orphaned_work().
+	 */
+	osockets = list_count_nodes(&lsc->listen.orphaned);
 	spin_unlock_irqrestore(&lsc->listen.lock, flags);
 
 	if (psockets > backlog ||
 	    rsockets > backlog ||
-	    (psockets + rsockets) > backlog) {
+	    osockets > backlog ||
+	    (psockets + rsockets + osockets) > backlog) {
 		smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
-			"Backlog[%d][%zu] full pending[%zu] ready[%zu]\n",
-			lsc->listen.backlog, backlog, psockets, rsockets);
+			"Backlog[%d][%zu] full pending[%zu] ready[%zu] orphaned[%zu]\n",
+			lsc->listen.backlog, backlog, psockets, rsockets, osockets);
 		return -EBUSY;
 	}
 
@@ -279,22 +362,54 @@ static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
 	if (ret)
 		goto set_settings_failed;
 
+	/*
+	 * Publish nsc on the pending list with accept.listener set before
+	 * smbdirect_accept_connect_request() calls rdma_accept(). Once the
+	 * connection can establish, smbdirect_accept_negotiate_recv_work()
+	 * may run and it must observe accept.listener, otherwise nsc would
+	 * be left stranded on the pending list forever.
+	 *
+	 * The listener's handler_mutex is held while we're called, so
+	 * smbdirect_socket_destroy() of the listener can't reach nsc on the
+	 * pending list before we're done.
+	 *
+	 * From here nsc is published on the pending list: on failure of
+	 * smbdirect_accept_connect_request() below we schedule nsc's teardown,
+	 * which orphans nsc off the listener so
+	 * smbdirect_listen_purge_orphaned_work() -> smbdirect_socket_release()
+	 * releases it.
+	 */
 	spin_lock_irqsave(&lsc->listen.lock, flags);
 	list_add_tail(&nsc->accept.list, &lsc->listen.pending);
-	nsc->accept.listener = lsc;
+	WRITE_ONCE(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;
+	if (ret) {
+		/*
+		 * The rdma_cm core holds both the listener's and nsc's
+		 * id_priv->handler_mutex across this CONNECT_REQUEST handler,
+		 * so neither can go away under us here and nsc cannot receive
+		 * any rdma event: it is safe to hand nsc to its teardown.
+		 *
+		 * That teardown must be deferred though: nsc->rdma.cm_id
+		 * (= new_id) has its handler_mutex held by us, so it cannot be
+		 * destroyed synchronously from here.
+		 * smbdirect_socket_schedule_cleanup() only queues work;
+		 * smbdirect_socket_cleanup_work() then orphans nsc off the
+		 * listener and smbdirect_listen_purge_orphaned_work() releases
+		 * it, destroying nsc's cm_id with rdma_destroy_id().
+		 */
+		smbdirect_socket_schedule_cleanup(nsc, ret);
+	}
 
+	/*
+	 * Always return 0 so the rdma_cm core keeps new_id: nsc owns it now.
+	 * On success it stays connected; on failure its deferred teardown
+	 * above destroys it.
+	 */
 	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:
 	/*
diff --git a/fs/smb/smbdirect/socket.c b/fs/smb/smbdirect/socket.c
index c36cb7cc0088b..e270e36b3145e 100644
--- a/fs/smb/smbdirect/socket.c
+++ b/fs/smb/smbdirect/socket.c
@@ -319,6 +319,15 @@ void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
 	 * (smbdirect_socket_destroy) to reap.
 	 */
 	if (sc->listen.backlog != -1) { /* was a listener */
+		/*
+		 * We don't move them to the orphaned list here,
+		 * each child does that itself at the end of its
+		 * smbdirect_socket_cleanup_work(), see
+		 * smbdirect_listen_orphan_socket(). Only that checks
+		 * accept.listener, which is still NULL for a child
+		 * that smbdirect_listen_connect_request() is still
+		 * setting up and will release itself on failure.
+		 */
 		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)
@@ -427,6 +436,15 @@ static void smbdirect_socket_cleanup_work(struct work_struct *work)
 	 * instances of one class -- harmless, but lockdep cannot tell).
 	 */
 	if (sc->listen.backlog != -1) { /* was a listener */
+		/*
+		 * We don't move them to the orphaned list here,
+		 * each child does that itself at the end of its
+		 * smbdirect_socket_cleanup_work(), see
+		 * smbdirect_listen_orphan_socket(). Only that checks
+		 * accept.listener, which is still NULL for a child
+		 * that smbdirect_listen_connect_request() is still
+		 * setting up and will release itself on failure.
+		 */
 		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)
@@ -486,6 +504,18 @@ static void smbdirect_socket_cleanup_work(struct work_struct *work)
 	 * in order to notice the broken connection.
 	 */
 	smbdirect_socket_wake_up_all(sc);
+
+	/*
+	 * If we're still on the pending or ready list
+	 * of a listener, we started the disconnect
+	 * as far as possible above, so we move ourself
+	 * to the orphaned list of the listener,
+	 * which will release us.
+	 *
+	 * This is a no-op internally if
+	 * sc->accept.listener is NULL.
+	 */
+	smbdirect_listen_orphan_socket(sc);
 }
 
 static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
@@ -544,6 +574,7 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
 	disable_work_sync(&sc->recv_io.posted.refill_work);
 	disable_work_sync(&sc->idle.immediate_work);
 	disable_delayed_work_sync(&sc->idle.timer_work);
+	disable_work_sync(&sc->listen.purge_orphaned_work);
 
 	if (sc->rdma.cm_id)
 		rdma_lock_handler(sc->rdma.cm_id);
@@ -607,6 +638,19 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
 	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);
+	/*
+	 * purge_orphaned_work is already disabled above,
+	 * so we also need to release the orphaned sockets.
+	 *
+	 * Clearing accept.listener under listen.lock
+	 * makes us responsible for releasing them and
+	 * prevents them from moving themselves to
+	 * the orphaned list via
+	 * smbdirect_listen_orphan_socket().
+	 */
+	list_splice_tail_init(&sc->listen.orphaned, &pending_list);
+	list_for_each_entry(psc, &pending_list, accept.list)
+		WRITE_ONCE(psc->accept.listener, NULL);
 	spin_unlock_irqrestore(&sc->listen.lock, flags);
 
 	/* It's not possible for upper layer to get to reassembly */
@@ -650,7 +694,6 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
 			"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 */
@@ -777,7 +820,17 @@ static void smbdirect_socket_release_destroy(struct kref *kref)
 	 * in DESTROYED state, before we free the memory.
 	 */
 	smbdirect_socket_destroy_sync(sc);
-	kfree(sc);
+
+	/*
+	 * Only a listener (backlog != -1) is ever dereferenced
+	 * via sc->accept.listener under rcu_read_lock(), see
+	 * smbdirect_listen_orphan_socket(). Other sockets can be
+	 * freed immediately.
+	 */
+	if (sc->listen.backlog != -1) /* was a listener */
+		kfree_rcu(sc, refs.rcu);
+	else
+		kfree(sc);
 }
 
 void smbdirect_socket_release(struct smbdirect_socket *sc)
diff --git a/fs/smb/smbdirect/socket.h b/fs/smb/smbdirect/socket.h
index c09eddd8ad16d..14114a2567a2e 100644
--- a/fs/smb/smbdirect/socket.h
+++ b/fs/smb/smbdirect/socket.h
@@ -151,6 +151,12 @@ struct smbdirect_socket {
 		 * the disconnect refcount.
 		 */
 		struct kref destroy;
+
+		/*
+		 * smbdirect_socket_release_destroy() uses
+		 * kfree_rcu(), see accept.listener.
+		 */
+		struct rcu_head rcu;
 	} refs;
 
 	/* RDMA related */
@@ -216,6 +222,16 @@ struct smbdirect_socket {
 		 * only be > 0.
 		 */
 		int backlog;
+		/*
+		 * Sockets on pending or ready that failed
+		 * move themselves to orphaned and queue
+		 * purge_orphaned_work, which releases them.
+		 * So that they are freed while the listener
+		 * is still alive. They still count against
+		 * the backlog until they are released.
+		 */
+		struct list_head orphaned;
+		struct work_struct purge_orphaned_work;
 	} listen;
 
 	/*
@@ -226,7 +242,30 @@ struct smbdirect_socket {
 	 * connection.
 	 */
 	struct {
+		/*
+		 * This is only set, protected by
+		 * listener->listen.lock, while the socket
+		 * is owned by the listener (on its pending,
+		 * ready or orphaned list). The one who clears
+		 * it is responsible for releasing the socket.
+		 *
+		 * The memory of a struct smbdirect_socket is
+		 * freed via kfree_rcu(), so the listener
+		 * can be dereferenced under rcu_read_lock(),
+		 * even if accept.listener is cleared and
+		 * the listener is released concurrently.
+		 */
 		struct smbdirect_socket *listener;
+		/*
+		 * Linkage on one of the listener's
+		 * listen.{pending,ready,orphaned} lists.
+		 *
+		 * accept.list and accept.listener are set
+		 * together under listener->listen.lock before
+		 * smbdirect_accept_connect_request() is called,
+		 * so whenever the socket is on one of those lists
+		 * accept.listener is that listener (never NULL).
+		 */
 		struct list_head list;
 	} accept;
 
@@ -588,6 +627,9 @@ static __always_inline void smbdirect_socket_init(struct smbdirect_socket *sc)
 	spin_lock_init(&sc->listen.lock);
 	INIT_LIST_HEAD(&sc->listen.pending);
 	INIT_LIST_HEAD(&sc->listen.ready);
+	INIT_LIST_HEAD(&sc->listen.orphaned);
+	INIT_WORK(&sc->listen.purge_orphaned_work, __smbdirect_socket_disabled_work);
+	disable_work_sync(&sc->listen.purge_orphaned_work);
 	sc->listen.backlog = -1; /* not a listener */
 	init_waitqueue_head(&sc->listen.wait_queue);
 
-- 
2.53.0

Re: [PATCH 7.2 03/10] smb: smbdirect: release failed pending sockets of a listener

[이성현]

Hi Greg and Stefan,

Thank you for fixing the SMB Direct listener backlog issue and
for considering the fix for 7.2-stable.

I also appreciate you preserving my Reported-by credit.

Best regards,
Lee Seong Hyeon

On Fri, Oct 30, 2026 at 7:15 PM Greg Kroah-Hartman
<gregkh@linuxfoundation.org> wrote:
>
> 7.2-stable review patch.  If anyone has any objections, please let me know.
>
> ------------------
>
> From: Stefan Metzmacher <metze@samba.org>
>
> [ Upstream commit 328c7e307e8411899e2f90beae83da9e4c45d700 ]
>
> A child of a listener only left the listen.pending or listen.ready
> list via smbdirect_socket_accept() or when the listener was
> destroyed. So every connection that failed before it was accepted
> (negotiate timeout, peer disconnect, invalid negotiate request, ...)
> stayed on the list and counted against the backlog until the listener
> was destroyed. After about backlog (10 for ksmbd) such connections
> smbdirect_listen_connect_request() rejected every new connection with
> -EBUSY, so an unauthenticated peer was able to permanently stop
> ksmbd from accepting SMB-Direct connections.
>
> Now a failed child moves itself to a new listen.orphaned list at the
> end of smbdirect_socket_cleanup_work(), after the disconnect was
> started, and queues listen.purge_orphaned_work, which releases it
> with smbdirect_socket_release(), without holding the listener's
> rdma_lock_handler() lock, in the same way smbdirect_socket_destroy()
> does for the remaining children. Orphaned sockets still count against
> the backlog until they are released.
>
> sc->accept.listener is now only changed under listener->listen.lock
> and whoever clears it is responsible for releasing the socket:
> smbdirect_socket_accept(), purge_orphaned_work or the destroy of the
> listener.
>
> In order to make that work:
>
> - a struct smbdirect_socket that is a listener is now freed via
>   kfree_rcu(), so that a child can dereference sc->accept.listener
>   under rcu_read_lock(), even if the listener is released concurrently.
>
> - smbdirect_listen_connect_request() adds nsc to the listen.pending
>   list and sets nsc->accept.listener before it calls
>   smbdirect_accept_connect_request(), and always returns 0 so the
>   rdma_cm core keeps the connection id and nsc owns it. On a non-zero
>   return it schedules nsc's teardown with
>   smbdirect_socket_schedule_cleanup(), which orphans it off the
>   listener so that purge_orphaned_work releases it; nsc destroys its
>   own connection id with rdma_destroy_id() in that teardown. The
>   teardown has to be deferred because the rdma_cm core holds nsc's
>   id_priv->handler_mutex across the handler, so nsc->rdma.cm_id must
>   not be destroyed synchronously from here. That same handler_mutex
>   (and the listener's) also guarantees nsc can't go away under us,
>   so setting the listener before the call is safe and lets a failure
>   orphan nsc.
>
> - smbdirect_accept_negotiate_recv_work() no longer moves a socket to
>   the ready list if it already failed or no longer belongs to the
>   listener. Every accepting socket has a listener, so if
>   sc->accept.listener is already cleared, it no longer sends a
>   negotiate response, as the socket is about to be released.
>
> - smbdirect_socket_destroy() of the listener disables
>   purge_orphaned_work and also releases the orphaned sockets.
>
> Fixes: dc691b91ad16 ("smb: smbdirect: introduce smbdirect_socket_{listen,accept}()")
> Reported-by: Lee Seong Hyeon <ihopenre@gmail.com>
> Closes: https://lore.kernel.org/linux-cifs/CANTrAmxL5sWU8Sn29JAGZvSq5PBB2OA+VKA0MHsZJMesagtfzA@mail.gmail.com/
> Reported-by: Sashiko <sashiko-bot+sashiko@kernel.org>
> Closes: https://sashiko.dev/#/patchset/cover.1791224972.git.metze%40samba.org
> Closes: https://sashiko.dev/#/patchset/cover.1791229120.git.metze%40samba.org
> Cc: Namjae Jeon <linkinjeon@kernel.org>
> Cc: Paulo Alcantara <pc@manguebit.org>
> Cc: Tom Talpey <tom@talpey.com>
> Cc: linux-cifs@vger.kernel.org
> Cc: samba-technical@lists.samba.org
> Assisted-by: Claude:claude-opus-5-5
> Signed-off-by: Stefan Metzmacher <metze@samba.org>
> Signed-off-by: Namjae Jeon <linkinjeon@kernel.org>
> Signed-off-by: Sasha Levin <sashal@kernel.org>
> ---
>  fs/smb/smbdirect/accept.c   | 141 +++++++++++++++++++++++-------------
>  fs/smb/smbdirect/internal.h |   2 +
>  fs/smb/smbdirect/listen.c   | 137 ++++++++++++++++++++++++++++++++---
>  fs/smb/smbdirect/socket.c   |  57 ++++++++++++++-
>  fs/smb/smbdirect/socket.h   |  42 +++++++++++
>  5 files changed, 314 insertions(+), 65 deletions(-)
>
> diff --git a/fs/smb/smbdirect/accept.c b/fs/smb/smbdirect/accept.c
> index 039c9bfbd8acd..66db978bd8f73 100644
> --- a/fs/smb/smbdirect/accept.c
> +++ b/fs/smb/smbdirect/accept.c
> @@ -19,15 +19,33 @@ 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;
> +       struct smbdirect_recv_io *recv_io = NULL;
>         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;
> +       /*
> +        * Install the real event handler (and the event we expect next)
> +        * before anything that can fail. On an early error the caller
> +        * (smbdirect_listen_connect_request()) only schedules our deferred
> +        * cleanup and returns 0 to the rdma_cm core, which keeps the cm_id, so
> +        * the cm_id must already carry smbdirect_accept_rdma_event_handler():
> +        * it turns any unexpected event (e.g. RDMA_CM_EVENT_DEVICE_REMOVAL)
> +        * into a clean smbdirect_socket_schedule_cleanup(). Otherwise the
> +        * placeholder smbdirect_socket_rdma_event_handler() installed by
> +        * smbdirect_socket_init_accepting() stays in place and returns -ESTALE,
> +        * which makes the rdma_cm core destroy the cm_id behind the back of our
> +        * pending teardown.
> +        */
> +       sc->rdma.expected_event = RDMA_CM_EVENT_ESTABLISHED;
> +       sc->rdma.cm_id->event_handler = smbdirect_accept_rdma_event_handler;
> +
> +       if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_CREATED)) {
> +               ret = -EINVAL;
> +               goto cleanup;
> +       }
>
>         /*
>          * First set what the we as server are able to support
> @@ -47,7 +65,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>                 smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
>                         "smbdirect_accept_init_params() failed %1pe\n",
>                         SMBDIRECT_DEBUG_ERR_PTR(ret));
> -               goto init_params_failed;
> +               goto cleanup;
>         }
>
>         ret = smbdirect_connection_create_qp(sc);
> @@ -55,7 +73,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>                 smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
>                         "smbdirect_connection_create_qp() failed %1pe\n",
>                         SMBDIRECT_DEBUG_ERR_PTR(ret));
> -               goto create_qp_failed;
> +               goto cleanup;
>         }
>
>         ret = smbdirect_connection_create_mem_pools(sc);
> @@ -63,7 +81,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>                 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;
> +               goto cleanup;
>         }
>
>         recv_io = smbdirect_connection_get_recv_io(sc);
> @@ -72,7 +90,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>                 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;
> +               goto cleanup;
>         }
>         recv_io->cqe.done = smbdirect_accept_negotiate_recv_done;
>
> @@ -86,7 +104,7 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>                 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;
> +               goto cleanup;
>         }
>         /*
>          * From here recv_io is known to the RDMA QP and needs ib_drain_qp and
> @@ -122,14 +140,12 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>         /* 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;
> +               goto cleanup;
>         }
>
>         /*
> @@ -143,31 +159,25 @@ int smbdirect_accept_connect_request(struct smbdirect_socket *sc,
>
>         return 0;
>
> -rdma_accept_failed:
> +cleanup:
>         /*
> -        * 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).
> +        * No RDMA state is torn down here, and in particular sc is not
> +        * released: the caller (smbdirect_listen_connect_request()) owns that
> +        * decision and schedules sc's teardown via
> +        * smbdirect_socket_schedule_cleanup() on a non-zero return. All of the
> +        * RDMA state - the QP, the memory pools and the cm_id itself - is then
> +        * torn down asynchronously by smbdirect_socket_cleanup_work() ->
> +        * smbdirect_socket_release() -> smbdirect_socket_destroy(), which
> +        * drains the QP (returning any posted recv_io to the pool) before
> +        * destroying the memory pools and destroys the cm_id with
> +        * rdma_destroy_id(), all in the correct order.
> +        *
> +        * The only resource not reachable by that teardown is a recv_io that
> +        * was taken from the pool but not yet posted to the QP (recv_io is set
> +        * to NULL after a successful post), so hand it back here.
>          */
> -       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;
>  }
>
> @@ -301,6 +311,7 @@ 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_socket *lsc;
>         struct smbdirect_recv_io *recv_io;
>         struct smbdirect_negotiate_req *nreq;
>         unsigned long flags;
> @@ -463,29 +474,55 @@ static void smbdirect_accept_negotiate_recv_work(struct work_struct *work)
>          */
>         sp->max_fragmented_send_size = max_fragmented_size;
>
> -       if (sc->accept.listener) {
> -               struct smbdirect_socket *lsc = sc->accept.listener;
> -               unsigned long flags;
> +       /*
> +        * Every accepting socket was created by
> +        * smbdirect_listen_connect_request() and has a
> +        * listener. If it's already cleared, the listener
> +        * (or its purge_orphaned_work) is responsible for
> +        * releasing us, so we must not send a negotiate
> +        * response.
> +        *
> +        * The memory of the listener is freed via kfree_rcu(),
> +        * see smbdirect_listen_orphan_socket().
> +        */
> +       rcu_read_lock();
> +       lsc = READ_ONCE(sc->accept.listener);
> +       if (!lsc) {
> +               rcu_read_unlock();
> +               return;
> +       }
>
> -               spin_lock_irqsave(&lsc->listen.lock, flags);
> -               list_del(&sc->accept.list);
> -               list_add_tail(&sc->accept.list, &lsc->listen.ready);
> +       spin_lock_irqsave(&lsc->listen.lock, flags);
> +       /*
> +        * The listener (or its purge_orphaned_work)
> +        * may have cleared sc->accept.listener in the
> +        * meantime and is responsible for releasing us.
> +        *
> +        * If we already failed, we're either
> +        * already on the orphaned list or
> +        * smbdirect_socket_cleanup_work() will
> +        * move us there.
> +        *
> +        * In both cases we must not move us
> +        * to the ready list.
> +        */
> +       if (sc->accept.listener == lsc && !READ_ONCE(sc->first_error)) {
> +               list_move_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;
>         }
> +       spin_unlock_irqrestore(&lsc->listen.lock, flags);
> +       rcu_read_unlock();
>
> -       ntstatus = le32_to_cpu(STATUS_SUCCESS);
> +       /*
> +        * 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;
>
>  not_supported:
>         smbdirect_accept_negotiate_finish(sc, ntstatus);
> @@ -838,7 +875,7 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
>                                        struct smbdirect_socket,
>                                        accept.list);
>         if (nsc) {
> -               nsc->accept.listener = NULL;
> +               WRITE_ONCE(nsc->accept.listener, NULL);
>                 list_del_init_careful(&nsc->accept.list);
>                 arg->is_empty = list_empty_careful(&lsc->listen.ready);
>         }
> diff --git a/fs/smb/smbdirect/internal.h b/fs/smb/smbdirect/internal.h
> index e9959e6dc13ae..69d0c9f899dec 100644
> --- a/fs/smb/smbdirect/internal.h
> +++ b/fs/smb/smbdirect/internal.h
> @@ -74,6 +74,8 @@ void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
>
>  void smbdirect_socket_destroy_sync(struct smbdirect_socket *sc);
>
> +void smbdirect_listen_orphan_socket(struct smbdirect_socket *sc);
> +
>  int smbdirect_socket_wait_for_credits(struct smbdirect_socket *sc,
>                                       enum smbdirect_socket_status expected_status,
>                                       int unexpected_errno,
> diff --git a/fs/smb/smbdirect/listen.c b/fs/smb/smbdirect/listen.c
> index 2f78bcaedbf82..3c91f00d95ce4 100644
> --- a/fs/smb/smbdirect/listen.c
> +++ b/fs/smb/smbdirect/listen.c
> @@ -10,6 +10,78 @@
>  static int smbdirect_listen_rdma_event_handler(struct rdma_cm_id *id,
>                                                struct rdma_cm_event *event);
>
> +/*
> + * This is called by a socket that failed while it
> + * is still on the pending or ready list of its
> + * listener, typically at the end of
> + * smbdirect_socket_cleanup_work(), when the
> + * disconnect was already started.
> + *
> + * It moves itself to the orphaned list and
> + * lets smbdirect_listen_purge_orphaned_work()
> + * release it. Otherwise it would stay on the
> + * pending or ready list until the listener is
> + * destroyed and fill up the backlog, so that
> + * no new connections would be accepted.
> + *
> + * It's fine to call this more than once.
> + */
> +void smbdirect_listen_orphan_socket(struct smbdirect_socket *sc)
> +{
> +       struct smbdirect_socket *lsc;
> +       unsigned long flags;
> +
> +       /*
> +        * The memory of the listener is freed via
> +        * kfree_rcu(), so it's safe to dereference it
> +        * under rcu_read_lock(), even if sc->accept.listener
> +        * is cleared and the listener is released concurrently.
> +        */
> +       rcu_read_lock();
> +       lsc = READ_ONCE(sc->accept.listener);
> +       if (!lsc) {
> +               rcu_read_unlock();
> +               return;
> +       }
> +
> +       spin_lock_irqsave(&lsc->listen.lock, flags);
> +       if (sc->accept.listener == lsc) {
> +               list_move_tail(&sc->accept.list, &lsc->listen.orphaned);
> +               queue_work(lsc->workqueues.cleanup, &lsc->listen.purge_orphaned_work);
> +       }
> +       spin_unlock_irqrestore(&lsc->listen.lock, flags);
> +       rcu_read_unlock();
> +}
> +
> +static void smbdirect_listen_purge_orphaned_work(struct work_struct *work)
> +{
> +       struct smbdirect_socket *lsc =
> +               container_of(work, struct smbdirect_socket, listen.purge_orphaned_work);
> +       struct smbdirect_socket *psc, *tsc;
> +       LIST_HEAD(orphaned_list);
> +       unsigned long flags;
> +
> +       /*
> +        * Clearing accept.listener under listen.lock
> +        * makes us responsible for releasing them.
> +        */
> +       spin_lock_irqsave(&lsc->listen.lock, flags);
> +       list_splice_tail_init(&lsc->listen.orphaned, &orphaned_list);
> +       list_for_each_entry(psc, &orphaned_list, accept.list)
> +               WRITE_ONCE(psc->accept.listener, NULL);
> +       spin_unlock_irqrestore(&lsc->listen.lock, flags);
> +
> +       /*
> +        * We don't hold the listener's rdma_lock_handler()
> +        * lock here, see smbdirect_socket_destroy()
> +        * for why that's important.
> +        */
> +       list_for_each_entry_safe(psc, tsc, &orphaned_list, accept.list) {
> +               list_del_init(&psc->accept.list);
> +               smbdirect_socket_release(psc);
> +       }
> +}
> +
>  int smbdirect_socket_listen(struct smbdirect_socket *sc, int backlog)
>  {
>         int ret;
> @@ -47,6 +119,9 @@ int smbdirect_socket_listen(struct smbdirect_socket *sc, int backlog)
>         sc->rdma.cm_id->event_handler = smbdirect_listen_rdma_event_handler;
>         rdma_unlock_handler(sc->rdma.cm_id);
>
> +       INIT_WORK(&sc->listen.purge_orphaned_work,
> +                 smbdirect_listen_purge_orphaned_work);
> +
>         ret = rdma_listen(sc->rdma.cm_id, backlog);
>         if (ret) {
>                 sc->first_error = ret;
> @@ -214,6 +289,7 @@ static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
>         size_t backlog = max_t(size_t, 1, lsc->listen.backlog);
>         size_t psockets;
>         size_t rsockets;
> +       size_t osockets;
>         int ret;
>
>         if (!smbdirect_frwr_is_supported(&new_id->device->attrs)) {
> @@ -254,14 +330,21 @@ static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
>         spin_lock_irqsave(&lsc->listen.lock, flags);
>         psockets = list_count_nodes(&lsc->listen.pending);
>         rsockets = list_count_nodes(&lsc->listen.ready);
> +       /*
> +        * Orphaned sockets still count against
> +        * the backlog until they are released by
> +        * smbdirect_listen_purge_orphaned_work().
> +        */
> +       osockets = list_count_nodes(&lsc->listen.orphaned);
>         spin_unlock_irqrestore(&lsc->listen.lock, flags);
>
>         if (psockets > backlog ||
>             rsockets > backlog ||
> -           (psockets + rsockets) > backlog) {
> +           osockets > backlog ||
> +           (psockets + rsockets + osockets) > backlog) {
>                 smbdirect_log_rdma_event(lsc, SMBDIRECT_LOG_ERR,
> -                       "Backlog[%d][%zu] full pending[%zu] ready[%zu]\n",
> -                       lsc->listen.backlog, backlog, psockets, rsockets);
> +                       "Backlog[%d][%zu] full pending[%zu] ready[%zu] orphaned[%zu]\n",
> +                       lsc->listen.backlog, backlog, psockets, rsockets, osockets);
>                 return -EBUSY;
>         }
>
> @@ -279,22 +362,54 @@ static int smbdirect_listen_connect_request(struct smbdirect_socket *lsc,
>         if (ret)
>                 goto set_settings_failed;
>
> +       /*
> +        * Publish nsc on the pending list with accept.listener set before
> +        * smbdirect_accept_connect_request() calls rdma_accept(). Once the
> +        * connection can establish, smbdirect_accept_negotiate_recv_work()
> +        * may run and it must observe accept.listener, otherwise nsc would
> +        * be left stranded on the pending list forever.
> +        *
> +        * The listener's handler_mutex is held while we're called, so
> +        * smbdirect_socket_destroy() of the listener can't reach nsc on the
> +        * pending list before we're done.
> +        *
> +        * From here nsc is published on the pending list: on failure of
> +        * smbdirect_accept_connect_request() below we schedule nsc's teardown,
> +        * which orphans nsc off the listener so
> +        * smbdirect_listen_purge_orphaned_work() -> smbdirect_socket_release()
> +        * releases it.
> +        */
>         spin_lock_irqsave(&lsc->listen.lock, flags);
>         list_add_tail(&nsc->accept.list, &lsc->listen.pending);
> -       nsc->accept.listener = lsc;
> +       WRITE_ONCE(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;
> +       if (ret) {
> +               /*
> +                * The rdma_cm core holds both the listener's and nsc's
> +                * id_priv->handler_mutex across this CONNECT_REQUEST handler,
> +                * so neither can go away under us here and nsc cannot receive
> +                * any rdma event: it is safe to hand nsc to its teardown.
> +                *
> +                * That teardown must be deferred though: nsc->rdma.cm_id
> +                * (= new_id) has its handler_mutex held by us, so it cannot be
> +                * destroyed synchronously from here.
> +                * smbdirect_socket_schedule_cleanup() only queues work;
> +                * smbdirect_socket_cleanup_work() then orphans nsc off the
> +                * listener and smbdirect_listen_purge_orphaned_work() releases
> +                * it, destroying nsc's cm_id with rdma_destroy_id().
> +                */
> +               smbdirect_socket_schedule_cleanup(nsc, ret);
> +       }
>
> +       /*
> +        * Always return 0 so the rdma_cm core keeps new_id: nsc owns it now.
> +        * On success it stays connected; on failure its deferred teardown
> +        * above destroys it.
> +        */
>         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:
>         /*
> diff --git a/fs/smb/smbdirect/socket.c b/fs/smb/smbdirect/socket.c
> index c36cb7cc0088b..e270e36b3145e 100644
> --- a/fs/smb/smbdirect/socket.c
> +++ b/fs/smb/smbdirect/socket.c
> @@ -319,6 +319,15 @@ void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
>          * (smbdirect_socket_destroy) to reap.
>          */
>         if (sc->listen.backlog != -1) { /* was a listener */
> +               /*
> +                * We don't move them to the orphaned list here,
> +                * each child does that itself at the end of its
> +                * smbdirect_socket_cleanup_work(), see
> +                * smbdirect_listen_orphan_socket(). Only that checks
> +                * accept.listener, which is still NULL for a child
> +                * that smbdirect_listen_connect_request() is still
> +                * setting up and will release itself on failure.
> +                */
>                 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)
> @@ -427,6 +436,15 @@ static void smbdirect_socket_cleanup_work(struct work_struct *work)
>          * instances of one class -- harmless, but lockdep cannot tell).
>          */
>         if (sc->listen.backlog != -1) { /* was a listener */
> +               /*
> +                * We don't move them to the orphaned list here,
> +                * each child does that itself at the end of its
> +                * smbdirect_socket_cleanup_work(), see
> +                * smbdirect_listen_orphan_socket(). Only that checks
> +                * accept.listener, which is still NULL for a child
> +                * that smbdirect_listen_connect_request() is still
> +                * setting up and will release itself on failure.
> +                */
>                 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)
> @@ -486,6 +504,18 @@ static void smbdirect_socket_cleanup_work(struct work_struct *work)
>          * in order to notice the broken connection.
>          */
>         smbdirect_socket_wake_up_all(sc);
> +
> +       /*
> +        * If we're still on the pending or ready list
> +        * of a listener, we started the disconnect
> +        * as far as possible above, so we move ourself
> +        * to the orphaned list of the listener,
> +        * which will release us.
> +        *
> +        * This is a no-op internally if
> +        * sc->accept.listener is NULL.
> +        */
> +       smbdirect_listen_orphan_socket(sc);
>  }
>
>  static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
> @@ -544,6 +574,7 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
>         disable_work_sync(&sc->recv_io.posted.refill_work);
>         disable_work_sync(&sc->idle.immediate_work);
>         disable_delayed_work_sync(&sc->idle.timer_work);
> +       disable_work_sync(&sc->listen.purge_orphaned_work);
>
>         if (sc->rdma.cm_id)
>                 rdma_lock_handler(sc->rdma.cm_id);
> @@ -607,6 +638,19 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
>         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);
> +       /*
> +        * purge_orphaned_work is already disabled above,
> +        * so we also need to release the orphaned sockets.
> +        *
> +        * Clearing accept.listener under listen.lock
> +        * makes us responsible for releasing them and
> +        * prevents them from moving themselves to
> +        * the orphaned list via
> +        * smbdirect_listen_orphan_socket().
> +        */
> +       list_splice_tail_init(&sc->listen.orphaned, &pending_list);
> +       list_for_each_entry(psc, &pending_list, accept.list)
> +               WRITE_ONCE(psc->accept.listener, NULL);
>         spin_unlock_irqrestore(&sc->listen.lock, flags);
>
>         /* It's not possible for upper layer to get to reassembly */
> @@ -650,7 +694,6 @@ static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
>                         "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 */
> @@ -777,7 +820,17 @@ static void smbdirect_socket_release_destroy(struct kref *kref)
>          * in DESTROYED state, before we free the memory.
>          */
>         smbdirect_socket_destroy_sync(sc);
> -       kfree(sc);
> +
> +       /*
> +        * Only a listener (backlog != -1) is ever dereferenced
> +        * via sc->accept.listener under rcu_read_lock(), see
> +        * smbdirect_listen_orphan_socket(). Other sockets can be
> +        * freed immediately.
> +        */
> +       if (sc->listen.backlog != -1) /* was a listener */
> +               kfree_rcu(sc, refs.rcu);
> +       else
> +               kfree(sc);
>  }
>
>  void smbdirect_socket_release(struct smbdirect_socket *sc)
> diff --git a/fs/smb/smbdirect/socket.h b/fs/smb/smbdirect/socket.h
> index c09eddd8ad16d..14114a2567a2e 100644
> --- a/fs/smb/smbdirect/socket.h
> +++ b/fs/smb/smbdirect/socket.h
> @@ -151,6 +151,12 @@ struct smbdirect_socket {
>                  * the disconnect refcount.
>                  */
>                 struct kref destroy;
> +
> +               /*
> +                * smbdirect_socket_release_destroy() uses
> +                * kfree_rcu(), see accept.listener.
> +                */
> +               struct rcu_head rcu;
>         } refs;
>
>         /* RDMA related */
> @@ -216,6 +222,16 @@ struct smbdirect_socket {
>                  * only be > 0.
>                  */
>                 int backlog;
> +               /*
> +                * Sockets on pending or ready that failed
> +                * move themselves to orphaned and queue
> +                * purge_orphaned_work, which releases them.
> +                * So that they are freed while the listener
> +                * is still alive. They still count against
> +                * the backlog until they are released.
> +                */
> +               struct list_head orphaned;
> +               struct work_struct purge_orphaned_work;
>         } listen;
>
>         /*
> @@ -226,7 +242,30 @@ struct smbdirect_socket {
>          * connection.
>          */
>         struct {
> +               /*
> +                * This is only set, protected by
> +                * listener->listen.lock, while the socket
> +                * is owned by the listener (on its pending,
> +                * ready or orphaned list). The one who clears
> +                * it is responsible for releasing the socket.
> +                *
> +                * The memory of a struct smbdirect_socket is
> +                * freed via kfree_rcu(), so the listener
> +                * can be dereferenced under rcu_read_lock(),
> +                * even if accept.listener is cleared and
> +                * the listener is released concurrently.
> +                */
>                 struct smbdirect_socket *listener;
> +               /*
> +                * Linkage on one of the listener's
> +                * listen.{pending,ready,orphaned} lists.
> +                *
> +                * accept.list and accept.listener are set
> +                * together under listener->listen.lock before
> +                * smbdirect_accept_connect_request() is called,
> +                * so whenever the socket is on one of those lists
> +                * accept.listener is that listener (never NULL).
> +                */
>                 struct list_head list;
>         } accept;
>
> @@ -588,6 +627,9 @@ static __always_inline void smbdirect_socket_init(struct smbdirect_socket *sc)
>         spin_lock_init(&sc->listen.lock);
>         INIT_LIST_HEAD(&sc->listen.pending);
>         INIT_LIST_HEAD(&sc->listen.ready);
> +       INIT_LIST_HEAD(&sc->listen.orphaned);
> +       INIT_WORK(&sc->listen.purge_orphaned_work, __smbdirect_socket_disabled_work);
> +       disable_work_sync(&sc->listen.purge_orphaned_work);
>         sc->listen.backlog = -1; /* not a listener */
>         init_waitqueue_head(&sc->listen.wait_queue);
>
> --
> 2.53.0
>
>
>

[PATCH 7.2 04/10] smb: smbdirect: dont hand out already failed sockets in smbdirect_socket_accept()

[Greg Kroah-Hartman]

7.2-stable review patch.  If anyone has any objections, please let me know.

------------------

From: Stefan Metzmacher <metze@samba.org>

[ Upstream commit 5c171c90c446bdce80ebaecfc5ed2c410bcc7427 ]

A socket on the ready list is in SMBDIRECT_SOCKET_NEGOTIATE_RUNNING,
but it may fail before smbdirect_socket_accept() picks it up, e.g.
RDMA_CM_EVENT_DISCONNECTED already moved it to
SMBDIRECT_SOCKET_DISCONNECTED.

smbdirect_socket_accept() unconditionally overwrote the status with
SMBDIRECT_SOCKET_CONNECTED and handed out the already disconnected
socket. smbdirect_socket_cleanup_work() then called rdma_disconnect()
on it again, and before the recent commit
smbdirect_socket_destroy_sync() waited forever for an
RDMA_CM_EVENT_DISCONNECTED that already happened.

Now we check first_error and change the status to
SMBDIRECT_SOCKET_CONNECTED only via cmpxchg() from
SMBDIRECT_SOCKET_NEGOTIATE_RUNNING, while we still hold listen.lock.
If the socket already failed, we move it to the orphaned list and
queue purge_orphaned_work, like smbdirect_listen_orphan_socket() does,
and try the next one. If we only found failed sockets, we wait for
the next one.

Fixes: dc691b91ad16 ("smb: smbdirect: introduce smbdirect_socket_{listen,accept}()")
Reported-by: Sashiko <sashiko-bot+sashiko@kernel.org>
Closes: https://sashiko.dev/#/patchset/cover.1791224972.git.metze%40samba.org
Closes: https://sashiko.dev/#/bug/linux-f9dbca09-30a7-4145-9bdb-4676759982c8
Cc: Namjae Jeon <linkinjeon@kernel.org>
Cc: Paulo Alcantara <pc@manguebit.org>
Cc: Tom Talpey <tom@talpey.com>
Cc: linux-cifs@vger.kernel.org
Cc: samba-technical@lists.samba.org
Assisted-by: Claude:claude-opus-5-5
Signed-off-by: Stefan Metzmacher <metze@samba.org>
Signed-off-by: Namjae Jeon <linkinjeon@kernel.org>
Signed-off-by: Sasha Levin <sashal@kernel.org>
---
 fs/smb/smbdirect/accept.c | 84 +++++++++++++++++++++++++++++++++------
 1 file changed, 72 insertions(+), 12 deletions(-)

diff --git a/fs/smb/smbdirect/accept.c b/fs/smb/smbdirect/accept.c
index 66db978bd8f73..d1bc08ac97575 100644
--- a/fs/smb/smbdirect/accept.c
+++ b/fs/smb/smbdirect/accept.c
@@ -835,7 +835,11 @@ static long smbdirect_socket_wait_for_accept(struct smbdirect_socket *lsc, long
 	if (ret < 0)
 		return ret;
 
-	return 0;
+	/*
+	 * Return the remaining timeout, so the caller can carry it
+	 * over to the next smbdirect_socket_wait_for_accept() call.
+	 */
+	return ret;
 }
 
 struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
@@ -845,6 +849,7 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
 	struct smbdirect_socket *nsc;
 	unsigned long flags;
 
+again:
 	if (lsc->status != SMBDIRECT_SOCKET_LISTENING) {
 		arg->err = -EINVAL;
 		return NULL;
@@ -856,7 +861,7 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
 	}
 
 	if (list_empty_careful(&lsc->listen.ready)) {
-		int ret;
+		long ret;
 
 		if (timeo == 0) {
 			arg->err = -EAGAIN;
@@ -864,16 +869,64 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
 		}
 
 		ret = smbdirect_socket_wait_for_accept(lsc, timeo);
-		if (ret) {
+		if (ret < 0) {
 			arg->err = ret;
 			return NULL;
 		}
+		/*
+		 * Carry the remaining timeout over, so that a stream of
+		 * failed connections that we orphan and skip (goto again)
+		 * can't reset the caller's timeout and wait forever.
+		 */
+		timeo = ret;
 	}
 
 	spin_lock_irqsave(&lsc->listen.lock, flags);
-	nsc = list_first_entry_or_null(&lsc->listen.ready,
-				       struct smbdirect_socket,
-				       accept.list);
+	/*
+	 * Recheck under the lock.
+	 */
+	if (lsc->status != SMBDIRECT_SOCKET_LISTENING) {
+		arg->err = -EINVAL;
+		spin_unlock_irqrestore(&lsc->listen.lock, flags);
+		return NULL;
+	}
+	if (lsc->first_error) {
+		arg->err = lsc->first_error;
+		spin_unlock_irqrestore(&lsc->listen.lock, flags);
+		return NULL;
+	}
+
+	while ((nsc = list_first_entry_or_null(&lsc->listen.ready,
+					       struct smbdirect_socket,
+					       accept.list))) {
+		/*
+		 * nsc may have failed after it was moved
+		 * to the ready list, e.g. RDMA_CM_EVENT_DISCONNECTED
+		 * already moved it to SMBDIRECT_SOCKET_DISCONNECTED.
+		 * We must not overwrite that with
+		 * SMBDIRECT_SOCKET_CONNECTED and hand
+		 * out an already disconnected socket.
+		 *
+		 * Doing this under listen.lock means nsc
+		 * still belongs to us, so we can just move
+		 * a failed socket to the orphaned list,
+		 * like smbdirect_listen_orphan_socket() does,
+		 * and try the next one.
+		 */
+		if (!READ_ONCE(nsc->first_error) &&
+		    cmpxchg(&nsc->status,
+			    SMBDIRECT_SOCKET_NEGOTIATE_RUNNING,
+			    SMBDIRECT_SOCKET_CONNECTED) ==
+		    SMBDIRECT_SOCKET_NEGOTIATE_RUNNING)
+			break;
+
+		smbdirect_log_rdma_event(nsc, SMBDIRECT_LOG_INFO,
+			"orphaning failed socket status=%s first_error=%1pe\n",
+			smbdirect_socket_status_string(nsc->status),
+			SMBDIRECT_DEBUG_ERR_PTR(nsc->first_error));
+		list_move_tail(&nsc->accept.list, &lsc->listen.orphaned);
+		queue_work(lsc->workqueues.cleanup, &lsc->listen.purge_orphaned_work);
+	}
 	if (nsc) {
 		WRITE_ONCE(nsc->accept.listener, NULL);
 		list_del_init_careful(&nsc->accept.list);
@@ -881,8 +934,13 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
 	}
 	spin_unlock_irqrestore(&lsc->listen.lock, flags);
 	if (!nsc) {
-		arg->err = -EAGAIN;
-		return NULL;
+		/*
+		 * If we only found failed sockets or no socket,
+		 * we wait for the next one.
+		 *
+		 * A possible -EAGAIN/-ETIMEOUT is handled above.
+		 */
+		goto again;
 	}
 
 	/*
@@ -891,12 +949,14 @@ struct smbdirect_socket *smbdirect_socket_accept(struct smbdirect_socket *lsc,
 	 * so it didn't grant any credits to us.
 	 *
 	 * The caller expects a connected socket
-	 * now as there are no credits anyway.
+	 * now as there are no credits anyway,
+	 * above we already changed to SMBDIRECT_SOCKET_CONNECTED
+	 * under the lsc->listen.lock and with cmpxchg.
 	 *
-	 * Then we send the negotiation response in
-	 * order to grant credits to the peer.
+	 * Now we send the negotiation response in
+	 * order to grant credits to the peer,
+	 * as the socket is now visible to the application layer.
 	 */
-	nsc->status = SMBDIRECT_SOCKET_CONNECTED;
 	smbdirect_accept_negotiate_finish(nsc, 0);
 
 	return nsc;
-- 
2.53.0