AI Trajectory


Trajectory: (Expand/Collapse All)
Seq Timestamp Type Name Duration
0/0 2026/09/14 04:28 flow patch-triage
1m Args:
null

Results:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND",
    "INFINIBAND_ADDR_TRANS"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_destroy_fini",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_conn_create_outgoing",
    "rds_sendmsg",
    "rds_send_queue_rm",
    "rds_release",
    "rds_tcp_accept_one"
  ],
  "KMSANReasoning": "The patch introduces reference counting (kref) and synchronization fixes for RDS connections to prevent use-after-free, races, and premature frees during connection teardown and module unloading. All newly added or modified data structures (e.g. struct rds_connection, rds_conn_path) continue to be allocated using zero-initializing allocators (kmem_cache_zalloc, kzalloc_objs), and all local variables are properly initialized. The only user-copy path touched is SIOCRDSGETTOS, which merely updates the spinlock protecting rs_tos (a single __u8 value). The risks associated with these changes are strictly related to object lifetimes, concurrent access, reference leaks, or locking deadlocks, which are effectively covered by KASAN and LOCKDEP. No uninitialized memory reads or information leak hazards are introduced or exposed, so a dedicated KMSAN fuzzing session is not warranted.",
  "NeedsKMSAN": false,
  "Reasoning": "The patch substantially refactors connection lifetime and reference counting (using kref) in the RDS subsystem (net/rds), deferring connection cleanup until all references from messages, incoming queues, sockets, and transports are dropped. These changes touch core networking logic reachable via AF_RDS sockets and could introduce subtle race conditions, refcounting leaks, use-after-free, or deadlocks.",
  "WorthFuzzing": true
}

1/1 2026/09/14 04:28 action read-patch-diff
0m Args:
null

Results:
{
  "PatchDiff": "commit fe0d65dad5bf3cdb3ebccbae9d834aa98bc8f6e0\nAuthor: syz-cluster \u003ctriage@syzkaller.com\u003e\nDate:   Mon Sep 14 04:28:33 2026 +0000\n\n    syz-cluster: applied patch under review\n\ndiff --git a/net/rds/af_rds.c b/net/rds/af_rds.c\nindex d5defe9172e36..1cc20b5cfd213 100644\n--- a/net/rds/af_rds.c\n+++ b/net/rds/af_rds.c\n@@ -80,6 +80,14 @@ static int rds_release(struct socket *sock)\n \trds_notify_queue_get(rs, NULL);\n \trds_notify_msg_zcopy_purge(\u0026rs-\u003ers_zcookie_queue);\n \n+\t/* drop the cached connection reference; no sendmsg can race\n+\t * with us here, the socket is going away\n+\t */\n+\tif (rs-\u003ers_conn) {\n+\t\trds_conn_put(rs-\u003ers_conn);\n+\t\trs-\u003ers_conn = NULL;\n+\t}\n+\n \tspin_lock_bh(\u0026rds_sock_lock);\n \tlist_del_init(\u0026rs-\u003ers_item);\n \tspin_unlock_bh(\u0026rds_sock_lock);\n@@ -255,6 +263,7 @@ static int rds_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)\n {\n \tstruct rds_sock *rs = rds_sk_to_rs(sock-\u003esk);\n \trds_tos_t utos, tos = 0;\n+\tunsigned long flags;\n \n \tswitch (cmd) {\n \tcase SIOCRDSSETTOS:\n@@ -267,18 +276,23 @@ static int rds_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)\n \t\telse\n \t\t\treturn -ENOIOCTLCMD;\n \n-\t\tspin_lock_bh(\u0026rds_sock_lock);\n+\t\t/* rs_conn is serialized by rs_lock (see rds_sendmsg());\n+\t\t * hold it across the \"no connection yet\" check and the\n+\t\t * rs_tos store so a racing sendmsg cannot cache a conn\n+\t\t * whose c_tos then disagrees with rs_tos.\n+\t\t */\n+\t\tspin_lock_irqsave(\u0026rs-\u003ers_lock, flags);\n \t\tif (rs-\u003ers_tos || rs-\u003ers_conn) {\n-\t\t\tspin_unlock_bh(\u0026rds_sock_lock);\n+\t\t\tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n \t\t\treturn -EINVAL;\n \t\t}\n \t\trs-\u003ers_tos = tos;\n-\t\tspin_unlock_bh(\u0026rds_sock_lock);\n+\t\tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n \t\tbreak;\n \tcase SIOCRDSGETTOS:\n-\t\tspin_lock_bh(\u0026rds_sock_lock);\n+\t\tspin_lock_irqsave(\u0026rs-\u003ers_lock, flags);\n \t\ttos = rs-\u003ers_tos;\n-\t\tspin_unlock_bh(\u0026rds_sock_lock);\n+\t\tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n \t\tif (put_user(tos, (rds_tos_t __user *)arg))\n \t\t\treturn -EFAULT;\n \t\tbreak;\ndiff --git a/net/rds/connection.c b/net/rds/connection.c\nindex b6c4beb50eaf0..f5f25e4a53b97 100644\n--- a/net/rds/connection.c\n+++ b/net/rds/connection.c\n@@ -47,7 +47,8 @@\n \n /* converting this to RCU is a chore for another day.. */\n static DEFINE_SPINLOCK(rds_conn_lock);\n-static unsigned long rds_conn_count;\n+/* woken whenever a transport's t_conn_count drops to zero */\n+static DECLARE_WAIT_QUEUE_HEAD(rds_conn_freed_waitq);\n static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];\n static struct kmem_cache *rds_conn_slab;\n \n@@ -79,7 +80,18 @@ static struct hlist_head *rds_conn_bucket(const struct in6_addr *laddr,\n \t\tvar |= RDS_INFO_CONNECTION_FLAG_##suffix;\t\\\n } while (0)\n \n-/* rcu read lock must be held or the connection spinlock */\n+/* rcu read lock must be held or the connection spinlock.\n+ * On success a reference is taken on the returned connection; the\n+ * caller must drop it with rds_conn_put().\n+ */\n+/* c_passive is written under rds_conn_lock and read under RCU */\n+static struct rds_connection *\n+rds_conn_passive_locked(struct rds_connection *conn)\n+{\n+\treturn rcu_dereference_protected(conn-\u003ec_passive,\n+\t\t\t\t\t lockdep_is_held(\u0026rds_conn_lock));\n+}\n+\n static struct rds_connection *rds_conn_lookup(struct net *net,\n \t\t\t\t\t      struct hlist_head *head,\n \t\t\t\t\t      const struct in6_addr *laddr,\n@@ -96,6 +108,17 @@ static struct rds_connection *rds_conn_lookup(struct net *net,\n \t\t    conn-\u003ec_tos == tos \u0026\u0026\n \t\t    net == rds_conn_net(conn) \u0026\u0026\n \t\t    conn-\u003ec_dev_if == dev_if) {\n+\t\t\t/* Only ever hand out a live reference.\n+\t\t\t * rds_conn_destroy() unhashes under\n+\t\t\t * rds_conn_lock and waits a grace period\n+\t\t\t * before dropping the initial reference, so\n+\t\t\t * an entry this traversal reaches still holds\n+\t\t\t * at least that one; the conditional get\n+\t\t\t * documents the contract rather than\n+\t\t\t * papering over a zero-refcount entry.\n+\t\t\t */\n+\t\t\tif (!kref_get_unless_zero(\u0026conn-\u003ec_refcount))\n+\t\t\t\tcontinue;\n \t\t\tret = conn;\n \t\t\tbreak;\n \t\t}\n@@ -161,6 +184,22 @@ static void __rds_conn_path_init(struct rds_connection *conn,\n \tcp-\u003ecp_flags = 0;\n }\n \n+/* Undo trans-\u003econn_alloc(): it may have allocated transport data for\n+ * every path of a multipath connection, not just for path 0.\n+ */\n+static void rds_conn_free_transport_data(struct rds_connection *conn,\n+\t\t\t\t\t int npaths)\n+{\n+\tstruct rds_conn_path *cp;\n+\tint i;\n+\n+\tfor (i = 0; i \u003c npaths; i++) {\n+\t\tcp = \u0026conn-\u003ec_path[i];\n+\t\tif (cp-\u003ecp_transport_data)\n+\t\t\tconn-\u003ec_trans-\u003econn_free(cp-\u003ecp_transport_data);\n+\t}\n+}\n+\n /*\n  * There is only every one 'conn' for a given pair of addresses in the\n  * system at a time.  They contain messages to be retransmitted and so\n@@ -197,7 +236,20 @@ static struct rds_connection *__rds_conn_create(struct net *net,\n \t\t * We need a second connection object into which we\n \t\t * can stick the other QP. */\n \t\tparent = conn;\n-\t\tconn = parent-\u003ec_passive;\n+\t\t/* The c_passive pointer holds a reference which is only\n+\t\t * dropped one synchronize_rcu() after the pointer is\n+\t\t * cleared, so within this RCU section a fetched pointer\n+\t\t * is always safe to take a reference on.  A passive conn\n+\t\t * whose own destroy has begun is not handed out, though:\n+\t\t * it is quiesced and about to clear the parent's pointer\n+\t\t * itself, and reusing it would re-arm a connection that\n+\t\t * nothing will tear down again.\n+\t\t */\n+\t\tconn = rcu_dereference(parent-\u003ec_passive);\n+\t\tif (conn \u0026\u0026 READ_ONCE(conn-\u003ec_destroy_in_prog))\n+\t\t\tconn = NULL;\n+\t\tif (conn)\n+\t\t\trds_conn_get(conn);\n \t}\n \trcu_read_unlock();\n \tif (conn)\n@@ -215,6 +267,7 @@ static struct rds_connection *__rds_conn_create(struct net *net,\n \t\tgoto out;\n \t}\n \n+\tkref_init(\u0026conn-\u003ec_refcount);\n \tINIT_HLIST_NODE(\u0026conn-\u003ec_hash_node);\n \tconn-\u003ec_laddr = *laddr;\n \tconn-\u003ec_isv6 = !ipv6_addr_v4mapped(laddr);\n@@ -278,12 +331,13 @@ static struct rds_connection *__rds_conn_create(struct net *net,\n \n \tinit_waitqueue_head(\u0026conn-\u003ec_hs_waitq);\n \tfor (i = 0; i \u003c npaths; i++) {\n+\t\tint seq = atomic_read(\u0026trans-\u003et_conn_count);\n+\n \t\t__rds_conn_path_init(conn, \u0026conn-\u003ec_path[i],\n \t\t\t\t     is_outgoing);\n \t\tconn-\u003ec_path[i].cp_index = i;\n \t\tconn-\u003ec_path[i].cp_wq =\n-\t\t\talloc_ordered_workqueue(\"krds_cp_wq#%lu/%d\", 0,\n-\t\t\t\t\t\trds_conn_count, i);\n+\t\t\talloc_ordered_workqueue(\"krds_cp_wq#%d/%d\", 0, seq, i);\n \t\tif (!conn-\u003ec_path[i].cp_wq)\n \t\t\tconn-\u003ec_path[i].cp_wq = rds_wq;\n \t}\n@@ -315,15 +369,46 @@ static struct rds_connection *__rds_conn_create(struct net *net,\n \tspin_lock_irqsave(\u0026rds_conn_lock, flags);\n \tif (parent) {\n \t\t/* Creating passive conn */\n-\t\tif (parent-\u003ec_passive) {\n-\t\t\ttrans-\u003econn_free(conn-\u003ec_path[0].cp_transport_data);\n+\t\tif (READ_ONCE(parent-\u003ec_destroy_in_prog)) {\n+\t\t\t/* The parent's destroy has begun (it sets the\n+\t\t\t * flag and snatches c_passive under this\n+\t\t\t * lock); do not install a new passive conn\n+\t\t\t * that nothing would ever destroy.\n+\t\t\t */\n+\t\t\trds_conn_free_transport_data(conn, npaths);\n+\t\t\tfree_cp = conn-\u003ec_path;\n+\t\t\tkmem_cache_free(rds_conn_slab, conn);\n+\t\t\tconn = ERR_PTR(-ENETDOWN);\n+\t\t} else if (rcu_access_pointer(parent-\u003ec_passive)) {\n+\t\t\tstruct rds_connection *passive;\n+\n+\t\t\tpassive = rds_conn_passive_locked(parent);\n+\t\t\trds_conn_free_transport_data(conn, npaths);\n \t\t\tfree_cp = conn-\u003ec_path;\n \t\t\tkmem_cache_free(rds_conn_slab, conn);\n-\t\t\tconn = parent-\u003ec_passive;\n+\t\t\tif (READ_ONCE(passive-\u003ec_destroy_in_prog)) {\n+\t\t\t\t/* Its destroy will clear the parent's\n+\t\t\t\t * pointer under this lock shortly; until\n+\t\t\t\t * then there is no usable passive conn.\n+\t\t\t\t */\n+\t\t\t\tconn = ERR_PTR(-ENETDOWN);\n+\t\t\t} else {\n+\t\t\t\trds_conn_get(passive);\n+\t\t\t\tconn = passive;\n+\t\t\t}\n \t\t} else {\n-\t\t\tparent-\u003ec_passive = conn;\n+\t\t\t/* The initial reference belongs to whoever\n+\t\t\t * destroys the conn (the transport's conn\n+\t\t\t * lists, as for any other conn).  Take one\n+\t\t\t * for the c_passive pointer - dropped when\n+\t\t\t * the parent is destroyed - and one for our\n+\t\t\t * caller.\n+\t\t\t */\n+\t\t\trds_conn_get(conn);\t/* c_passive */\n+\t\t\trds_conn_get(conn);\t/* caller */\n+\t\t\trcu_assign_pointer(parent-\u003ec_passive, conn);\n \t\t\trds_cong_add_conn(conn);\n-\t\t\trds_conn_count++;\n+\t\t\tatomic_inc(\u0026conn-\u003ec_trans-\u003et_conn_count);\n \t\t}\n \t} else {\n \t\t/* Creating normal conn */\n@@ -332,33 +417,28 @@ static struct rds_connection *__rds_conn_create(struct net *net,\n \t\tfound = rds_conn_lookup(net, head, laddr, faddr, trans,\n \t\t\t\t\ttos, dev_if);\n \t\tif (found) {\n-\t\t\tstruct rds_conn_path *cp;\n-\t\t\tint i;\n-\n-\t\t\tfor (i = 0; i \u003c npaths; i++) {\n-\t\t\t\tcp = \u0026conn-\u003ec_path[i];\n-\t\t\t\t/* The -\u003econn_alloc invocation may have\n-\t\t\t\t * allocated resource for all paths, so all\n-\t\t\t\t * of them may have to be freed here.\n-\t\t\t\t */\n-\t\t\t\tif (cp-\u003ecp_transport_data)\n-\t\t\t\t\ttrans-\u003econn_free(cp-\u003ecp_transport_data);\n-\t\t\t}\n+\t\t\trds_conn_free_transport_data(conn, npaths);\n \t\t\tfree_cp = conn-\u003ec_path;\n \t\t\tkmem_cache_free(rds_conn_slab, conn);\n \t\t\tconn = found;\n \t\t} else {\n \t\t\tconn-\u003ec_my_gen_num = rds_gen_num;\n \t\t\tconn-\u003ec_peer_gen_num = 0;\n+\t\t\t/* the initial reference belongs to whoever\n+\t\t\t * destroys the conn; take one for our caller\n+\t\t\t */\n+\t\t\trds_conn_get(conn);\n \t\t\thlist_add_head_rcu(\u0026conn-\u003ec_hash_node, head);\n \t\t\trds_cong_add_conn(conn);\n-\t\t\trds_conn_count++;\n+\t\t\tatomic_inc(\u0026conn-\u003ec_trans-\u003et_conn_count);\n \t\t}\n \t}\n \tspin_unlock_irqrestore(\u0026rds_conn_lock, flags);\n \trcu_read_unlock();\n \n out:\n+\tif (parent)\n+\t\trds_conn_put(parent);\n \tif (free_cp) {\n \t\tfor (i = 0; i \u003c npaths; i++)\n \t\t\tif (free_cp[i].cp_wq != rds_wq)\n@@ -466,9 +546,10 @@ void rds_conn_shutdown(struct rds_conn_path *cp)\n \t\t\t * Quiesce the reconnect timer before bailing\n \t\t\t * out, though.  When a pending destroy did\n \t\t\t * suppress the queue, no later pass runs, and\n-\t\t\t * rds_conn_path_destroy() is about to flush\n-\t\t\t * cp_down_w and free the path: it must not\n-\t\t\t * find cp_conn_w still armed.  A successor\n+\t\t\t * rds_conn_path_quiesce() is about to flush\n+\t\t\t * cp_down_w, ahead of the path's deferred\n+\t\t\t * free: it must not find cp_conn_w still\n+\t\t\t * armed.  A successor\n \t\t\t * pass, when there is one, re-arms the\n \t\t\t * reconnect from its own tail.\n \t\t\t */\n@@ -515,12 +596,16 @@ void rds_conn_shutdown(struct rds_conn_path *cp)\n \t\tconn-\u003ec_trans-\u003econn_slots_available(conn, false);\n }\n \n-/* destroy a single rds_conn_path. rds_conn_destroy() iterates over\n- * all paths using rds_conn_path_destroy()\n+/* quiesce a single rds_conn_path: shut it down and tear down any\n+ * queued messages.  rds_conn_destroy() iterates over all paths using\n+ * rds_conn_path_quiesce(); the transport state and the workqueue are\n+ * freed later, from rds_conn_path_free().\n  */\n-static void rds_conn_path_destroy(struct rds_conn_path *cp)\n+static void rds_conn_path_quiesce(struct rds_conn_path *cp)\n {\n \tstruct rds_message *rm, *rtmp;\n+\tunsigned long flags;\n+\tLIST_HEAD(purge);\n \n \tif (!cp-\u003ecp_transport_data)\n \t\treturn;\n@@ -532,10 +617,16 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)\n \trds_conn_path_drop(cp, true);\n \tflush_work(\u0026cp-\u003ecp_down_w);\n \n-\t/* tear down queued messages */\n-\tlist_for_each_entry_safe(rm, rtmp,\n-\t\t\t\t \u0026cp-\u003ecp_send_queue,\n-\t\t\t\t m_conn_item) {\n+\t/* Tear down queued messages.  Take the queue under cp_lock:\n+\t * a sender that still holds a reference can be inside\n+\t * rds_send_queue_rm() right now, and it tests\n+\t * rds_destroy_pending() under the same lock, so after this\n+\t * splice nothing is added behind our back.\n+\t */\n+\tspin_lock_irqsave(\u0026cp-\u003ecp_lock, flags);\n+\tlist_splice_init(\u0026cp-\u003ecp_send_queue, \u0026purge);\n+\tspin_unlock_irqrestore(\u0026cp-\u003ecp_lock, flags);\n+\tlist_for_each_entry_safe(rm, rtmp, \u0026purge, m_conn_item) {\n \t\tlist_del_init(\u0026rm-\u003em_conn_item);\n \t\tBUG_ON(!list_empty(\u0026rm-\u003em_sock_item));\n \t\trds_message_put(rm);\n@@ -547,6 +638,16 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)\n \tWARN_ON(delayed_work_pending(\u0026cp-\u003ecp_recv_w));\n \tWARN_ON(delayed_work_pending(\u0026cp-\u003ecp_conn_w));\n \tWARN_ON(work_pending(\u0026cp-\u003ecp_down_w));\n+}\n+\n+/* free a quiesced rds_conn_path's transport state and workqueue; runs\n+ * from rds_conn_destroy_fini() once the last connection reference is\n+ * dropped.\n+ */\n+static void rds_conn_path_free(struct rds_conn_path *cp)\n+{\n+\tif (!cp-\u003ecp_transport_data)\n+\t\treturn;\n \n \tif (cp-\u003ecp_wq != rds_wq) {\n \t\tdestroy_workqueue(cp-\u003ecp_wq);\n@@ -556,17 +657,96 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)\n \tcp-\u003ecp_conn-\u003ec_trans-\u003econn_free(cp-\u003ecp_transport_data);\n }\n \n+/* Free a connection.  This runs from rds_conn_put() when the last\n+ * reference is dropped, after rds_conn_destroy() has quiesced the\n+ * connection and dropped the initial reference.\n+ */\n+static void rds_conn_destroy_fini(struct kref *kref)\n+{\n+\tstruct rds_connection *conn = container_of(kref, struct rds_connection,\n+\t\t\t\t\t\t   c_refcount);\n+\tint npaths = (conn-\u003ec_trans-\u003et_mp_capable ? RDS_MPATH_WORKERS : 1);\n+\tstruct rds_transport *trans = conn-\u003ec_trans;\n+\tint i;\n+\n+\tfor (i = 0; i \u003c npaths; i++)\n+\t\trds_conn_path_free(\u0026conn-\u003ec_path[i]);\n+\n+\tkfree(conn-\u003ec_path);\n+\tkmem_cache_free(rds_conn_slab, conn);\n+\n+\t/* only after everything the transport module owns has been\n+\t * freed above may its unload proceed\n+\t */\n+\tif (!atomic_dec_return(\u0026trans-\u003et_conn_count))\n+\t\twake_up_all(\u0026rds_conn_freed_waitq);\n+}\n+\n+/* Wait for all of @trans's connections to be freed; the free runs\n+ * asynchronously once rds_conn_destroy() has quiesced a connection.\n+ * Called on transport module unload, after the transport has destroyed\n+ * all of its connections.  A connection reference can be held for an\n+ * application-controlled time - an unread datagram pins the inc that\n+ * carries it, and thus the connection - so the wait is unbounded: the\n+ * frees that run after unload call into this module's text (conn_free,\n+ * inc_free) and free into its slabs, so proceeding while any remain\n+ * would be a use-after-free, not a leak.  Warn periodically so a stuck\n+ * count is diagnosable, but never stop waiting.  This matches the\n+ * historical RDS contract that teardown does not discard queued data.\n+ */\n+void rds_conn_wait_conns_freed(struct rds_transport *trans,\n+\t\t\t       void (*resweep)(void))\n+{\n+\tunsigned long warn_interval =\n+\t\t\tmsecs_to_jiffies(RDS_CONN_FREE_WARN_INTERVAL_MS);\n+\tunsigned long warn_at = jiffies + warn_interval;\n+\n+\twhile (!wait_event_timeout(rds_conn_freed_waitq,\n+\t\t\t\t   !atomic_read(\u0026trans-\u003et_conn_count),\n+\t\t\t\t   msecs_to_jiffies(RDS_CONN_FREE_POLL_MS))) {\n+\t\t/* A transport whose teardown is asynchronous (IB moves a\n+\t\t * connection off its device from the shutdown work) gives\n+\t\t * us a resweep to destroy what has arrived since.\n+\t\t */\n+\t\tif (resweep)\n+\t\t\tresweep();\n+\t\tif (time_after_eq(jiffies, warn_at)) {\n+\t\t\tpr_warn(\"RDS/%s: still waiting for %d connection(s) to be freed before unload\\n\",\n+\t\t\t\ttrans-\u003et_name,\n+\t\t\t\tatomic_read(\u0026trans-\u003et_conn_count));\n+\t\t\twarn_at = jiffies + warn_interval;\n+\t\t}\n+\t}\n+}\n+EXPORT_SYMBOL_GPL(rds_conn_wait_conns_freed);\n+\n+void rds_conn_get(struct rds_connection *conn)\n+{\n+\tkref_get(\u0026conn-\u003ec_refcount);\n+}\n+EXPORT_SYMBOL_GPL(rds_conn_get);\n+\n+void rds_conn_put(struct rds_connection *conn)\n+{\n+\tkref_put(\u0026conn-\u003ec_refcount, rds_conn_destroy_fini);\n+}\n+EXPORT_SYMBOL_GPL(rds_conn_put);\n+\n /*\n  * Stop and free a connection.\n  *\n- * This can only be used in very limited circumstances.  It assumes that once\n- * the conn has been shutdown that no one else is referencing the connection.\n- * We can only ensure this in the rmmod path in the current code.\n+ * Quiesces the connection synchronously (workers cancelled, transport\n+ * connections shut down, queued messages dropped) and drops the\n+ * initial reference.  The memory - including the transport's\n+ * per-connection state and the path workqueues - is freed once the\n+ * last rds_conn_put() runs, which may be after this returns.\n  */\n void rds_conn_destroy(struct rds_connection *conn)\n {\n-\tunsigned long flags;\n \tint i;\n+\tstruct rds_connection *passive, *parent;\n+\tstruct hlist_head *head;\n+\tbool was_passive = false;\n \tstruct rds_conn_path *cp;\n \tint npaths = (conn-\u003ec_trans-\u003et_mp_capable ? RDS_MPATH_WORKERS : 1);\n \n@@ -574,16 +754,67 @@ void rds_conn_destroy(struct rds_connection *conn)\n \t\t \"%pI4\\n\", conn, \u0026conn-\u003ec_laddr,\n \t\t \u0026conn-\u003ec_faddr);\n \n-\t/* Ensure conn will not be scheduled for reconnect */\n+\t/* Make rds_destroy_pending() true for this conn.  Together with\n+\t * the synchronize_rcu() below this stops the work-requeueing\n+\t * sites (which all test rds_destroy_pending() under\n+\t * rcu_read_lock()) from queueing new work on the path\n+\t * workqueues once we start cancelling and destroying them.\n+\t *\n+\t * Now that the transport state stays discoverable (e.g. on the\n+\t * transports' connection lists) until the final rds_conn_put(),\n+\t * a conn can be handed to rds_conn_destroy() more than once -\n+\t * e.g. dropped for a protocol version mismatch and then found\n+\t * again at module unload.  Only the first caller proceeds; the\n+\t * unhash also happens under rds_conn_lock, so a looked-up conn\n+\t * can never be quiesced twice.\n+\t */\n \tspin_lock_irq(\u0026rds_conn_lock);\n+\tif (conn-\u003ec_destroy_in_prog) {\n+\t\tspin_unlock_irq(\u0026rds_conn_lock);\n+\t\treturn;\n+\t}\n+\tWRITE_ONCE(conn-\u003ec_destroy_in_prog, true);\n+\n+\t/* Ensure conn will not be scheduled for reconnect */\n \thlist_del_init_rcu(\u0026conn-\u003ec_hash_node);\n+\n+\t/* Snatch c_passive while holding the lock:\n+\t * __rds_conn_create() dereferences it under rcu_read_lock()\n+\t * (and refuses to install a new one once c_destroy_in_prog is\n+\t * set, which it checks under this lock).  After the\n+\t * synchronize_rcu() below no one can pick the pointer up any\n+\t * more and its reference can be dropped.\n+\t */\n+\tpassive = rds_conn_passive_locked(conn);\n+\tRCU_INIT_POINTER(conn-\u003ec_passive, NULL);\n+\n+\t/* If we are a parent's passive twin, invalidate its pointer to\n+\t * us as well, so that __rds_conn_create() cannot hand out a\n+\t * connection whose teardown has begun.  The parent is the\n+\t * hashed connection for our key (a passive conn is never\n+\t * hashed, and we unhashed ourselves above); it holds its\n+\t * initial reference for as long as it is hashed, so the lookup\n+\t * reference dropped below cannot be its last.\n+\t */\n+\thead = rds_conn_bucket(\u0026conn-\u003ec_laddr, \u0026conn-\u003ec_faddr);\n+\trcu_read_lock();\n+\tparent = rds_conn_lookup(rds_conn_net(conn), head, \u0026conn-\u003ec_laddr,\n+\t\t\t\t \u0026conn-\u003ec_faddr, conn-\u003ec_trans, conn-\u003ec_tos,\n+\t\t\t\t conn-\u003ec_dev_if);\n+\trcu_read_unlock();\n+\tif (parent \u0026\u0026 rds_conn_passive_locked(parent) == conn) {\n+\t\tRCU_INIT_POINTER(parent-\u003ec_passive, NULL);\n+\t\twas_passive = true;\n+\t}\n \tspin_unlock_irq(\u0026rds_conn_lock);\n+\tif (parent)\n+\t\trds_conn_put(parent);\n \tsynchronize_rcu();\n \n \t/* shut the connection down */\n \tfor (i = 0; i \u003c npaths; i++) {\n \t\tcp = \u0026conn-\u003ec_path[i];\n-\t\trds_conn_path_destroy(cp);\n+\t\trds_conn_path_quiesce(cp);\n \t\tBUG_ON(!list_empty(\u0026cp-\u003ecp_retrans));\n \t}\n \n@@ -594,12 +825,19 @@ void rds_conn_destroy(struct rds_connection *conn)\n \t */\n \trds_cong_remove_conn(conn);\n \n-\tkfree(conn-\u003ec_path);\n-\tkmem_cache_free(rds_conn_slab, conn);\n+\t/* drop the reference our c_passive pointer held, if any, and\n+\t * the one a parent's c_passive pointer held on us; neither can\n+\t * be the last, since the initial reference is dropped below\n+\t */\n+\tif (passive)\n+\t\trds_conn_put(passive);\n+\tif (was_passive)\n+\t\trds_conn_put(conn);\n \n-\tspin_lock_irqsave(\u0026rds_conn_lock, flags);\n-\trds_conn_count--;\n-\tspin_unlock_irqrestore(\u0026rds_conn_lock, flags);\n+\t/* drop the initial reference; the connection is freed from\n+\t * rds_conn_destroy_fini() once every holder has dropped theirs\n+\t */\n+\trds_conn_put(conn);\n }\n EXPORT_SYMBOL_GPL(rds_conn_destroy);\n \ndiff --git a/net/rds/ib.c b/net/rds/ib.c\nindex 786f39169bc14..3fc2de9d19d55 100644\n--- a/net/rds/ib.c\n+++ b/net/rds/ib.c\n@@ -525,10 +525,7 @@ static void rds_ib_set_unloading(void)\n \n static bool rds_ib_is_unloading(struct rds_connection *conn)\n {\n-\tstruct rds_conn_path *cp = \u0026conn-\u003ec_path[0];\n-\n-\treturn (test_bit(RDS_DESTROY_PENDING, \u0026cp-\u003ecp_flags) ||\n-\t\tatomic_read(\u0026rds_ib_unloading) != 0);\n+\treturn atomic_read(\u0026rds_ib_unloading) != 0;\n }\n \n void rds_ib_exit(void)\n@@ -540,7 +537,24 @@ void rds_ib_exit(void)\n \trds_info_deregister_func(RDS6_INFO_IB_CONNECTIONS, rds6_ib_ic_info);\n #endif\n \trds_ib_unregister_client();\n+\n+\t/* rds_ib_dev_shutdown() only dropped the connections still\n+\t * attached to a device; each moves itself to ib_nodev_conns\n+\t * from its shutdown work.  Destroy what is there now and keep\n+\t * sweeping the list while the wait sees connections outstanding,\n+\t * so a late arrival is destroyed rather than waited on forever.\n+\t */\n \trds_ib_destroy_nodev_conns();\n+\trds_conn_wait_conns_freed(\u0026rds_ib_transport,\n+\t\t\t\t  rds_ib_destroy_nodev_conns);\n+\n+\t/* Tearing down the last connection may have dropped the final\n+\t * reference on a device, deferring rds_ib_dev_free() to rds_wq.\n+\t * Drain it before the module goes away; it queues nothing\n+\t * further on rds_wq.\n+\t */\n+\tflush_workqueue(rds_wq);\n+\n \trds_ib_sysctl_exit();\n \trds_ib_recv_exit();\n \trds_trans_unregister(\u0026rds_ib_transport);\ndiff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c\nindex 4feb0edc360c8..257e03ed48618 100644\n--- a/net/rds/ib_cm.c\n+++ b/net/rds/ib_cm.c\n@@ -874,6 +874,13 @@ int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id,\n \t * see the comment above rds_queue_reconnect()\n \t */\n \tmutex_lock(\u0026conn-\u003ec_cm_lock);\n+\t/* A destroy that has already quiesced this conn leaves it in\n+\t * RDS_CONN_DOWN with no cm_id, exactly what the transition\n+\t * below would happily claim; nothing would tear the new cm_id\n+\t * and QP down again before the conn is freed.  Reject instead.\n+\t */\n+\tif (rds_destroy_pending(conn))\n+\t\tgoto out;\n \tif (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_CONNECTING)) {\n \t\tif (rds_conn_state(conn) == RDS_CONN_UP) {\n \t\t\trdsdebug(\"incoming connect while connecting\\n\");\n@@ -924,8 +931,14 @@ int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id,\n \t\trds_ib_conn_error(conn, \"rdma_accept failed\\n\");\n \n out:\n-\tif (conn)\n+\tif (conn) {\n \t\tmutex_unlock(\u0026conn-\u003ec_cm_lock);\n+\t\t/* The conn stays reachable through cm_id-\u003econtext\n+\t\t * without a reference of its own: connection destroy\n+\t\t * shuts the cm_id down before the conn is freed.\n+\t\t */\n+\t\trds_conn_put(conn);\n+\t}\n \tif (err)\n \t\trdma_reject(cm_id, \u0026err, sizeof(int),\n \t\t\t    IB_CM_REJ_CONSUMER_DEFINED);\n@@ -1271,6 +1284,7 @@ void rds_ib_conn_free(void *arg)\n {\n \tstruct rds_ib_connection *ic = arg;\n \tspinlock_t\t*lock_ptr;\n+\tunsigned long flags;\n \n \trdsdebug(\"ic %p\\n\", ic);\n \n@@ -1278,12 +1292,18 @@ void rds_ib_conn_free(void *arg)\n \t * Conn is either on a dev's list or on the nodev list.\n \t * A race with shutdown() or connect() would cause problems\n \t * (since rds_ibdev would change) but that should never happen.\n+\t *\n+\t * Callers may hold rds_conn_lock with interrupts disabled\n+\t * (__rds_conn_create() undoing a lost creation race), so do not\n+\t * re-enable interrupts unconditionally here.\n \t */\n \tlock_ptr = ic-\u003erds_ibdev ? \u0026ic-\u003erds_ibdev-\u003espinlock : \u0026ib_nodev_conns_lock;\n \n-\tspin_lock_irq(lock_ptr);\n-\tlist_del(\u0026ic-\u003eib_node);\n-\tspin_unlock_irq(lock_ptr);\n+\tspin_lock_irqsave(lock_ptr, flags);\n+\t/* already unlinked if a transport teardown gathered us first */\n+\tif (!list_empty(\u0026ic-\u003eib_node))\n+\t\tlist_del(\u0026ic-\u003eib_node);\n+\tspin_unlock_irqrestore(lock_ptr, flags);\n \n \trds_ib_recv_free_caches(ic);\n \ndiff --git a/net/rds/ib_rdma.c b/net/rds/ib_rdma.c\nindex db7e92e7bd29f..91db43a0e7d73 100644\n--- a/net/rds/ib_rdma.c\n+++ b/net/rds/ib_rdma.c\n@@ -165,11 +165,21 @@ void rds_ib_destroy_nodev_conns(void)\n \n \t/* avoid calling conn_destroy with irqs off */\n \tspin_lock_irq(\u0026ib_nodev_conns_lock);\n-\tlist_splice(\u0026ib_nodev_conns, \u0026tmp_list);\n+\tlist_splice_init(\u0026ib_nodev_conns, \u0026tmp_list);\n \tspin_unlock_irq(\u0026ib_nodev_conns_lock);\n \n-\tlist_for_each_entry_safe(ic, _ic, \u0026tmp_list, ib_node)\n+\t/* rds_conn_destroy() can return before the connection is freed,\n+\t * and it is the free - rds_ib_conn_free() - that unlinks ib_node.\n+\t * tmp_list lives on this stack frame, so unlink each node before\n+\t * its destroy; the free then finds it empty and leaves it alone.\n+\t */\n+\tlist_for_each_entry_safe(ic, _ic, \u0026tmp_list, ib_node) {\n+\t\tspin_lock_irq(\u0026ib_nodev_conns_lock);\n+\t\tlist_del_init(\u0026ic-\u003eib_node);\n+\t\tspin_unlock_irq(\u0026ib_nodev_conns_lock);\n+\n \t\trds_conn_destroy(ic-\u003econn);\n+\t}\n }\n \n void rds_ib_get_mr_info(struct rds_ib_device *rds_ibdev, struct rds_info_rdma_connection *iinfo)\ndiff --git a/net/rds/ib_recv.c b/net/rds/ib_recv.c\nindex bd6cb3ffaa571..7d45808544a0d 100644\n--- a/net/rds/ib_recv.c\n+++ b/net/rds/ib_recv.c\n@@ -458,7 +458,11 @@ void rds_ib_recv_refill(struct rds_connection *conn, int prefill, gfp_t gfp)\n \t    (must_wake ||\n \t    (can_wait \u0026\u0026 rds_ib_ring_low(\u0026ic-\u003ei_recv_ring)) ||\n \t    rds_ib_ring_empty(\u0026ic-\u003ei_recv_ring))) {\n-\t\tqueue_delayed_work(conn-\u003ec_path-\u003ecp_wq, \u0026conn-\u003ec_recv_w, 1);\n+\t\trcu_read_lock();\n+\t\tif (!rds_destroy_pending(conn))\n+\t\t\tqueue_delayed_work(conn-\u003ec_path-\u003ecp_wq,\n+\t\t\t\t\t   \u0026conn-\u003ec_recv_w, 1);\n+\t\trcu_read_unlock();\n \t}\n \tif (can_wait)\n \t\tcond_resched();\ndiff --git a/net/rds/ib_send.c b/net/rds/ib_send.c\nindex d6be95542119f..bc411e96ad12d 100644\n--- a/net/rds/ib_send.c\n+++ b/net/rds/ib_send.c\n@@ -298,8 +298,13 @@ void rds_ib_send_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc)\n \trds_ib_sub_signaled(ic, nr_sig);\n \n \tif (test_and_clear_bit(RDS_LL_SEND_FULL, \u0026conn-\u003ec_flags) ||\n-\t    test_bit(0, \u0026conn-\u003ec_map_queued))\n-\t\tqueue_delayed_work(conn-\u003ec_path-\u003ecp_wq, \u0026conn-\u003ec_send_w, 0);\n+\t    test_bit(0, \u0026conn-\u003ec_map_queued)) {\n+\t\trcu_read_lock();\n+\t\tif (!rds_destroy_pending(conn))\n+\t\t\tqueue_delayed_work(conn-\u003ec_path-\u003ecp_wq,\n+\t\t\t\t\t   \u0026conn-\u003ec_send_w, 0);\n+\t\trcu_read_unlock();\n+\t}\n \n \t/* We expect errors as the qp is drained during shutdown */\n \tif (wc-\u003estatus != IB_WC_SUCCESS \u0026\u0026 rds_conn_up(conn)) {\n@@ -420,8 +425,13 @@ void rds_ib_send_add_credits(struct rds_connection *conn, unsigned int credits)\n \t\t\ttest_bit(RDS_LL_SEND_FULL, \u0026conn-\u003ec_flags) ? \", ll_send_full\" : \"\");\n \n \tatomic_add(IB_SET_SEND_CREDITS(credits), \u0026ic-\u003ei_credits);\n-\tif (test_and_clear_bit(RDS_LL_SEND_FULL, \u0026conn-\u003ec_flags))\n-\t\tqueue_delayed_work(conn-\u003ec_path-\u003ecp_wq, \u0026conn-\u003ec_send_w, 0);\n+\tif (test_and_clear_bit(RDS_LL_SEND_FULL, \u0026conn-\u003ec_flags)) {\n+\t\trcu_read_lock();\n+\t\tif (!rds_destroy_pending(conn))\n+\t\t\tqueue_delayed_work(conn-\u003ec_path-\u003ecp_wq,\n+\t\t\t\t\t   \u0026conn-\u003ec_send_w, 0);\n+\t\trcu_read_unlock();\n+\t}\n \n \tWARN_ON(IB_GET_SEND_CREDITS(credits) \u003e= 16384);\n \ndiff --git a/net/rds/loop.c b/net/rds/loop.c\nindex e6b0750bbedab..42e6b841b42c6 100644\n--- a/net/rds/loop.c\n+++ b/net/rds/loop.c\n@@ -156,6 +156,28 @@ static int rds_loop_conn_alloc(struct rds_connection *conn, gfp_t gfp)\n \treturn 0;\n }\n \n+/* Destroy the connections whose nodes were gathered on @tmp_list.\n+ *\n+ * rds_conn_destroy() can return before the connection is freed, and\n+ * it is the free - rds_loop_conn_free() - that unlinks loop_node.\n+ * @tmp_list lives on the caller's stack, so unlink each node before\n+ * its destroy; the free then finds it empty and leaves it alone.\n+ */\n+static void rds_loop_destroy_gathered_conns(struct list_head *tmp_list)\n+{\n+\tstruct rds_loop_connection *lc, *_lc;\n+\n+\tlist_for_each_entry_safe(lc, _lc, tmp_list, loop_node) {\n+\t\tWARN_ON(rcu_access_pointer(lc-\u003econn-\u003ec_passive));\n+\n+\t\tspin_lock_irq(\u0026loop_conns_lock);\n+\t\tlist_del_init(\u0026lc-\u003eloop_node);\n+\t\tspin_unlock_irq(\u0026loop_conns_lock);\n+\n+\t\trds_conn_destroy(lc-\u003econn);\n+\t}\n+}\n+\n static void rds_loop_conn_free(void *arg)\n {\n \tstruct rds_loop_connection *lc = arg;\n@@ -163,7 +185,9 @@ static void rds_loop_conn_free(void *arg)\n \n \trdsdebug(\"lc %p\\n\", lc);\n \tspin_lock_irqsave(\u0026loop_conns_lock, flags);\n-\tlist_del(\u0026lc-\u003eloop_node);\n+\t/* already unlinked if a transport teardown gathered us first */\n+\tif (!list_empty(\u0026lc-\u003eloop_node))\n+\t\tlist_del(\u0026lc-\u003eloop_node);\n \tspin_unlock_irqrestore(\u0026loop_conns_lock, flags);\n \tkfree(lc);\n }\n@@ -180,7 +204,6 @@ static void rds_loop_conn_path_shutdown(struct rds_conn_path *cp)\n \n void rds_loop_exit(void)\n {\n-\tstruct rds_loop_connection *lc, *_lc;\n \tLIST_HEAD(tmp_list);\n \n \trds_loop_set_unloading();\n@@ -191,10 +214,9 @@ void rds_loop_exit(void)\n \tINIT_LIST_HEAD(\u0026loop_conns);\n \tspin_unlock_irq(\u0026loop_conns_lock);\n \n-\tlist_for_each_entry_safe(lc, _lc, \u0026tmp_list, loop_node) {\n-\t\tWARN_ON(lc-\u003econn-\u003ec_passive);\n-\t\trds_conn_destroy(lc-\u003econn);\n-\t}\n+\trds_loop_destroy_gathered_conns(\u0026tmp_list);\n+\n+\trds_conn_wait_conns_freed(\u0026rds_loop_transport, NULL);\n }\n \n static void rds_loop_kill_conns(struct net *net)\n@@ -212,10 +234,7 @@ static void rds_loop_kill_conns(struct net *net)\n \t}\n \tspin_unlock_irq(\u0026loop_conns_lock);\n \n-\tlist_for_each_entry_safe(lc, _lc, \u0026tmp_list, loop_node) {\n-\t\tWARN_ON(lc-\u003econn-\u003ec_passive);\n-\t\trds_conn_destroy(lc-\u003econn);\n-\t}\n+\trds_loop_destroy_gathered_conns(\u0026tmp_list);\n }\n \n static void __net_exit rds_loop_exit_net(struct net *net)\ndiff --git a/net/rds/message.c b/net/rds/message.c\nindex f25f2592586f7..29e95028e61e4 100644\n--- a/net/rds/message.c\n+++ b/net/rds/message.c\n@@ -182,6 +182,18 @@ static void rds_message_purge(struct rds_message *rm)\n \t\tkref_put(\u0026rm-\u003eatomic.op_rdma_mr-\u003er_kref, __rds_put_mr_final);\n }\n \n+static void rds_message_free(struct rds_message *rm)\n+{\n+\t/* get in rds_send_queue_rm(), rds_send_probe() or the congestion\n+\t * map path of rds_send_xmit().  Messages that were never queued on\n+\t * a connection have no reference to drop.\n+\t */\n+\tif (rm-\u003em_inc.i_conn)\n+\t\trds_conn_put(rm-\u003em_inc.i_conn);\n+\n+\tkfree(rm);\n+}\n+\n static void rds_message_unpin_worker(struct work_struct *work)\n {\n \tstruct rds_message *rm = container_of(work, struct rds_message,\n@@ -192,7 +204,7 @@ static void rds_message_unpin_worker(struct work_struct *work)\n \tif (rm-\u003eatomic.op_unpin_deferred)\n \t\trds_atomic_op_unpin_page(\u0026rm-\u003eatomic);\n \n-\tkfree(rm);\n+\trds_message_free(rm);\n }\n \n void rds_message_put(struct rds_message *rm)\n@@ -217,7 +229,7 @@ void rds_message_put(struct rds_message *rm)\n \t\t\treturn;\n \t\t}\n \n-\t\tkfree(rm);\n+\t\trds_message_free(rm);\n \t}\n }\n EXPORT_SYMBOL_GPL(rds_message_put);\ndiff --git a/net/rds/rdma_transport.c b/net/rds/rdma_transport.c\nindex b15cf316b23a2..584e9867810f4 100644\n--- a/net/rds/rdma_transport.c\n+++ b/net/rds/rdma_transport.c\n@@ -63,6 +63,18 @@ static int rds_rdma_cm_event_handler_cmn(struct rdma_cm_id *cm_id,\n \tif (cm_id-\u003edevice-\u003enode_type == RDMA_NODE_IB_CA)\n \t\ttrans = \u0026rds_ib_transport;\n \n+\t/* cm_id-\u003econtext carries no reference of its own.  Pin the\n+\t * connection for the duration of the handler: what the callbacks\n+\t * below do may drop the last reference other than ours, and the\n+\t * mutex released at out: lives in the connection's path array.\n+\t * A connection already being freed gets no events handled.\n+\t */\n+\tif (conn \u0026\u0026 !rds_conn_get_unless_zero(conn)) {\n+\t\trdsdebug(\"conn %p id %p is being freed, ignoring event\\n\",\n+\t\t\t conn, cm_id);\n+\t\treturn 0;\n+\t}\n+\n \t/* Prevent shutdown from tearing down the connection\n \t * while we're executing. */\n \tif (conn) {\n@@ -171,8 +183,10 @@ static int rds_rdma_cm_event_handler_cmn(struct rdma_cm_id *cm_id,\n \t}\n \n out:\n-\tif (conn)\n+\tif (conn) {\n \t\tmutex_unlock(\u0026conn-\u003ec_cm_lock);\n+\t\trds_conn_put(conn);\n+\t}\n \n \trdsdebug(\"id %p event %u (%s) handling ret %d\\n\", cm_id, event-\u003eevent,\n \t\t rdma_event_msg(event-\u003eevent), ret);\ndiff --git a/net/rds/rds.h b/net/rds/rds.h\nindex 2db49573dacd5..06d48c2821efd 100644\n--- a/net/rds/rds.h\n+++ b/net/rds/rds.h\n@@ -89,7 +89,6 @@ enum {\n #define RDS_RECONNECT_PENDING\t1\n #define RDS_IN_XMIT\t\t2\n #define RDS_RECV_REFILL\t\t3\n-#define\tRDS_DESTROY_PENDING\t4\n \n /* Max number of multipaths per RDS connection. Must be a power of 2 */\n #define\tRDS_MPATH_WORKERS\t8\n@@ -138,6 +137,12 @@ struct rds_conn_path {\n /* One rds_connection per RDS address pair */\n struct rds_connection {\n \tstruct hlist_node\tc_hash_node;\n+\t/* Free of the connection memory (not the teardown of its\n+\t * transport state - that stays synchronous in\n+\t * rds_conn_destroy()) is deferred until the last reference is\n+\t * dropped via rds_conn_put().\n+\t */\n+\tstruct kref\t\tc_refcount;\n \tstruct in6_addr\t\tc_laddr;\n \tstruct in6_addr\t\tc_faddr;\n \tint\t\t\tc_dev_if; /* ifindex used for this conn */\n@@ -148,7 +153,15 @@ struct rds_connection {\n \t\t\t\tc_pad_to_32:29;\n \tint\t\t\tc_npaths;\n \tbool\t\t\tc_with_sport_idx;\n-\tstruct rds_connection\t*c_passive;\n+\t/* Set once, by rds_conn_destroy(), before it cancels the path\n+\t * works; read through rds_destroy_pending().  A site that arms\n+\t * a path work must test the predicate and queue the work inside\n+\t * one rcu_read_lock() section: the synchronize_rcu() that\n+\t * follows the store is what keeps a queue issued after the\n+\t * cancellation from landing on a destroyed workqueue.\n+\t */\n+\tbool\t\t\tc_destroy_in_prog;\n+\tstruct rds_connection __rcu *c_passive;\n \tstruct rds_transport\t*c_trans;\n \n \tstruct rds_cong_map\t*c_lcong;\n@@ -544,6 +557,12 @@ struct rds_transport {\n \tunsigned int\t\tt_prefer_loopback:1,\n \t\t\t\tt_mp_capable:1;\n \tunsigned int\t\tt_type;\n+\t/* Connections of this transport not yet freed; freeing runs\n+\t * asynchronously once rds_conn_destroy() has quiesced a\n+\t * connection, so transport module unload has to wait for this\n+\t * to reach zero (rds_conn_wait_conns_freed()).\n+\t */\n+\tatomic_t\t\tt_conn_count;\n \n \tint (*laddr_check)(struct net *net, const struct in6_addr *addr,\n \t\t\t   __u32 scope_id);\n@@ -819,6 +838,20 @@ struct rds_connection *rds_conn_create_outgoing(struct net *net,\n \t\t\t\t\t\tu8 tos, gfp_t gfp, int dev_if);\n void rds_conn_shutdown(struct rds_conn_path *cpath);\n void rds_conn_destroy(struct rds_connection *conn);\n+void rds_conn_get(struct rds_connection *conn);\n+void rds_conn_put(struct rds_connection *conn);\n+/* take a reference unless the connection is already being freed */\n+static inline bool rds_conn_get_unless_zero(struct rds_connection *conn)\n+{\n+\treturn kref_get_unless_zero(\u0026conn-\u003ec_refcount);\n+}\n+/* transport unload waits for its connections to be freed, polling at\n+ * the first interval and warning at the second\n+ */\n+#define RDS_CONN_FREE_POLL_MS\t\t100\n+#define RDS_CONN_FREE_WARN_INTERVAL_MS\t10000\n+void rds_conn_wait_conns_freed(struct rds_transport *trans,\n+\t\t\t       void (*resweep)(void));\n void rds_conn_drop(struct rds_connection *conn);\n void rds_conn_path_drop(struct rds_conn_path *cpath, bool destroy);\n void rds_conn_connect_if_down(struct rds_connection *conn);\n@@ -994,7 +1027,8 @@ void __rds_put_mr_final(struct kref *kref);\n \n static inline bool rds_destroy_pending(struct rds_connection *conn)\n {\n-\treturn !check_net(rds_conn_net(conn)) ||\n+\treturn READ_ONCE(conn-\u003ec_destroy_in_prog) ||\n+\t       !check_net(rds_conn_net(conn)) ||\n \t       (conn-\u003ec_trans-\u003et_unloading \u0026\u0026 conn-\u003ec_trans-\u003et_unloading(conn));\n }\n \ndiff --git a/net/rds/recv.c b/net/rds/recv.c\nindex 6204e577a90ae..b031c0b43af83 100644\n--- a/net/rds/recv.c\n+++ b/net/rds/recv.c\n@@ -46,6 +46,7 @@ void rds_inc_init(struct rds_incoming *inc, struct rds_connection *conn,\n {\n \trefcount_set(\u0026inc-\u003ei_refcount, 1);\n \tINIT_LIST_HEAD(\u0026inc-\u003ei_item);\n+\trds_conn_get(conn);\t/* put in rds_inc_put() */\n \tinc-\u003ei_conn = conn;\n \tinc-\u003ei_conn_path = NULL;\n \tinc-\u003ei_saddr = *saddr;\n@@ -61,6 +62,7 @@ void rds_inc_path_init(struct rds_incoming *inc, struct rds_conn_path *cp,\n {\n \trefcount_set(\u0026inc-\u003ei_refcount, 1);\n \tINIT_LIST_HEAD(\u0026inc-\u003ei_item);\n+\trds_conn_get(cp-\u003ecp_conn);\t/* put in rds_inc_put() */\n \tinc-\u003ei_conn = cp-\u003ecp_conn;\n \tinc-\u003ei_conn_path = cp;\n \tinc-\u003ei_saddr = *saddr;\n@@ -81,9 +83,19 @@ void rds_inc_put(struct rds_incoming *inc)\n {\n \trdsdebug(\"put inc %p ref %d\\n\", inc, refcount_read(\u0026inc-\u003ei_refcount));\n \tif (refcount_dec_and_test(\u0026inc-\u003ei_refcount)) {\n+\t\tstruct rds_connection *conn = inc-\u003ei_conn;\n+\n \t\tBUG_ON(!list_empty(\u0026inc-\u003ei_item));\n \n-\t\tinc-\u003ei_conn-\u003ec_trans-\u003einc_free(inc);\n+\t\t/* inc_free() can free the memory @inc lives in, so the\n+\t\t * connection reference has to be dropped through the\n+\t\t * copy taken above.\n+\t\t */\n+\t\tconn-\u003ec_trans-\u003einc_free(inc);\n+\t\t/* get in rds_inc_init(), rds_inc_path_init() or\n+\t\t * rds_recv_incoming()\n+\t\t */\n+\t\trds_conn_put(conn);\n \t}\n }\n EXPORT_SYMBOL_GPL(rds_inc_put);\n@@ -325,6 +337,13 @@ void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,\n \tunsigned long flags;\n \tstruct rds_conn_path *cp;\n \n+\t/* every caller initialized @inc with rds_inc_init() or\n+\t * rds_inc_path_init() first, so i_conn already holds a reference.\n+\t * Take the new one before dropping the old, so that re-pointing an\n+\t * inc at the connection it already refers to cannot free it.\n+\t */\n+\trds_conn_get(conn);\n+\trds_conn_put(inc-\u003ei_conn);\n \tinc-\u003ei_conn = conn;\n \tinc-\u003ei_rx_jiffies = jiffies;\n \tif (conn-\u003ec_trans-\u003et_mp_capable)\ndiff --git a/net/rds/send.c b/net/rds/send.c\nindex 1afa981e5c06d..f59a487b079d5 100644\n--- a/net/rds/send.c\n+++ b/net/rds/send.c\n@@ -290,6 +290,8 @@ int rds_send_xmit(struct rds_conn_path *cp)\n \t\t\t}\n \t\t\trm-\u003edata.op_active = 1;\n \t\t\trm-\u003em_inc.i_conn_path = cp;\n+\t\t\t/* put in rds_message_put() */\n+\t\t\trds_conn_get(cp-\u003ecp_conn);\n \t\t\trm-\u003em_inc.i_conn = cp-\u003ecp_conn;\n \n \t\t\tcp-\u003ecp_xmit_rm = rm;\n@@ -928,6 +930,19 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,\n \t * and poll() now knows no more data can be sent.\n \t */\n \tif (rs-\u003ers_snd_bytes \u003c rds_sk_sndbuf(rs)) {\n+\t\t/* rds_conn_path_quiesce() empties cp_send_queue under\n+\t\t * cp_lock once the connection's destroy has begun.  Test\n+\t\t * for that under the same lock, before touching either\n+\t\t * queue: a message added after the purge would hold a\n+\t\t * connection reference nothing ever drops.\n+\t\t */\n+\t\tspin_lock(\u0026cp-\u003ecp_lock);\n+\t\tif (rds_destroy_pending(conn)) {\n+\t\t\tspin_unlock(\u0026cp-\u003ecp_lock);\n+\t\t\t*queued = -EAGAIN;\n+\t\t\tgoto unlock;\n+\t\t}\n+\n \t\trs-\u003ers_snd_bytes += len;\n \n \t\t/* let recv side know we are close to send space exhaustion.\n@@ -947,11 +962,11 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,\n \t\t/* The code ordering is a little weird, but we're\n \t\t   trying to minimize the time we hold c_lock */\n \t\trds_message_populate_header(\u0026rm-\u003em_inc.i_hdr, sport, dport, 0);\n+\t\trds_conn_get(conn);\t/* put in rds_message_put() */\n \t\trm-\u003em_inc.i_conn = conn;\n \t\trm-\u003em_inc.i_conn_path = cp;\n \t\trds_message_addref(rm);\n \n-\t\tspin_lock(\u0026cp-\u003ecp_lock);\n \t\trm-\u003em_inc.i_hdr.h_sequence = cpu_to_be64(cp-\u003ecp_next_tx_seq++);\n \t\tlist_add_tail(\u0026rm-\u003em_conn_item, \u0026cp-\u003ecp_send_queue);\n \t\tset_bit(RDS_MSG_ON_CONN, \u0026rm-\u003em_flags);\n@@ -964,6 +979,7 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,\n \t\t*queued = 1;\n \t}\n \n+unlock:\n \tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n out:\n \treturn *queued;\n@@ -1159,13 +1175,14 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \tDECLARE_SOCKADDR(struct sockaddr_in *, usin, msg-\u003emsg_name);\n \t__be16 dport;\n \tstruct rds_message *rm = NULL;\n-\tstruct rds_connection *conn;\n+\tstruct rds_connection *conn = NULL;\n \tint ret = 0;\n \tint queued = 0, allocated_mr = 0;\n \tint nonblock = msg-\u003emsg_flags \u0026 MSG_DONTWAIT;\n \tlong timeo = sock_sndtimeo(sk, nonblock);\n \tstruct rds_conn_path *cpath;\n \tstruct in6_addr daddr;\n+\tunsigned long flags;\n \t__u32 scope_id = 0;\n \tsize_t rdma_payload_len = 0;\n \tbool zcopy = ((msg-\u003emsg_flags \u0026 MSG_ZEROCOPY) \u0026\u0026\n@@ -1340,11 +1357,29 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \trm-\u003em_daddr = daddr;\n \n \t/* rds_conn_create has a spinlock that runs with IRQ off.\n-\t * Caching the conn in the socket helps a lot. */\n-\tif (rs-\u003ers_conn \u0026\u0026 ipv6_addr_equal(\u0026rs-\u003ers_conn-\u003ec_faddr, \u0026daddr) \u0026\u0026\n-\t    rs-\u003ers_tos == rs-\u003ers_conn-\u003ec_tos) {\n-\t\tconn = rs-\u003ers_conn;\n+\t * Caching the conn in the socket helps a lot.\n+\t *\n+\t * The cached rs_conn holds a connection reference; take one of\n+\t * our own for the duration of this call (dropped on both exit\n+\t * paths), so that neither a concurrent sender replacing the\n+\t * cache nor rds_conn_destroy() can free the connection under\n+\t * us.  A cached connection whose destruction has begun is not\n+\t * reused: dropping it here lets the next sendmsg look up or\n+\t * create a live one instead of returning -EAGAIN forever.\n+\t */\n+\tspin_lock_irqsave(\u0026rs-\u003ers_lock, flags);\n+\tconn = rs-\u003ers_conn;\n+\tif (conn \u0026\u0026 ipv6_addr_equal(\u0026conn-\u003ec_faddr, \u0026daddr) \u0026\u0026\n+\t    rs-\u003ers_tos == conn-\u003ec_tos \u0026\u0026 !rds_destroy_pending(conn)) {\n+\t\trds_conn_get(conn);\n \t} else {\n+\t\tconn = NULL;\n+\t}\n+\tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n+\n+\tif (!conn) {\n+\t\tstruct rds_connection *old;\n+\n \t\tconn = rds_conn_create_outgoing(sock_net(sock-\u003esk),\n \t\t\t\t\t\t\u0026rs-\u003ers_bound_addr, \u0026daddr,\n \t\t\t\t\t\trs-\u003ers_transport, rs-\u003ers_tos,\n@@ -1352,9 +1387,17 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \t\t\t\t\t\tscope_id);\n \t\tif (IS_ERR(conn)) {\n \t\t\tret = PTR_ERR(conn);\n+\t\t\tconn = NULL;\n \t\t\tgoto out;\n \t\t}\n+\t\t/* hand the cache its own reference */\n+\t\trds_conn_get(conn);\n+\t\tspin_lock_irqsave(\u0026rs-\u003ers_lock, flags);\n+\t\told = rs-\u003ers_conn;\n \t\trs-\u003ers_conn = conn;\n+\t\tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n+\t\tif (old)\n+\t\t\trds_conn_put(old);\n \t}\n \n \tif (conn-\u003ec_trans-\u003et_mp_capable) {\n@@ -1378,9 +1421,14 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \t\t * outstanding.\n \t\t */\n \t\tif (!test_and_set_bit(RDS_RECONNECT_PENDING,\n-\t\t\t\t      \u0026conn-\u003ec_path[0].cp_flags))\n-\t\t\tqueue_delayed_work(conn-\u003ec_path[0].cp_wq,\n-\t\t\t\t\t   \u0026conn-\u003ec_path[0].cp_conn_w, 0);\n+\t\t\t\t      \u0026conn-\u003ec_path[0].cp_flags)) {\n+\t\t\trcu_read_lock();\n+\t\t\tif (!rds_destroy_pending(conn))\n+\t\t\t\tqueue_delayed_work(conn-\u003ec_path[0].cp_wq,\n+\t\t\t\t\t\t   \u0026conn-\u003ec_path[0].cp_conn_w,\n+\t\t\t\t\t\t   0);\n+\t\t\trcu_read_unlock();\n+\t\t}\n \t\trds_send_ping(conn, 0);\n \t}\n \n@@ -1442,6 +1490,11 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \t\t\tret = -ETIMEDOUT;\n \t\tgoto out;\n \t}\n+\t/* rds_send_queue_rm() refused: the connection is being destroyed */\n+\tif (queued \u003c 0) {\n+\t\tret = queued;\n+\t\tgoto out;\n+\t}\n \n \t/*\n \t * By now we've committed to the send.  We reuse rds_send_worker()\n@@ -1469,6 +1522,8 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \t\tkfree(vct.vec[ind].iov);\n \tkfree(vct.vec);\n \n+\trds_conn_put(conn);\n+\n \treturn payload_len;\n \n out:\n@@ -1476,6 +1531,9 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)\n \t\tkfree(vct.vec[ind].iov);\n \tkfree(vct.vec);\n \n+\tif (conn)\n+\t\trds_conn_put(conn);\n+\n \t/* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.\n \t * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN\n \t * or in any other way, we need to destroy the MR again */\n@@ -1522,6 +1580,7 @@ rds_send_probe(struct rds_conn_path *cp, __be16 sport,\n \tlist_add_tail(\u0026rm-\u003em_conn_item, \u0026cp-\u003ecp_send_queue);\n \tset_bit(RDS_MSG_ON_CONN, \u0026rm-\u003em_flags);\n \trds_message_addref(rm);\n+\trds_conn_get(cp-\u003ecp_conn);\t/* put in rds_message_put() */\n \trm-\u003em_inc.i_conn = cp-\u003ecp_conn;\n \trm-\u003em_inc.i_conn_path = cp;\n \ndiff --git a/net/rds/tcp.c b/net/rds/tcp.c\nindex 774a71f88d375..a71d6a4f0939c 100644\n--- a/net/rds/tcp.c\n+++ b/net/rds/tcp.c\n@@ -502,6 +502,28 @@ static bool rds_tcp_is_unloading(struct rds_connection *conn)\n \treturn atomic_read(\u0026rds_tcp_unloading) != 0;\n }\n \n+/* Destroy the connections whose nodes were gathered on @tmp_list.\n+ *\n+ * rds_conn_destroy() can return before the connection is freed, and\n+ * it is the free - rds_tcp_conn_free() - that unlinks t_tcp_node.\n+ * Since @tmp_list lives on the caller's stack, unlink each node here\n+ * and mark it detached before its destroy, so that a free that runs\n+ * after the caller has returned does not write into a dead frame.\n+ */\n+static void rds_tcp_destroy_gathered_conns(struct list_head *tmp_list)\n+{\n+\tstruct rds_tcp_connection *tc, *_tc;\n+\n+\tlist_for_each_entry_safe(tc, _tc, tmp_list, t_tcp_node) {\n+\t\tspin_lock_irq(\u0026rds_tcp_conn_lock);\n+\t\tlist_del_init(\u0026tc-\u003et_tcp_node);\n+\t\ttc-\u003et_tcp_node_detached = true;\n+\t\tspin_unlock_irq(\u0026rds_tcp_conn_lock);\n+\n+\t\trds_conn_destroy(tc-\u003et_cpath-\u003ecp_conn);\n+\t}\n+}\n+\n static void rds_tcp_destroy_conns(void)\n {\n \tstruct rds_tcp_connection *tc, *_tc;\n@@ -515,8 +537,7 @@ static void rds_tcp_destroy_conns(void)\n \t}\n \tspin_unlock_irq(\u0026rds_tcp_conn_lock);\n \n-\tlist_for_each_entry_safe(tc, _tc, \u0026tmp_list, t_tcp_node)\n-\t\trds_conn_destroy(tc-\u003et_cpath-\u003ecp_conn);\n+\trds_tcp_destroy_gathered_conns(\u0026tmp_list);\n }\n \n static void rds_tcp_exit(void);\n@@ -698,8 +719,7 @@ static void rds_tcp_kill_sock(struct net *net)\n \t\t}\n \t}\n \tspin_unlock_irq(\u0026rds_tcp_conn_lock);\n-\tlist_for_each_entry_safe(tc, _tc, \u0026tmp_list, t_tcp_node)\n-\t\trds_conn_destroy(tc-\u003et_cpath-\u003ecp_conn);\n+\trds_tcp_destroy_gathered_conns(\u0026tmp_list);\n }\n \n static void __net_exit rds_tcp_exit_net(struct net *net)\n@@ -805,6 +825,7 @@ static void rds_tcp_exit(void)\n #endif\n \tunregister_pernet_device(\u0026rds_tcp_net_ops);\n \trds_tcp_destroy_conns();\n+\trds_conn_wait_conns_freed(\u0026rds_tcp_transport, NULL);\n \trds_trans_unregister(\u0026rds_tcp_transport);\n \trds_tcp_recv_exit();\n \tkmem_cache_destroy(rds_tcp_conn_slab);\ndiff --git a/net/rds/tcp_listen.c b/net/rds/tcp_listen.c\nindex 13fa60c1985bb..e22ea9ca8c1ce 100644\n--- a/net/rds/tcp_listen.c\n+++ b/net/rds/tcp_listen.c\n@@ -153,7 +153,7 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)\n {\n \tstruct socket *listen_sock = rtn-\u003erds_tcp_listen_sock;\n \tstruct socket *new_sock = NULL;\n-\tstruct rds_connection *conn;\n+\tstruct rds_connection *conn = NULL;\n \tint ret;\n \tstruct inet_sock *inet;\n \tstruct rds_tcp_connection *rs_tcp = NULL;\n@@ -229,6 +229,7 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)\n \n \tif (IS_ERR(conn)) {\n \t\tret = PTR_ERR(conn);\n+\t\tconn = NULL;\n \t\tgoto out;\n \t}\n \t/* An incoming SYN request came in, and TCP just accepted it.\n@@ -277,7 +278,15 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)\n \tcp = rs_tcp-\u003et_cpath;\n \tconn_state = rds_conn_path_state(cp);\n \tWARN_ON(conn_state == RDS_CONN_UP);\n-\tif (conn_state != RDS_CONN_CONNECTING \u0026\u0026 conn_state != RDS_CONN_ERROR) {\n+\t/* A connection whose destroy has begun has been quiesced and is\n+\t * only waiting for its last reference: its paths sit in\n+\t * RDS_CONN_DOWN, which rds_tcp_accept_one_path() happily claims.\n+\t * Installing a socket on it would leave sk_user_data pointing\n+\t * at a path that is about to be freed.\n+\t */\n+\tif (rds_destroy_pending(conn) ||\n+\t    (conn_state != RDS_CONN_CONNECTING \u0026\u0026\n+\t     conn_state != RDS_CONN_ERROR)) {\n \t\trds_conn_path_drop(cp, 0);\n \t\tgoto rst_nsk;\n \t}\n@@ -316,10 +325,14 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)\n \t */\n \tif (READ_ONCE(sk-\u003esk_state) == TCP_CLOSE_WAIT ||\n \t    READ_ONCE(sk-\u003esk_state) == TCP_LAST_ACK ||\n-\t    READ_ONCE(sk-\u003esk_state) == TCP_CLOSE)\n+\t    READ_ONCE(sk-\u003esk_state) == TCP_CLOSE) {\n \t\trds_conn_path_drop(cp, 0);\n-\telse\n-\t\tqueue_delayed_work(cp-\u003ecp_wq, \u0026cp-\u003ecp_recv_w, 0);\n+\t} else {\n+\t\trcu_read_lock();\n+\t\tif (!rds_destroy_pending(cp-\u003ecp_conn))\n+\t\t\tqueue_delayed_work(cp-\u003ecp_wq, \u0026cp-\u003ecp_recv_w, 0);\n+\t\trcu_read_unlock();\n+\t}\n \n \tsock_put(sk);\n \n@@ -343,6 +356,8 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)\n \t\tmutex_unlock(\u0026rs_tcp-\u003et_conn_path_lock);\n \tif (new_sock)\n \t\tsock_release(new_sock);\n+\tif (conn)\n+\t\trds_conn_put(conn);\n \n \tmutex_unlock(\u0026rtn-\u003erds_tcp_accept_lock);\n \n"
}

2/1 2026/09/14 04:28 agent patch-evaluator
0m Model:
core

Args:
null

Results:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND",
    "INFINIBAND_ADDR_TRANS"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_destroy_fini",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_conn_create_outgoing",
    "rds_sendmsg",
    "rds_send_queue_rm",
    "rds_release",
    "rds_tcp_accept_one"
  ],
  "Reasoning": "The patch substantially refactors connection lifetime and reference counting (using kref) in the RDS subsystem (net/rds), deferring connection cleanup until all references from messages, incoming queues, sockets, and transports are dropped. These changes touch core networking logic reachable via AF_RDS sockets and could introduce subtle race conditions, refcounting leaks, use-after-free, or deadlocks.",
  "WorthFuzzing": true
}

Instruction:
You are an expert Linux kernel maintainer and security engineer.
Your job is to review a provided patch series and evaluate whether it warrants fuzzing with syzkaller.

IMPORTANT: The changes have ALREADY been applied and committed as the HEAD commit in
your workspace. Do NOT rely on internal assumptions. You must actively use your code access
tools to inspect the actual source code, callers, and surrounding context.

================================================================================
1. CORE TRIAGE PHILOSOPHY
================================================================================
The goal of patch fuzzing is to discover crashes, regressions, exposed latent bugs,
and newly triggered assertions introduced by the patch series.

- REACHABILITY IS THE PRIMARY GATE:
  Fuzzing can only discover bugs in code that can actually execute in standard virtualized
  environments (GCE or QEMU, utilizing software-emulated devices like USB gadgets, netdev, tun/tap).
  If the modified code is structurally unreachable (see Section 2), it MUST NOT be fuzzed,
  regardless of whether it adds assertions or complex logic.

- DO NOT BLINDLY TRUST "NO FUNCTIONAL CHANGE" (NFCI) OR "REFACTORING" CLAIMS:
  Patch authors routinely label changes as "cleanups", "refactorings", or state
  "No functional change intended". Do NOT take these claims at face value.
  Code refactorings that rearrange logic, introduce helper functions, or alter state management
  in core subsystems frequently introduce subtle semantic shifts or uncover latent kernel bugs.
  If reachable executable code is modified or refactored, it MUST be fuzzed.

- NEW OR MODIFIED ASSERTIONS IN REACHABLE CODE MUST BE FUZZED:
  When a patch introduces or modifies runtime checks or assertions (e.g., WARN_ON*, VM_WARN_ON*,
  BUG_ON*, lockdep_assert*) in reachable code paths, it enforces new or stricter invariants.
  Even if the author believes the invariant always holds, fuzzing is essential to verify whether
  an unusual sequence of operations can violate it.

================================================================================
2. WHEN TO RETURN WorthFuzzing=false (NEGATIVE CRITERIA)
================================================================================
Return WorthFuzzing=false ONLY IF all modified code falls strictly into one or more of these categories:

- Non-kernel and non-executable changes:
  * Modifications to Documentation/, comments, or spelling fixes.
  * User-space directories, self-tests, samples, or scripts (e.g., tools/, samples/, scripts/, usr/)
    that do not affect the compiled kernel image (vmlinux) or kernel modules.
  * Purely decorative logging (e.g., message strings in pr_err, printk, dev_info) or tracepoints
    that do not alter control flow or data structures.
  * Build system or Kconfig changes that do not alter compiled C logic.
- Structurally unreachable hardware:
  * Vendor-specific PCIe switches, SmartNICs, or GPU drivers (e.g., mlxsw, pds_core, qed,
    ionic, amdgpu) requiring physical ASIC/PCIe cards not emulated in standard QEMU.
- Unreachable execution paths:
  * Driver teardown callbacks (.remove, .shutdown, pci_unregister_driver) executed only during
    physical PCI hot-unplug or manual sysfs driver unbinding.
  * Code paths exclusive to architectures other than the target architecture.

================================================================================
3. WHEN TO RETURN WorthFuzzing=true (POSITIVE CRITERIA)
================================================================================
Return WorthFuzzing=true whenever the patch touches reachable executable code, including:
- Core Subsystems:
  * Any logic modifications in memory management (mm/), synchronization/locking (kernel/locking/),
    BPF, scheduler, core networking, VFS, or syscall handling.
- Refactorings and Code Cleanups:
  * Any restructuring of reachable data structures, helper abstractions, or algorithm flows.
- Runtime Assertions and Defensive Checks:
  * Any introduction or alteration of assertions (WARN_ON*, VM_WARN_ON*, BUG_ON*, etc.) in reachable paths.
- Reachable Drivers and Protocols:
  * Drivers accessible via virtual buses (virtio, USB gadget, loopback, netlink, binder, sockets, etc.).

================================================================================
4. EXTRACTING FocusSymbols (PREVENTING DILUTION)
================================================================================
When WorthFuzzing=true, you must extract specific kernel functions into FocusSymbols to guide the fuzzer:

- AVOID UBIQUITOUS LIFECYCLE HOT-PATHS:
  Do NOT list generic, ubiquitous functions called by almost every program in the corpus
  (including, but not limited to: general memory allocators and deallocators, page fault
  and trap handlers, or core synchronization primitives; this is not an exhaustive list).
  Listing ubiquitous functions causes the fuzzer to classify thousands of unrelated tests as "focused",
  which severely dilutes fuzzing effort away from the actual changes.

- TARGET SPECIFIC FEATURE LOGIC AND ENTRYPOINTS:
  List functions that specifically implement the logic being added or altered, or direct API entrypoints
  for the subsystem feature under review.

- HANDLING STATIC INLINE FUNCTIONS IN HEADERS (.h):
  Compiler-inlined static functions (such as static inlines in mm/*.h or include/linux/*.h) lack
  distinct symbol addresses in vmlinux and cannot be targeted directly by symbol coverage filters.
  If the changes are primarily in static inline helpers, identify non-static, feature-specific caller
  functions in .c files that exercise them (avoiding ubiquitous lifecycle wrappers).

================================================================================
5. IDENTIFYING EnableConfigs
================================================================================
Identify any specific CONFIG_ options required to properly compile and reach the modified code:
- Inspect Kconfig files and #ifdef guards; do not make assumptions.
- Check "depends on" lines in Kconfig to include any non-standard parent subsystem configs needed.
- Strip any 'CONFIG_' prefix (e.g., return "NET_IPV4" instead of "CONFIG_NET_IPV4").
Prefer calling several tools at the same time to save round-trips.


Use set-results tool to provide results of the analysis.
It must be called exactly once before the final reply.
Ignore results of this tool.

Prompt:
Target architecture: amd64

For your convenience, here is the diff of the changes:
commit fe0d65dad5bf3cdb3ebccbae9d834aa98bc8f6e0
Author: syz-cluster <triage@syzkaller.com>
Date:   Mon Sep 14 04:28:33 2026 +0000

    syz-cluster: applied patch under review

diff --git a/net/rds/af_rds.c b/net/rds/af_rds.c
index d5defe9172e36..1cc20b5cfd213 100644
--- a/net/rds/af_rds.c
+++ b/net/rds/af_rds.c
@@ -80,6 +80,14 @@ static int rds_release(struct socket *sock)
 	rds_notify_queue_get(rs, NULL);
 	rds_notify_msg_zcopy_purge(&rs->rs_zcookie_queue);
 
+	/* drop the cached connection reference; no sendmsg can race
+	 * with us here, the socket is going away
+	 */
+	if (rs->rs_conn) {
+		rds_conn_put(rs->rs_conn);
+		rs->rs_conn = NULL;
+	}
+
 	spin_lock_bh(&rds_sock_lock);
 	list_del_init(&rs->rs_item);
 	spin_unlock_bh(&rds_sock_lock);
@@ -255,6 +263,7 @@ static int rds_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
 {
 	struct rds_sock *rs = rds_sk_to_rs(sock->sk);
 	rds_tos_t utos, tos = 0;
+	unsigned long flags;
 
 	switch (cmd) {
 	case SIOCRDSSETTOS:
@@ -267,18 +276,23 @@ static int rds_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
 		else
 			return -ENOIOCTLCMD;
 
-		spin_lock_bh(&rds_sock_lock);
+		/* rs_conn is serialized by rs_lock (see rds_sendmsg());
+		 * hold it across the "no connection yet" check and the
+		 * rs_tos store so a racing sendmsg cannot cache a conn
+		 * whose c_tos then disagrees with rs_tos.
+		 */
+		spin_lock_irqsave(&rs->rs_lock, flags);
 		if (rs->rs_tos || rs->rs_conn) {
-			spin_unlock_bh(&rds_sock_lock);
+			spin_unlock_irqrestore(&rs->rs_lock, flags);
 			return -EINVAL;
 		}
 		rs->rs_tos = tos;
-		spin_unlock_bh(&rds_sock_lock);
+		spin_unlock_irqrestore(&rs->rs_lock, flags);
 		break;
 	case SIOCRDSGETTOS:
-		spin_lock_bh(&rds_sock_lock);
+		spin_lock_irqsave(&rs->rs_lock, flags);
 		tos = rs->rs_tos;
-		spin_unlock_bh(&rds_sock_lock);
+		spin_unlock_irqrestore(&rs->rs_lock, flags);
 		if (put_user(tos, (rds_tos_t __user *)arg))
 			return -EFAULT;
 		break;
diff --git a/net/rds/connection.c b/net/rds/connection.c
index b6c4beb50eaf0..f5f25e4a53b97 100644
--- a/net/rds/connection.c
+++ b/net/rds/connection.c
@@ -47,7 +47,8 @@
 
 /* converting this to RCU is a chore for another day.. */
 static DEFINE_SPINLOCK(rds_conn_lock);
-static unsigned long rds_conn_count;
+/* woken whenever a transport's t_conn_count drops to zero */
+static DECLARE_WAIT_QUEUE_HEAD(rds_conn_freed_waitq);
 static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];
 static struct kmem_cache *rds_conn_slab;
 
@@ -79,7 +80,18 @@ static struct hlist_head *rds_conn_bucket(const struct in6_addr *laddr,
 		var |= RDS_INFO_CONNECTION_FLAG_##suffix;	\
 } while (0)
 
-/* rcu read lock must be held or the connection spinlock */
+/* rcu read lock must be held or the connection spinlock.
+ * On success a reference is taken on the returned connection; the
+ * caller must drop it with rds_conn_put().
+ */
+/* c_passive is written under rds_conn_lock and read under RCU */
+static struct rds_connection *
+rds_conn_passive_locked(struct rds_connection *conn)
+{
+	return rcu_dereference_protected(conn->c_passive,
+					 lockdep_is_held(&rds_conn_lock));
+}
+
 static struct rds_connection *rds_conn_lookup(struct net *net,
 					      struct hlist_head *head,
 					      const struct in6_addr *laddr,
@@ -96,6 +108,17 @@ static struct rds_connection *rds_conn_lookup(struct net *net,
 		    conn->c_tos == tos &&
 		    net == rds_conn_net(conn) &&
 		    conn->c_dev_if == dev_if) {
+			/* Only ever hand out a live reference.
+			 * rds_conn_destroy() unhashes under
+			 * rds_conn_lock and waits a grace period
+			 * before dropping the initial reference, so
+			 * an entry this traversal reaches still holds
+			 * at least that one; the conditional get
+			 * documents the contract rather than
+			 * papering over a zero-refcount entry.
+			 */
+			if (!kref_get_unless_zero(&conn->c_refcount))
+				continue;
 			ret = conn;
 			break;
 		}
@@ -161,6 +184,22 @@ static void __rds_conn_path_init(struct rds_connection *conn,
 	cp->cp_flags = 0;
 }
 
+/* Undo trans->conn_alloc(): it may have allocated transport data for
+ * every path of a multipath connection, not just for path 0.
+ */
+static void rds_conn_free_transport_data(struct rds_connection *conn,
+					 int npaths)
+{
+	struct rds_conn_path *cp;
+	int i;
+
+	for (i = 0; i < npaths; i++) {
+		cp = &conn->c_path[i];
+		if (cp->cp_transport_data)
+			conn->c_trans->conn_free(cp->cp_transport_data);
+	}
+}
+
 /*
  * There is only every one 'conn' for a given pair of addresses in the
  * system at a time.  They contain messages to be retransmitted and so
@@ -197,7 +236,20 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		 * We need a second connection object into which we
 		 * can stick the other QP. */
 		parent = conn;
-		conn = parent->c_passive;
+		/* The c_passive pointer holds a reference which is only
+		 * dropped one synchronize_rcu() after the pointer is
+		 * cleared, so within this RCU section a fetched pointer
+		 * is always safe to take a reference on.  A passive conn
+		 * whose own destroy has begun is not handed out, though:
+		 * it is quiesced and about to clear the parent's pointer
+		 * itself, and reusing it would re-arm a connection that
+		 * nothing will tear down again.
+		 */
+		conn = rcu_dereference(parent->c_passive);
+		if (conn && READ_ONCE(conn->c_destroy_in_prog))
+			conn = NULL;
+		if (conn)
+			rds_conn_get(conn);
 	}
 	rcu_read_unlock();
 	if (conn)
@@ -215,6 +267,7 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		goto out;
 	}
 
+	kref_init(&conn->c_refcount);
 	INIT_HLIST_NODE(&conn->c_hash_node);
 	conn->c_laddr = *laddr;
 	conn->c_isv6 = !ipv6_addr_v4mapped(laddr);
@@ -278,12 +331,13 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 
 	init_waitqueue_head(&conn->c_hs_waitq);
 	for (i = 0; i < npaths; i++) {
+		int seq = atomic_read(&trans->t_conn_count);
+
 		__rds_conn_path_init(conn, &conn->c_path[i],
 				     is_outgoing);
 		conn->c_path[i].cp_index = i;
 		conn->c_path[i].cp_wq =
-			alloc_ordered_workqueue("krds_cp_wq#%lu/%d", 0,
-						rds_conn_count, i);
+			alloc_ordered_workqueue("krds_cp_wq#%d/%d", 0, seq, i);
 		if (!conn->c_path[i].cp_wq)
 			conn->c_path[i].cp_wq = rds_wq;
 	}
@@ -315,15 +369,46 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 	spin_lock_irqsave(&rds_conn_lock, flags);
 	if (parent) {
 		/* Creating passive conn */
-		if (parent->c_passive) {
-			trans->conn_free(conn->c_path[0].cp_transport_data);
+		if (READ_ONCE(parent->c_destroy_in_prog)) {
+			/* The parent's destroy has begun (it sets the
+			 * flag and snatches c_passive under this
+			 * lock); do not install a new passive conn
+			 * that nothing would ever destroy.
+			 */
+			rds_conn_free_transport_data(conn, npaths);
+			free_cp = conn->c_path;
+			kmem_cache_free(rds_conn_slab, conn);
+			conn = ERR_PTR(-ENETDOWN);
+		} else if (rcu_access_pointer(parent->c_passive)) {
+			struct rds_connection *passive;
+
+			passive = rds_conn_passive_locked(parent);
+			rds_conn_free_transport_data(conn, npaths);
 			free_cp = conn->c_path;
 			kmem_cache_free(rds_conn_slab, conn);
-			conn = parent->c_passive;
+			if (READ_ONCE(passive->c_destroy_in_prog)) {
+				/* Its destroy will clear the parent's
+				 * pointer under this lock shortly; until
+				 * then there is no usable passive conn.
+				 */
+				conn = ERR_PTR(-ENETDOWN);
+			} else {
+				rds_conn_get(passive);
+				conn = passive;
+			}
 		} else {
-			parent->c_passive = conn;
+			/* The initial reference belongs to whoever
+			 * destroys the conn (the transport's conn
+			 * lists, as for any other conn).  Take one
+			 * for the c_passive pointer - dropped when
+			 * the parent is destroyed - and one for our
+			 * caller.
+			 */
+			rds_conn_get(conn);	/* c_passive */
+			rds_conn_get(conn);	/* caller */
+			rcu_assign_pointer(parent->c_passive, conn);
 			rds_cong_add_conn(conn);
-			rds_conn_count++;
+			atomic_inc(&conn->c_trans->t_conn_count);
 		}
 	} else {
 		/* Creating normal conn */
@@ -332,33 +417,28 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		found = rds_conn_lookup(net, head, laddr, faddr, trans,
 					tos, dev_if);
 		if (found) {
-			struct rds_conn_path *cp;
-			int i;
-
-			for (i = 0; i < npaths; i++) {
-				cp = &conn->c_path[i];
-				/* The ->conn_alloc invocation may have
-				 * allocated resource for all paths, so all
-				 * of them may have to be freed here.
-				 */
-				if (cp->cp_transport_data)
-					trans->conn_free(cp->cp_transport_data);
-			}
+			rds_conn_free_transport_data(conn, npaths);
 			free_cp = conn->c_path;
 			kmem_cache_free(rds_conn_slab, conn);
 			conn = found;
 		} else {
 			conn->c_my_gen_num = rds_gen_num;
 			conn->c_peer_gen_num = 0;
+			/* the initial reference belongs to whoever
+			 * destroys the conn; take one for our caller
+			 */
+			rds_conn_get(conn);
 			hlist_add_head_rcu(&conn->c_hash_node, head);
 			rds_cong_add_conn(conn);
-			rds_conn_count++;
+			atomic_inc(&conn->c_trans->t_conn_count);
 		}
 	}
 	spin_unlock_irqrestore(&rds_conn_lock, flags);
 	rcu_read_unlock();
 
 out:
+	if (parent)
+		rds_conn_put(parent);
 	if (free_cp) {
 		for (i = 0; i < npaths; i++)
 			if (free_cp[i].cp_wq != rds_wq)
@@ -466,9 +546,10 @@ void rds_conn_shutdown(struct rds_conn_path *cp)
 			 * Quiesce the reconnect timer before bailing
 			 * out, though.  When a pending destroy did
 			 * suppress the queue, no later pass runs, and
-			 * rds_conn_path_destroy() is about to flush
-			 * cp_down_w and free the path: it must not
-			 * find cp_conn_w still armed.  A successor
+			 * rds_conn_path_quiesce() is about to flush
+			 * cp_down_w, ahead of the path's deferred
+			 * free: it must not find cp_conn_w still
+			 * armed.  A successor
 			 * pass, when there is one, re-arms the
 			 * reconnect from its own tail.
 			 */
@@ -515,12 +596,16 @@ void rds_conn_shutdown(struct rds_conn_path *cp)
 		conn->c_trans->conn_slots_available(conn, false);
 }
 
-/* destroy a single rds_conn_path. rds_conn_destroy() iterates over
- * all paths using rds_conn_path_destroy()
+/* quiesce a single rds_conn_path: shut it down and tear down any
+ * queued messages.  rds_conn_destroy() iterates over all paths using
+ * rds_conn_path_quiesce(); the transport state and the workqueue are
+ * freed later, from rds_conn_path_free().
  */
-static void rds_conn_path_destroy(struct rds_conn_path *cp)
+static void rds_conn_path_quiesce(struct rds_conn_path *cp)
 {
 	struct rds_message *rm, *rtmp;
+	unsigned long flags;
+	LIST_HEAD(purge);
 
 	if (!cp->cp_transport_data)
 		return;
@@ -532,10 +617,16 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)
 	rds_conn_path_drop(cp, true);
 	flush_work(&cp->cp_down_w);
 
-	/* tear down queued messages */
-	list_for_each_entry_safe(rm, rtmp,
-				 &cp->cp_send_queue,
-				 m_conn_item) {
+	/* Tear down queued messages.  Take the queue under cp_lock:
+	 * a sender that still holds a reference can be inside
+	 * rds_send_queue_rm() right now, and it tests
+	 * rds_destroy_pending() under the same lock, so after this
+	 * splice nothing is added behind our back.
+	 */
+	spin_lock_irqsave(&cp->cp_lock, flags);
+	list_splice_init(&cp->cp_send_queue, &purge);
+	spin_unlock_irqrestore(&cp->cp_lock, flags);
+	list_for_each_entry_safe(rm, rtmp, &purge, m_conn_item) {
 		list_del_init(&rm->m_conn_item);
 		BUG_ON(!list_empty(&rm->m_sock_item));
 		rds_message_put(rm);
@@ -547,6 +638,16 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)
 	WARN_ON(delayed_work_pending(&cp->cp_recv_w));
 	WARN_ON(delayed_work_pending(&cp->cp_conn_w));
 	WARN_ON(work_pending(&cp->cp_down_w));
+}
+
+/* free a quiesced rds_conn_path's transport state and workqueue; runs
+ * from rds_conn_destroy_fini() once the last connection reference is
+ * dropped.
+ */
+static void rds_conn_path_free(struct rds_conn_path *cp)
+{
+	if (!cp->cp_transport_data)
+		return;
 
 	if (cp->cp_wq != rds_wq) {
 		destroy_workqueue(cp->cp_wq);
@@ -556,17 +657,96 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)
 	cp->cp_conn->c_trans->conn_free(cp->cp_transport_data);
 }
 
+/* Free a connection.  This runs from rds_conn_put() when the last
+ * reference is dropped, after rds_conn_destroy() has quiesced the
+ * connection and dropped the initial reference.
+ */
+static void rds_conn_destroy_fini(struct kref *kref)
+{
+	struct rds_connection *conn = container_of(kref, struct rds_connection,
+						   c_refcount);
+	int npaths = (conn->c_trans->t_mp_capable ? RDS_MPATH_WORKERS : 1);
+	struct rds_transport *trans = conn->c_trans;
+	int i;
+
+	for (i = 0; i < npaths; i++)
+		rds_conn_path_free(&conn->c_path[i]);
+
+	kfree(conn->c_path);
+	kmem_cache_free(rds_conn_slab, conn);
+
+	/* only after everything the transport module owns has been
+	 * freed above may its unload proceed
+	 */
+	if (!atomic_dec_return(&trans->t_conn_count))
+		wake_up_all(&rds_conn_freed_waitq);
+}
+
+/* Wait for all of @trans's connections to be freed; the free runs
+ * asynchronously once rds_conn_destroy() has quiesced a connection.
+ * Called on transport module unload, after the transport has destroyed
+ * all of its connections.  A connection reference can be held for an
+ * application-controlled time - an unread datagram pins the inc that
+ * carries it, and thus the connection - so the wait is unbounded: the
+ * frees that run after unload call into this module's text (conn_free,
+ * inc_free) and free into its slabs, so proceeding while any remain
+ * would be a use-after-free, not a leak.  Warn periodically so a stuck
+ * count is diagnosable, but never stop waiting.  This matches the
+ * historical RDS contract that teardown does not discard queued data.
+ */
+void rds_conn_wait_conns_freed(struct rds_transport *trans,
+			       void (*resweep)(void))
+{
+	unsigned long warn_interval =
+			msecs_to_jiffies(RDS_CONN_FREE_WARN_INTERVAL_MS);
+	unsigned long warn_at = jiffies + warn_interval;
+
+	while (!wait_event_timeout(rds_conn_freed_waitq,
+				   !atomic_read(&trans->t_conn_count),
+				   msecs_to_jiffies(RDS_CONN_FREE_POLL_MS))) {
+		/* A transport whose teardown is asynchronous (IB moves a
+		 * connection off its device from the shutdown work) gives
+		 * us a resweep to destroy what has arrived since.
+		 */
+		if (resweep)
+			resweep();
+		if (time_after_eq(jiffies, warn_at)) {
+			pr_warn("RDS/%s: still waiting for %d connection(s) to be freed before unload\n",
+				trans->t_name,
+				atomic_read(&trans->t_conn_count));
+			warn_at = jiffies + warn_interval;
+		}
+	}
+}
+EXPORT_SYMBOL_GPL(rds_conn_wait_conns_freed);
+
+void rds_conn_get(struct rds_connection *conn)
+{
+	kref_get(&conn->c_refcount);
+}
+EXPORT_SYMBOL_GPL(rds_conn_get);
+
+void rds_conn_put(struct rds_connection *conn)
+{
+	kref_put(&conn->c_refcount, rds_conn_destroy_fini);
+}
+EXPORT_SYMBOL_GPL(rds_conn_put);
+
 /*
  * Stop and free a connection.
  *
- * This can only be used in very limited circumstances.  It assumes that once
- * the conn has been shutdown that no one else is referencing the connection.
- * We can only ensure this in the rmmod path in the current code.
+ * Quiesces the connection synchronously (workers cancelled, transport
+ * connections shut down, queued messages dropped) and drops the
+ * initial reference.  The memory - including the transport's
+ * per-connection state and the path workqueues - is freed once the
+ * last rds_conn_put() runs, which may be after this returns.
  */
 void rds_conn_destroy(struct rds_connection *conn)
 {
-	unsigned long flags;
 	int i;
+	struct rds_connection *passive, *parent;
+	struct hlist_head *head;
+	bool was_passive = false;
 	struct rds_conn_path *cp;
 	int npaths = (conn->c_trans->t_mp_capable ? RDS_MPATH_WORKERS : 1);
 
@@ -574,16 +754,67 @@ void rds_conn_destroy(struct rds_connection *conn)
 		 "%pI4\n", conn, &conn->c_laddr,
 		 &conn->c_faddr);
 
-	/* Ensure conn will not be scheduled for reconnect */
+	/* Make rds_destroy_pending() true for this conn.  Together with
+	 * the synchronize_rcu() below this stops the work-requeueing
+	 * sites (which all test rds_destroy_pending() under
+	 * rcu_read_lock()) from queueing new work on the path
+	 * workqueues once we start cancelling and destroying them.
+	 *
+	 * Now that the transport state stays discoverable (e.g. on the
+	 * transports' connection lists) until the final rds_conn_put(),
+	 * a conn can be handed to rds_conn_destroy() more than once -
+	 * e.g. dropped for a protocol version mismatch and then found
+	 * again at module unload.  Only the first caller proceeds; the
+	 * unhash also happens under rds_conn_lock, so a looked-up conn
+	 * can never be quiesced twice.
+	 */
 	spin_lock_irq(&rds_conn_lock);
+	if (conn->c_destroy_in_prog) {
+		spin_unlock_irq(&rds_conn_lock);
+		return;
+	}
+	WRITE_ONCE(conn->c_destroy_in_prog, true);
+
+	/* Ensure conn will not be scheduled for reconnect */
 	hlist_del_init_rcu(&conn->c_hash_node);
+
+	/* Snatch c_passive while holding the lock:
+	 * __rds_conn_create() dereferences it under rcu_read_lock()
+	 * (and refuses to install a new one once c_destroy_in_prog is
+	 * set, which it checks under this lock).  After the
+	 * synchronize_rcu() below no one can pick the pointer up any
+	 * more and its reference can be dropped.
+	 */
+	passive = rds_conn_passive_locked(conn);
+	RCU_INIT_POINTER(conn->c_passive, NULL);
+
+	/* If we are a parent's passive twin, invalidate its pointer to
+	 * us as well, so that __rds_conn_create() cannot hand out a
+	 * connection whose teardown has begun.  The parent is the
+	 * hashed connection for our key (a passive conn is never
+	 * hashed, and we unhashed ourselves above); it holds its
+	 * initial reference for as long as it is hashed, so the lookup
+	 * reference dropped below cannot be its last.
+	 */
+	head = rds_conn_bucket(&conn->c_laddr, &conn->c_faddr);
+	rcu_read_lock();
+	parent = rds_conn_lookup(rds_conn_net(conn), head, &conn->c_laddr,
+				 &conn->c_faddr, conn->c_trans, conn->c_tos,
+				 conn->c_dev_if);
+	rcu_read_unlock();
+	if (parent && rds_conn_passive_locked(parent) == conn) {
+		RCU_INIT_POINTER(parent->c_passive, NULL);
+		was_passive = true;
+	}
 	spin_unlock_irq(&rds_conn_lock);
+	if (parent)
+		rds_conn_put(parent);
 	synchronize_rcu();
 
 	/* shut the connection down */
 	for (i = 0; i < npaths; i++) {
 		cp = &conn->c_path[i];
-		rds_conn_path_destroy(cp);
+		rds_conn_path_quiesce(cp);
 		BUG_ON(!list_empty(&cp->cp_retrans));
 	}
 
@@ -594,12 +825,19 @@ void rds_conn_destroy(struct rds_connection *conn)
 	 */
 	rds_cong_remove_conn(conn);
 
-	kfree(conn->c_path);
-	kmem_cache_free(rds_conn_slab, conn);
+	/* drop the reference our c_passive pointer held, if any, and
+	 * the one a parent's c_passive pointer held on us; neither can
+	 * be the last, since the initial reference is dropped below
+	 */
+	if (passive)
+		rds_conn_put(passive);
+	if (was_passive)
+		rds_conn_put(conn);
 
-	spin_lock_irqsave(&rds_conn_lock, flags);
-	rds_conn_count--;
-	spin_unlock_irqrestore(&rds_conn_lock, flags);
+	/* drop the initial reference; the connection is freed from
+	 * rds_conn_destroy_fini() once every holder has dropped theirs
+	 */
+	rds_conn_put(conn);
 }
 EXPORT_SYMBOL_GPL(rds_conn_destroy);
 
diff --git a/net/rds/ib.c b/net/rds/ib.c
index 786f39169bc14..3fc2de9d19d55 100644
--- a/net/rds/ib.c
+++ b/net/rds/ib.c
@@ -525,10 +525,7 @@ static void rds_ib_set_unloading(void)
 
 static bool rds_ib_is_unloading(struct rds_connection *conn)
 {
-	struct rds_conn_path *cp = &conn->c_path[0];
-
-	return (test_bit(RDS_DESTROY_PENDING, &cp->cp_flags) ||
-		atomic_read(&rds_ib_unloading) != 0);
+	return atomic_read(&rds_ib_unloading) != 0;
 }
 
 void rds_ib_exit(void)
@@ -540,7 +537,24 @@ void rds_ib_exit(void)
 	rds_info_deregister_func(RDS6_INFO_IB_CONNECTIONS, rds6_ib_ic_info);
 #endif
 	rds_ib_unregister_client();
+
+	/* rds_ib_dev_shutdown() only dropped the connections still
+	 * attached to a device; each moves itself to ib_nodev_conns
+	 * from its shutdown work.  Destroy what is there now and keep
+	 * sweeping the list while the wait sees connections outstanding,
+	 * so a late arrival is destroyed rather than waited on forever.
+	 */
 	rds_ib_destroy_nodev_conns();
+	rds_conn_wait_conns_freed(&rds_ib_transport,
+				  rds_ib_destroy_nodev_conns);
+
+	/* Tearing down the last connection may have dropped the final
+	 * reference on a device, deferring rds_ib_dev_free() to rds_wq.
+	 * Drain it before the module goes away; it queues nothing
+	 * further on rds_wq.
+	 */
+	flush_workqueue(rds_wq);
+
 	rds_ib_sysctl_exit();
 	rds_ib_recv_exit();
 	rds_trans_unregister(&rds_ib_transport);
diff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c
index 4feb0edc360c8..257e03ed48618 100644
--- a/net/rds/ib_cm.c
+++ b/net/rds/ib_cm.c
@@ -874,6 +874,13 @@ int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id,
 	 * see the comment above rds_queue_reconnect()
 	 */
 	mutex_lock(&conn->c_cm_lock);
+	/* A destroy that has already quiesced this conn leaves it in
+	 * RDS_CONN_DOWN with no cm_id, exactly what the transition
+	 * below would happily claim; nothing would tear the new cm_id
+	 * and QP down again before the conn is freed.  Reject instead.
+	 */
+	if (rds_destroy_pending(conn))
+		goto out;
 	if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_CONNECTING)) {
 		if (rds_conn_state(conn) == RDS_CONN_UP) {
 			rdsdebug("incoming connect while connecting\n");
@@ -924,8 +931,14 @@ int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id,
 		rds_ib_conn_error(conn, "rdma_accept failed\n");
 
 out:
-	if (conn)
+	if (conn) {
 		mutex_unlock(&conn->c_cm_lock);
+		/* The conn stays reachable through cm_id->context
+		 * without a reference of its own: connection destroy
+		 * shuts the cm_id down before the conn is freed.
+		 */
+		rds_conn_put(conn);
+	}
 	if (err)
 		rdma_reject(cm_id, &err, sizeof(int),
 			    IB_CM_REJ_CONSUMER_DEFINED);
@@ -1271,6 +1284,7 @@ void rds_ib_conn_free(void *arg)
 {
 	struct rds_ib_connection *ic = arg;
 	spinlock_t	*lock_ptr;
+	unsigned long flags;
 
 	rdsdebug("ic %p\n", ic);
 
@@ -1278,12 +1292,18 @@ void rds_ib_conn_free(void *arg)
 	 * Conn is either on a dev's list or on the nodev list.
 	 * A race with shutdown() or connect() would cause problems
 	 * (since rds_ibdev would change) but that should never happen.
+	 *
+	 * Callers may hold rds_conn_lock with interrupts disabled
+	 * (__rds_conn_create() undoing a lost creation race), so do not
+	 * re-enable interrupts unconditionally here.
 	 */
 	lock_ptr = ic->rds_ibdev ? &ic->rds_ibdev->spinlock : &ib_nodev_conns_lock;
 
-	spin_lock_irq(lock_ptr);
-	list_del(&ic->ib_node);
-	spin_unlock_irq(lock_ptr);
+	spin_lock_irqsave(lock_ptr, flags);
+	/* already unlinked if a transport teardown gathered us first */
+	if (!list_empty(&ic->ib_node))
+		list_del(&ic->ib_node);
+	spin_unlock_irqrestore(lock_ptr, flags);
 
 	rds_ib_recv_free_caches(ic);
 
diff --git a/net/rds/ib_rdma.c b/net/rds/ib_rdma.c
index db7e92e7bd29f..91db43a0e7d73 100644
--- a/net/rds/ib_rdma.c
+++ b/net/rds/ib_rdma.c
@@ -165,11 +165,21 @@ void rds_ib_destroy_nodev_conns(void)
 
 	/* avoid calling conn_destroy with irqs off */
 	spin_lock_irq(&ib_nodev_conns_lock);
-	list_splice(&ib_nodev_conns, &tmp_list);
+	list_splice_init(&ib_nodev_conns, &tmp_list);
 	spin_unlock_irq(&ib_nodev_conns_lock);
 
-	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node)
+	/* rds_conn_destroy() can return before the connection is freed,
+	 * and it is the free - rds_ib_conn_free() - that unlinks ib_node.
+	 * tmp_list lives on this stack frame, so unlink each node before
+	 * its destroy; the free then finds it empty and leaves it alone.
+	 */
+	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node) {
+		spin_lock_irq(&ib_nodev_conns_lock);
+		list_del_init(&ic->ib_node);
+		spin_unlock_irq(&ib_nodev_conns_lock);
+
 		rds_conn_destroy(ic->conn);
+	}
 }
 
 void rds_ib_get_mr_info(struct rds_ib_device *rds_ibdev, struct rds_info_rdma_connection *iinfo)
diff --git a/net/rds/ib_recv.c b/net/rds/ib_recv.c
index bd6cb3ffaa571..7d45808544a0d 100644
--- a/net/rds/ib_recv.c
+++ b/net/rds/ib_recv.c
@@ -458,7 +458,11 @@ void rds_ib_recv_refill(struct rds_connection *conn, int prefill, gfp_t gfp)
 	    (must_wake ||
 	    (can_wait && rds_ib_ring_low(&ic->i_recv_ring)) ||
 	    rds_ib_ring_empty(&ic->i_recv_ring))) {
-		queue_delayed_work(conn->c_path->cp_wq, &conn->c_recv_w, 1);
+		rcu_read_lock();
+		if (!rds_destroy_pending(conn))
+			queue_delayed_work(conn->c_path->cp_wq,
+					   &conn->c_recv_w, 1);
+		rcu_read_unlock();
 	}
 	if (can_wait)
 		cond_resched();
diff --git a/net/rds/ib_send.c b/net/rds/ib_send.c
index d6be95542119f..bc411e96ad12d 100644
--- a/net/rds/ib_send.c
+++ b/net/rds/ib_send.c
@@ -298,8 +298,13 @@ void rds_ib_send_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc)
 	rds_ib_sub_signaled(ic, nr_sig);
 
 	if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags) ||
-	    test_bit(0, &conn->c_map_queued))
-		queue_delayed_work(conn->c_path->cp_wq, &conn->c_send_w, 0);
+	    test_bit(0, &conn->c_map_queued)) {
+		rcu_read_lock();
+		if (!rds_destroy_pending(conn))
+			queue_delayed_work(conn->c_path->cp_wq,
+					   &conn->c_send_w, 0);
+		rcu_read_unlock();
+	}
 
 	/* We expect errors as the qp is drained during shutdown */
 	if (wc->status != IB_WC_SUCCESS && rds_conn_up(conn)) {
@@ -420,8 +425,13 @@ void rds_ib_send_add_credits(struct rds_connection *conn, unsigned int credits)
 			test_bit(RDS_LL_SEND_FULL, &conn->c_flags) ? ", ll_send_full" : "");
 
 	atomic_add(IB_SET_SEND_CREDITS(credits), &ic->i_credits);
-	if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags))
-		queue_delayed_work(conn->c_path->cp_wq, &conn->c_send_w, 0);
+	if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags)) {
+		rcu_read_lock();
+		if (!rds_destroy_pending(conn))
+			queue_delayed_work(conn->c_path->cp_wq,
+					   &conn->c_send_w, 0);
+		rcu_read_unlock();
+	}
 
 	WARN_ON(IB_GET_SEND_CREDITS(credits) >= 16384);
 
diff --git a/net/rds/loop.c b/net/rds/loop.c
index e6b0750bbedab..42e6b841b42c6 100644
--- a/net/rds/loop.c
+++ b/net/rds/loop.c
@@ -156,6 +156,28 @@ static int rds_loop_conn_alloc(struct rds_connection *conn, gfp_t gfp)
 	return 0;
 }
 
+/* Destroy the connections whose nodes were gathered on @tmp_list.
+ *
+ * rds_conn_destroy() can return before the connection is freed, and
+ * it is the free - rds_loop_conn_free() - that unlinks loop_node.
+ * @tmp_list lives on the caller's stack, so unlink each node before
+ * its destroy; the free then finds it empty and leaves it alone.
+ */
+static void rds_loop_destroy_gathered_conns(struct list_head *tmp_list)
+{
+	struct rds_loop_connection *lc, *_lc;
+
+	list_for_each_entry_safe(lc, _lc, tmp_list, loop_node) {
+		WARN_ON(rcu_access_pointer(lc->conn->c_passive));
+
+		spin_lock_irq(&loop_conns_lock);
+		list_del_init(&lc->loop_node);
+		spin_unlock_irq(&loop_conns_lock);
+
+		rds_conn_destroy(lc->conn);
+	}
+}
+
 static void rds_loop_conn_free(void *arg)
 {
 	struct rds_loop_connection *lc = arg;
@@ -163,7 +185,9 @@ static void rds_loop_conn_free(void *arg)
 
 	rdsdebug("lc %p\n", lc);
 	spin_lock_irqsave(&loop_conns_lock, flags);
-	list_del(&lc->loop_node);
+	/* already unlinked if a transport teardown gathered us first */
+	if (!list_empty(&lc->loop_node))
+		list_del(&lc->loop_node);
 	spin_unlock_irqrestore(&loop_conns_lock, flags);
 	kfree(lc);
 }
@@ -180,7 +204,6 @@ static void rds_loop_conn_path_shutdown(struct rds_conn_path *cp)
 
 void rds_loop_exit(void)
 {
-	struct rds_loop_connection *lc, *_lc;
 	LIST_HEAD(tmp_list);
 
 	rds_loop_set_unloading();
@@ -191,10 +214,9 @@ void rds_loop_exit(void)
 	INIT_LIST_HEAD(&loop_conns);
 	spin_unlock_irq(&loop_conns_lock);
 
-	list_for_each_entry_safe(lc, _lc, &tmp_list, loop_node) {
-		WARN_ON(lc->conn->c_passive);
-		rds_conn_destroy(lc->conn);
-	}
+	rds_loop_destroy_gathered_conns(&tmp_list);
+
+	rds_conn_wait_conns_freed(&rds_loop_transport, NULL);
 }
 
 static void rds_loop_kill_conns(struct net *net)
@@ -212,10 +234,7 @@ static void rds_loop_kill_conns(struct net *net)
 	}
 	spin_unlock_irq(&loop_conns_lock);
 
-	list_for_each_entry_safe(lc, _lc, &tmp_list, loop_node) {
-		WARN_ON(lc->conn->c_passive);
-		rds_conn_destroy(lc->conn);
-	}
+	rds_loop_destroy_gathered_conns(&tmp_list);
 }
 
 static void __net_exit rds_loop_exit_net(struct net *net)
diff --git a/net/rds/message.c b/net/rds/message.c
index f25f2592586f7..29e95028e61e4 100644
--- a/net/rds/message.c
+++ b/net/rds/message.c
@@ -182,6 +182,18 @@ static void rds_message_purge(struct rds_message *rm)
 		kref_put(&rm->atomic.op_rdma_mr->r_kref, __rds_put_mr_final);
 }
 
+static void rds_message_free(struct rds_message *rm)
+{
+	/* get in rds_send_queue_rm(), rds_send_probe() or the congestion
+	 * map path of rds_send_xmit().  Messages that were never queued on
+	 * a connection have no reference to drop.
+	 */
+	if (rm->m_inc.i_conn)
+		rds_conn_put(rm->m_inc.i_conn);
+
+	kfree(rm);
+}
+
 static void rds_message_unpin_worker(struct work_struct *work)
 {
 	struct rds_message *rm = container_of(work, struct rds_message,
@@ -192,7 +204,7 @@ static void rds_message_unpin_worker(struct work_struct *work)
 	if (rm->atomic.op_unpin_deferred)
 		rds_atomic_op_unpin_page(&rm->atomic);
 
-	kfree(rm);
+	rds_message_free(rm);
 }
 
 void rds_message_put(struct rds_message *rm)
@@ -217,7 +229,7 @@ void rds_message_put(struct rds_message *rm)
 			return;
 		}
 
-		kfree(rm);
+		rds_message_free(rm);
 	}
 }
 EXPORT_SYMBOL_GPL(rds_message_put);
diff --git a/net/rds/rdma_transport.c b/net/rds/rdma_transport.c
index b15cf316b23a2..584e9867810f4 100644
--- a/net/rds/rdma_transport.c
+++ b/net/rds/rdma_transport.c
@@ -63,6 +63,18 @@ static int rds_rdma_cm_event_handler_cmn(struct rdma_cm_id *cm_id,
 	if (cm_id->device->node_type == RDMA_NODE_IB_CA)
 		trans = &rds_ib_transport;
 
+	/* cm_id->context carries no reference of its own.  Pin the
+	 * connection for the duration of the handler: what the callbacks
+	 * below do may drop the last reference other than ours, and the
+	 * mutex released at out: lives in the connection's path array.
+	 * A connection already being freed gets no events handled.
+	 */
+	if (conn && !rds_conn_get_unless_zero(conn)) {
+		rdsdebug("conn %p id %p is being freed, ignoring event\n",
+			 conn, cm_id);
+		return 0;
+	}
+
 	/* Prevent shutdown from tearing down the connection
 	 * while we're executing. */
 	if (conn) {
@@ -171,8 +183,10 @@ static int rds_rdma_cm_event_handler_cmn(struct rdma_cm_id *cm_id,
 	}
 
 out:
-	if (conn)
+	if (conn) {
 		mutex_unlock(&conn->c_cm_lock);
+		rds_conn_put(conn);
+	}
 
 	rdsdebug("id %p event %u (%s) handling ret %d\n", cm_id, event->event,
 		 rdma_event_msg(event->event), ret);
diff --git a/net/rds/rds.h b/net/rds/rds.h
index 2db49573dacd5..06d48c2821efd 100644
--- a/net/rds/rds.h
+++ b/net/rds/rds.h
@@ -89,7 +89,6 @@ enum {
 #define RDS_RECONNECT_PENDING	1
 #define RDS_IN_XMIT		2
 #define RDS_RECV_REFILL		3
-#define	RDS_DESTROY_PENDING	4
 
 /* Max number of multipaths per RDS connection. Must be a power of 2 */
 #define	RDS_MPATH_WORKERS	8
@@ -138,6 +137,12 @@ struct rds_conn_path {
 /* One rds_connection per RDS address pair */
 struct rds_connection {
 	struct hlist_node	c_hash_node;
+	/* Free of the connection memory (not the teardown of its
+	 * transport state - that stays synchronous in
+	 * rds_conn_destroy()) is deferred until the last reference is
+	 * dropped via rds_conn_put().
+	 */
+	struct kref		c_refcount;
 	struct in6_addr		c_laddr;
 	struct in6_addr		c_faddr;
 	int			c_dev_if; /* ifindex used for this conn */
@@ -148,7 +153,15 @@ struct rds_connection {
 				c_pad_to_32:29;
 	int			c_npaths;
 	bool			c_with_sport_idx;
-	struct rds_connection	*c_passive;
+	/* Set once, by rds_conn_destroy(), before it cancels the path
+	 * works; read through rds_destroy_pending().  A site that arms
+	 * a path work must test the predicate and queue the work inside
+	 * one rcu_read_lock() section: the synchronize_rcu() that
+	 * follows the store is what keeps a queue issued after the
+	 * cancellation from landing on a destroyed workqueue.
+	 */
+	bool			c_destroy_in_prog;
+	struct rds_connection __rcu *c_passive;
 	struct rds_transport	*c_trans;
 
 	struct rds_cong_map	*c_lcong;
@@ -544,6 +557,12 @@ struct rds_transport {
 	unsigned int		t_prefer_loopback:1,
 				t_mp_capable:1;
 	unsigned int		t_type;
+	/* Connections of this transport not yet freed; freeing runs
+	 * asynchronously once rds_conn_destroy() has quiesced a
+	 * connection, so transport module unload has to wait for this
+	 * to reach zero (rds_conn_wait_conns_freed()).
+	 */
+	atomic_t		t_conn_count;
 
 	int (*laddr_check)(struct net *net, const struct in6_addr *addr,
 			   __u32 scope_id);
@@ -819,6 +838,20 @@ struct rds_connection *rds_conn_create_outgoing(struct net *net,
 						u8 tos, gfp_t gfp, int dev_if);
 void rds_conn_shutdown(struct rds_conn_path *cpath);
 void rds_conn_destroy(struct rds_connection *conn);
+void rds_conn_get(struct rds_connection *conn);
+void rds_conn_put(struct rds_connection *conn);
+/* take a reference unless the connection is already being freed */
+static inline bool rds_conn_get_unless_zero(struct rds_connection *conn)
+{
+	return kref_get_unless_zero(&conn->c_refcount);
+}
+/* transport unload waits for its connections to be freed, polling at
+ * the first interval and warning at the second
+ */
+#define RDS_CONN_FREE_POLL_MS		100
+#define RDS_CONN_FREE_WARN_INTERVAL_MS	10000
+void rds_conn_wait_conns_freed(struct rds_transport *trans,
+			       void (*resweep)(void));
 void rds_conn_drop(struct rds_connection *conn);
 void rds_conn_path_drop(struct rds_conn_path *cpath, bool destroy);
 void rds_conn_connect_if_down(struct rds_connection *conn);
@@ -994,7 +1027,8 @@ void __rds_put_mr_final(struct kref *kref);
 
 static inline bool rds_destroy_pending(struct rds_connection *conn)
 {
-	return !check_net(rds_conn_net(conn)) ||
+	return READ_ONCE(conn->c_destroy_in_prog) ||
+	       !check_net(rds_conn_net(conn)) ||
 	       (conn->c_trans->t_unloading && conn->c_trans->t_unloading(conn));
 }
 
diff --git a/net/rds/recv.c b/net/rds/recv.c
index 6204e577a90ae..b031c0b43af83 100644
--- a/net/rds/recv.c
+++ b/net/rds/recv.c
@@ -46,6 +46,7 @@ void rds_inc_init(struct rds_incoming *inc, struct rds_connection *conn,
 {
 	refcount_set(&inc->i_refcount, 1);
 	INIT_LIST_HEAD(&inc->i_item);
+	rds_conn_get(conn);	/* put in rds_inc_put() */
 	inc->i_conn = conn;
 	inc->i_conn_path = NULL;
 	inc->i_saddr = *saddr;
@@ -61,6 +62,7 @@ void rds_inc_path_init(struct rds_incoming *inc, struct rds_conn_path *cp,
 {
 	refcount_set(&inc->i_refcount, 1);
 	INIT_LIST_HEAD(&inc->i_item);
+	rds_conn_get(cp->cp_conn);	/* put in rds_inc_put() */
 	inc->i_conn = cp->cp_conn;
 	inc->i_conn_path = cp;
 	inc->i_saddr = *saddr;
@@ -81,9 +83,19 @@ void rds_inc_put(struct rds_incoming *inc)
 {
 	rdsdebug("put inc %p ref %d\n", inc, refcount_read(&inc->i_refcount));
 	if (refcount_dec_and_test(&inc->i_refcount)) {
+		struct rds_connection *conn = inc->i_conn;
+
 		BUG_ON(!list_empty(&inc->i_item));
 
-		inc->i_conn->c_trans->inc_free(inc);
+		/* inc_free() can free the memory @inc lives in, so the
+		 * connection reference has to be dropped through the
+		 * copy taken above.
+		 */
+		conn->c_trans->inc_free(inc);
+		/* get in rds_inc_init(), rds_inc_path_init() or
+		 * rds_recv_incoming()
+		 */
+		rds_conn_put(conn);
 	}
 }
 EXPORT_SYMBOL_GPL(rds_inc_put);
@@ -325,6 +337,13 @@ void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,
 	unsigned long flags;
 	struct rds_conn_path *cp;
 
+	/* every caller initialized @inc with rds_inc_init() or
+	 * rds_inc_path_init() first, so i_conn already holds a reference.
+	 * Take the new one before dropping the old, so that re-pointing an
+	 * inc at the connection it already refers to cannot free it.
+	 */
+	rds_conn_get(conn);
+	rds_conn_put(inc->i_conn);
 	inc->i_conn = conn;
 	inc->i_rx_jiffies = jiffies;
 	if (conn->c_trans->t_mp_capable)
diff --git a/net/rds/send.c b/net/rds/send.c
index 1afa981e5c06d..f59a487b079d5 100644
--- a/net/rds/send.c
+++ b/net/rds/send.c
@@ -290,6 +290,8 @@ int rds_send_xmit(struct rds_conn_path *cp)
 			}
 			rm->data.op_active = 1;
 			rm->m_inc.i_conn_path = cp;
+			/* put in rds_message_put() */
+			rds_conn_get(cp->cp_conn);
 			rm->m_inc.i_conn = cp->cp_conn;
 
 			cp->cp_xmit_rm = rm;
@@ -928,6 +930,19 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
 	 * and poll() now knows no more data can be sent.
 	 */
 	if (rs->rs_snd_bytes < rds_sk_sndbuf(rs)) {
+		/* rds_conn_path_quiesce() empties cp_send_queue under
+		 * cp_lock once the connection's destroy has begun.  Test
+		 * for that under the same lock, before touching either
+		 * queue: a message added after the purge would hold a
+		 * connection reference nothing ever drops.
+		 */
+		spin_lock(&cp->cp_lock);
+		if (rds_destroy_pending(conn)) {
+			spin_unlock(&cp->cp_lock);
+			*queued = -EAGAIN;
+			goto unlock;
+		}
+
 		rs->rs_snd_bytes += len;
 
 		/* let recv side know we are close to send space exhaustion.
@@ -947,11 +962,11 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
 		/* The code ordering is a little weird, but we're
 		   trying to minimize the time we hold c_lock */
 		rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport, 0);
+		rds_conn_get(conn);	/* put in rds_message_put() */
 		rm->m_inc.i_conn = conn;
 		rm->m_inc.i_conn_path = cp;
 		rds_message_addref(rm);
 
-		spin_lock(&cp->cp_lock);
 		rm->m_inc.i_hdr.h_sequence = cpu_to_be64(cp->cp_next_tx_seq++);
 		list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
 		set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
@@ -964,6 +979,7 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
 		*queued = 1;
 	}
 
+unlock:
 	spin_unlock_irqrestore(&rs->rs_lock, flags);
 out:
 	return *queued;
@@ -1159,13 +1175,14 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	DECLARE_SOCKADDR(struct sockaddr_in *, usin, msg->msg_name);
 	__be16 dport;
 	struct rds_message *rm = NULL;
-	struct rds_connection *conn;
+	struct rds_connection *conn = NULL;
 	int ret = 0;
 	int queued = 0, allocated_mr = 0;
 	int nonblock = msg->msg_flags & MSG_DONTWAIT;
 	long timeo = sock_sndtimeo(sk, nonblock);
 	struct rds_conn_path *cpath;
 	struct in6_addr daddr;
+	unsigned long flags;
 	__u32 scope_id = 0;
 	size_t rdma_payload_len = 0;
 	bool zcopy = ((msg->msg_flags & MSG_ZEROCOPY) &&
@@ -1340,11 +1357,29 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	rm->m_daddr = daddr;
 
 	/* rds_conn_create has a spinlock that runs with IRQ off.
-	 * Caching the conn in the socket helps a lot. */
-	if (rs->rs_conn && ipv6_addr_equal(&rs->rs_conn->c_faddr, &daddr) &&
-	    rs->rs_tos == rs->rs_conn->c_tos) {
-		conn = rs->rs_conn;
+	 * Caching the conn in the socket helps a lot.
+	 *
+	 * The cached rs_conn holds a connection reference; take one of
+	 * our own for the duration of this call (dropped on both exit
+	 * paths), so that neither a concurrent sender replacing the
+	 * cache nor rds_conn_destroy() can free the connection under
+	 * us.  A cached connection whose destruction has begun is not
+	 * reused: dropping it here lets the next sendmsg look up or
+	 * create a live one instead of returning -EAGAIN forever.
+	 */
+	spin_lock_irqsave(&rs->rs_lock, flags);
+	conn = rs->rs_conn;
+	if (conn && ipv6_addr_equal(&conn->c_faddr, &daddr) &&
+	    rs->rs_tos == conn->c_tos && !rds_destroy_pending(conn)) {
+		rds_conn_get(conn);
 	} else {
+		conn = NULL;
+	}
+	spin_unlock_irqrestore(&rs->rs_lock, flags);
+
+	if (!conn) {
+		struct rds_connection *old;
+
 		conn = rds_conn_create_outgoing(sock_net(sock->sk),
 						&rs->rs_bound_addr, &daddr,
 						rs->rs_transport, rs->rs_tos,
@@ -1352,9 +1387,17 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 						scope_id);
 		if (IS_ERR(conn)) {
 			ret = PTR_ERR(conn);
+			conn = NULL;
 			goto out;
 		}
+		/* hand the cache its own reference */
+		rds_conn_get(conn);
+		spin_lock_irqsave(&rs->rs_lock, flags);
+		old = rs->rs_conn;
 		rs->rs_conn = conn;
+		spin_unlock_irqrestore(&rs->rs_lock, flags);
+		if (old)
+			rds_conn_put(old);
 	}
 
 	if (conn->c_trans->t_mp_capable) {
@@ -1378,9 +1421,14 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 		 * outstanding.
 		 */
 		if (!test_and_set_bit(RDS_RECONNECT_PENDING,
-				      &conn->c_path[0].cp_flags))
-			queue_delayed_work(conn->c_path[0].cp_wq,
-					   &conn->c_path[0].cp_conn_w, 0);
+				      &conn->c_path[0].cp_flags)) {
+			rcu_read_lock();
+			if (!rds_destroy_pending(conn))
+				queue_delayed_work(conn->c_path[0].cp_wq,
+						   &conn->c_path[0].cp_conn_w,
+						   0);
+			rcu_read_unlock();
+		}
 		rds_send_ping(conn, 0);
 	}
 
@@ -1442,6 +1490,11 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 			ret = -ETIMEDOUT;
 		goto out;
 	}
+	/* rds_send_queue_rm() refused: the connection is being destroyed */
+	if (queued < 0) {
+		ret = queued;
+		goto out;
+	}
 
 	/*
 	 * By now we've committed to the send.  We reuse rds_send_worker()
@@ -1469,6 +1522,8 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 		kfree(vct.vec[ind].iov);
 	kfree(vct.vec);
 
+	rds_conn_put(conn);
+
 	return payload_len;
 
 out:
@@ -1476,6 +1531,9 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 		kfree(vct.vec[ind].iov);
 	kfree(vct.vec);
 
+	if (conn)
+		rds_conn_put(conn);
+
 	/* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
 	 * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
 	 * or in any other way, we need to destroy the MR again */
@@ -1522,6 +1580,7 @@ rds_send_probe(struct rds_conn_path *cp, __be16 sport,
 	list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
 	set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
 	rds_message_addref(rm);
+	rds_conn_get(cp->cp_conn);	/* put in rds_message_put() */
 	rm->m_inc.i_conn = cp->cp_conn;
 	rm->m_inc.i_conn_path = cp;
 
diff --git a/net/rds/tcp.c b/net/rds/tcp.c
index 774a71f88d375..a71d6a4f0939c 100644
--- a/net/rds/tcp.c
+++ b/net/rds/tcp.c
@@ -502,6 +502,28 @@ static bool rds_tcp_is_unloading(struct rds_connection *conn)
 	return atomic_read(&rds_tcp_unloading) != 0;
 }
 
+/* Destroy the connections whose nodes were gathered on @tmp_list.
+ *
+ * rds_conn_destroy() can return before the connection is freed, and
+ * it is the free - rds_tcp_conn_free() - that unlinks t_tcp_node.
+ * Since @tmp_list lives on the caller's stack, unlink each node here
+ * and mark it detached before its destroy, so that a free that runs
+ * after the caller has returned does not write into a dead frame.
+ */
+static void rds_tcp_destroy_gathered_conns(struct list_head *tmp_list)
+{
+	struct rds_tcp_connection *tc, *_tc;
+
+	list_for_each_entry_safe(tc, _tc, tmp_list, t_tcp_node) {
+		spin_lock_irq(&rds_tcp_conn_lock);
+		list_del_init(&tc->t_tcp_node);
+		tc->t_tcp_node_detached = true;
+		spin_unlock_irq(&rds_tcp_conn_lock);
+
+		rds_conn_destroy(tc->t_cpath->cp_conn);
+	}
+}
+
 static void rds_tcp_destroy_conns(void)
 {
 	struct rds_tcp_connection *tc, *_tc;
@@ -515,8 +537,7 @@ static void rds_tcp_destroy_conns(void)
 	}
 	spin_unlock_irq(&rds_tcp_conn_lock);
 
-	list_for_each_entry_safe(tc, _tc, &tmp_list, t_tcp_node)
-		rds_conn_destroy(tc->t_cpath->cp_conn);
+	rds_tcp_destroy_gathered_conns(&tmp_list);
 }
 
 static void rds_tcp_exit(void);
@@ -698,8 +719,7 @@ static void rds_tcp_kill_sock(struct net *net)
 		}
 	}
 	spin_unlock_irq(&rds_tcp_conn_lock);
-	list_for_each_entry_safe(tc, _tc, &tmp_list, t_tcp_node)
-		rds_conn_destroy(tc->t_cpath->cp_conn);
+	rds_tcp_destroy_gathered_conns(&tmp_list);
 }
 
 static void __net_exit rds_tcp_exit_net(struct net *net)
@@ -805,6 +825,7 @@ static void rds_tcp_exit(void)
 #endif
 	unregister_pernet_device(&rds_tcp_net_ops);
 	rds_tcp_destroy_conns();
+	rds_conn_wait_conns_freed(&rds_tcp_transport, NULL);
 	rds_trans_unregister(&rds_tcp_transport);
 	rds_tcp_recv_exit();
 	kmem_cache_destroy(rds_tcp_conn_slab);
diff --git a/net/rds/tcp_listen.c b/net/rds/tcp_listen.c
index 13fa60c1985bb..e22ea9ca8c1ce 100644
--- a/net/rds/tcp_listen.c
+++ b/net/rds/tcp_listen.c
@@ -153,7 +153,7 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 {
 	struct socket *listen_sock = rtn->rds_tcp_listen_sock;
 	struct socket *new_sock = NULL;
-	struct rds_connection *conn;
+	struct rds_connection *conn = NULL;
 	int ret;
 	struct inet_sock *inet;
 	struct rds_tcp_connection *rs_tcp = NULL;
@@ -229,6 +229,7 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 
 	if (IS_ERR(conn)) {
 		ret = PTR_ERR(conn);
+		conn = NULL;
 		goto out;
 	}
 	/* An incoming SYN request came in, and TCP just accepted it.
@@ -277,7 +278,15 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 	cp = rs_tcp->t_cpath;
 	conn_state = rds_conn_path_state(cp);
 	WARN_ON(conn_state == RDS_CONN_UP);
-	if (conn_state != RDS_CONN_CONNECTING && conn_state != RDS_CONN_ERROR) {
+	/* A connection whose destroy has begun has been quiesced and is
+	 * only waiting for its last reference: its paths sit in
+	 * RDS_CONN_DOWN, which rds_tcp_accept_one_path() happily claims.
+	 * Installing a socket on it would leave sk_user_data pointing
+	 * at a path that is about to be freed.
+	 */
+	if (rds_destroy_pending(conn) ||
+	    (conn_state != RDS_CONN_CONNECTING &&
+	     conn_state != RDS_CONN_ERROR)) {
 		rds_conn_path_drop(cp, 0);
 		goto rst_nsk;
 	}
@@ -316,10 +325,14 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 	 */
 	if (READ_ONCE(sk->sk_state) == TCP_CLOSE_WAIT ||
 	    READ_ONCE(sk->sk_state) == TCP_LAST_ACK ||
-	    READ_ONCE(sk->sk_state) == TCP_CLOSE)
+	    READ_ONCE(sk->sk_state) == TCP_CLOSE) {
 		rds_conn_path_drop(cp, 0);
-	else
-		queue_delayed_work(cp->cp_wq, &cp->cp_recv_w, 0);
+	} else {
+		rcu_read_lock();
+		if (!rds_destroy_pending(cp->cp_conn))
+			queue_delayed_work(cp->cp_wq, &cp->cp_recv_w, 0);
+		rcu_read_unlock();
+	}
 
 	sock_put(sk);
 
@@ -343,6 +356,8 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 		mutex_unlock(&rs_tcp->t_conn_path_lock);
 	if (new_sock)
 		sock_release(new_sock);
+	if (conn)
+		rds_conn_put(conn);
 
 	mutex_unlock(&rtn->rds_tcp_accept_lock);
 

3/2 2026/09/14 04:28 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 21400
							output: 27
							thoughts: 408
						

Thoughts:
**RDS Patch Analysis: A Fuzzing Perspective**

Okay, so this patch touches `net/rds`, which immediately flags my attention. RDS, Reliable Datagram Sockets, is a pretty core network protocol family (AF_RDS/PF_RDS), and I see the changes span across a lot of its core components: `af_rds.c`, `connection.c`, the various InfiniBand files (`ib.c`, `ib_cm.c`, etc.), the TCP implementation (`tcp.c`, `tcp_listen.c`), loopback (`loop.c`), and common utilities (`message.c`, `rdma_transport.c`, `rds.h`, `recv.c`, `send.c`).

My initial thought is, "Is this reachable and therefore relevant to fuzzing?". Yes, absolutely. RDS is exposed through standard socket syscalls, it has loopback and TCP transport, and Syzkaller loves to target these. Syzkaller, in particular, frequently fuzzes RDS sockets, and the `setsockopt`, `ioctl`, `sendmsg`, and `recvmsg` paths.

The patch itself refactors and fixes connection lifetime and refcounting within RDS. Specifically, I see mentions of adding a `c_refcount`, functions like `rds_conn_get` and `rds_conn_put`, and flags like `c_destroy_in_prog`. There are also locking fixes and changes in the workqueue interactions across the RDS code. This is significant because any modification in refcounting, synchronization, or error handling has the potential to introduce subtle bugs that fuzzing can uncover. This definitely warrants further analysis from a fuzzing perspective, so I'm putting it down as `WorthFuzzing = true`.

Now, I'm going to carefully go through the Kconfig requirements and the symbols. I'm going to start with `net/rds/Kconfig` to make sure there are no dependency issues I might have missed. Gotta be thorough.



4/2 2026/09/14 04:28 tool grepper
0m Args:
{
  "Expression": "RDS",
  "PathPrefix": "net/rds/Kconfig"
}

Results:
{
  "Output": "net/rds/Kconfig-2-\nnet/rds/Kconfig:3:config RDS\nnet/rds/Kconfig-4-\ttristate \"The Reliable Datagram Sockets Protocol\"\n--\nnet/rds/Kconfig-6-\thelp\nnet/rds/Kconfig:7:\t  The RDS (Reliable Datagram Sockets) protocol provides reliable,\nnet/rds/Kconfig-8-\t  sequenced delivery of datagrams over Infiniband or TCP.\nnet/rds/Kconfig-9-\nnet/rds/Kconfig:10:config RDS_RDMA\nnet/rds/Kconfig:11:\ttristate \"RDS over Infiniband\"\nnet/rds/Kconfig:12:\tdepends on RDS \u0026\u0026 INFINIBAND \u0026\u0026 INFINIBAND_ADDR_TRANS\nnet/rds/Kconfig-13-\thelp\nnet/rds/Kconfig:14:\t  Allow RDS to use Infiniband as a transport.\nnet/rds/Kconfig-15-\t  This transport supports RDMA operations.\nnet/rds/Kconfig-16-\nnet/rds/Kconfig:17:config RDS_TCP\nnet/rds/Kconfig:18:\ttristate \"RDS over TCP\"\nnet/rds/Kconfig:19:\tdepends on RDS\nnet/rds/Kconfig-20-\tdepends on IPV6 || !IPV6\nnet/rds/Kconfig-21-\thelp\nnet/rds/Kconfig:22:\t  Allow RDS to use TCP as a transport.\nnet/rds/Kconfig-23-\t  This transport does not support RDMA operations.\nnet/rds/Kconfig-24-\nnet/rds/Kconfig:25:config RDS_DEBUG\nnet/rds/Kconfig:26:\tbool \"RDS debugging messages\"\nnet/rds/Kconfig:27:\tdepends on RDS\nnet/rds/Kconfig-28-\tdefault n\nnet/rds/Kconfig-29-\nnet/rds/Kconfig:30:config GCOV_PROFILE_RDS\nnet/rds/Kconfig:31:\tbool \"Enable GCOV profiling on RDS\"\nnet/rds/Kconfig-32-\tdepends on GCOV_KERNEL\nnet/rds/Kconfig-33-\thelp\nnet/rds/Kconfig:34:\t  Enable GCOV profiling on RDS for checking which functions/lines\nnet/rds/Kconfig-35-\t  are executed.\n"
}

5/2 2026/09/14 04:28 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 22844
							output: 31
							thoughts: 19
						

Thoughts:
Here's my attempt at summarizing the thought process, tailored for an expert audience:

**Investigating the Build Process for `net/rds`**

Okay, so I need to understand how the files within `net/rds` are being compiled. Let's start by digging into the `Makefile` located in that directory. That's the logical starting point – the `Makefile` is where all the build rules and dependencies should be defined. From there, I should be able to trace the compiler flags, object files, and ultimately, the linking process. This should quickly give me a clear picture of what's happening under the hood with this particular component. Knowing the compilation steps is key to understanding the overall build strategy and where I might want to modify things.



6/2 2026/09/14 04:28 tool read-file
0m Args:
{
  "File": "net/rds/Makefile",
  "FirstLine": 1,
  "LineCount": 20
}

Results:
{
  "Contents": "   1:\t# SPDX-License-Identifier: GPL-2.0\n   2:\tobj-$(CONFIG_RDS) += rds.o\n   3:\trds-y :=\taf_rds.o bind.o cong.o connection.o info.o message.o   \\\n   4:\t\t\t\trecv.o send.o stats.o sysctl.o threads.o transport.o \\\n   5:\t\t\t\tloop.o page.o rdma.o\n   6:\t\n   7:\tobj-$(CONFIG_RDS_RDMA) += rds_rdma.o\n   8:\trds_rdma-y :=\trdma_transport.o \\\n   9:\t\t\t\tib.o ib_cm.o ib_recv.o ib_ring.o ib_send.o ib_stats.o \\\n  10:\t\t\t\tib_sysctl.o ib_rdma.o ib_frmr.o\n  11:\t\n  12:\t\n  13:\tobj-$(CONFIG_RDS_TCP) += rds_tcp.o\n  14:\trds_tcp-y :=\t\ttcp.o tcp_connect.o tcp_listen.o tcp_recv.o \\\n  15:\t\t\t\ttcp_send.o tcp_stats.o\n  16:\t\n  17:\tccflags-$(CONFIG_RDS_DEBUG)\t:=\t-DRDS_DEBUG\n  18:\t\n  19:\t# for GCOV coverage profiling\n  20:\tifdef CONFIG_GCOV_PROFILE_RDS\n"
}

7/2 2026/09/14 04:28 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 23424
							output: 25
							thoughts: 196
						

Thoughts:
**My Configuration Quest**

Okay, so I'm digging into these `RDS` configurations – `RDS`, `RDS_TCP`, and `RDS_RDMA`. Specifically, it's about the interaction of `RDS`, `RDS_TCP` and virtualized environments like QEMU. That's a good starting point. I wonder if `RDS_RDMA` is even possible in a QEMU setup. Soft-RoCE or SIW could potentially give us InfiniBand in a virtualized setting, but `RDS_RDMA` relies on `INFINIBAND` and `INFINIBAND_ADDR_TRANS`, which is the bottleneck here.

Let's start with a systematic approach. First, I'll examine the dependencies for `RDS`, `RDS_TCP` and `RDS_RDMA`. I need to figure out what other configurations are *required*.

Alright, let's start with `RDS` itself. No immediate dependencies jump out. Looking into `net/rds/Kconfig` now...
```
config RDS
	tristate "The Reliable Datagram Sockets Protocol"
```
Nothing explicit under `depends on` for `RDS`, at least within this file. It looks like it inherits dependencies from its parent, within `net/Kconfig`. Time to investigate `net/Kconfig` and see what's being pulled in from there.



8/2 2026/09/14 04:28 tool grepper
0m Error:
no matches

Args:
{
  "Expression": "RDS",
  "PathPrefix": "net/Kconfig"
}

Results:
{
  "error": "no matches"
}

9/2 2026/09/14 04:28 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 23951
							output: 25
							thoughts: 11
						

Thoughts:
Here's my attempt at summarizing the thought process, tailored for an expert audience:

**Tracing the Origins of `net/rds`**

Okay, so I need to figure out the source of `net/rds`. It's a key piece of infrastructure, and understanding its origins is crucial for any kind of performance analysis, debugging, or even just general system understanding. Let's see...I'm thinking about the most efficient path to this. Perhaps there's an obvious header file, build flag, or configuration setting that directly points to the source code location? Maybe a symbol table lookup on the running system would be a good starting point to identify the library that is being used? Or even a `grep` through the system configuration files for a relevant setting or dependency? I should probably start with a quick check of the standard documentation for the system. Let's start with a systematic approach.



10/2 2026/09/14 04:28 tool grepper
0m Args:
{
  "Expression": "net/rds",
  "PathPrefix": "net/"
}

Results:
{
  "Output": "net/Kconfig=254=source \"net/sctp/Kconfig\"\nnet/Kconfig:255:source \"net/rds/Kconfig\"\nnet/Kconfig-256-source \"net/tipc/Kconfig\"\n--\nnet/rds/ib_sysctl.c=114=int rds_ib_sysctl_init(void)\nnet/rds/ib_sysctl.c-115-{\nnet/rds/ib_sysctl.c:116:\trds_ib_sysctl_hdr = register_net_sysctl(\u0026init_net, \"net/rds/ib\", rds_ib_sysctl_table);\nnet/rds/ib_sysctl.c-117-\tif (!rds_ib_sysctl_hdr)\n--\nnet/rds/sysctl.c=99=int rds_sysctl_init(void)\n--\nnet/rds/sysctl.c-104-\trds_sysctl_reg_table =\nnet/rds/sysctl.c:105:\t\tregister_net_sysctl(\u0026init_net, \"net/rds\", rds_sysctl_rds_table);\nnet/rds/sysctl.c-106-\tif (!rds_sysctl_reg_table)\n--\nnet/rds/tcp.c=633=static __net_init int rds_tcp_init_net(struct net *net)\n--\nnet/rds/tcp.c-658-\ttbl[RDS_TCP_RCVBUF].data = \u0026rtn-\u003ercvbuf_size;\nnet/rds/tcp.c:659:\trtn-\u003erds_tcp_sysctl = register_net_sysctl_sz(net, \"net/rds/tcp\", tbl,\nnet/rds/tcp.c-660-\t\t\t\t\t\t     ARRAY_SIZE(rds_tcp_sysctl_table));\n"
}

11/2 2026/09/14 04:28 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 24638
							output: 30
							thoughts: 2028
						

Thoughts:
**RDS Refcounting and Connection Management: A Deep Dive**

Okay, here's what I've got after going through these changes. First, for configuration, we are looking at `CONFIG_RDS`, `CONFIG_RDS_TCP`, and `CONFIG_RDS_RDMA`. To be thorough, if we're enabling `RDS_RDMA`, then we *must* also include `INFINIBAND` and `INFINIBAND_ADDR_TRANS` because of the dependencies. So, the key configs for fuzzing or testing are `RDS`, `RDS_TCP`, `RDS_RDMA`, `INFINIBAND`, and `INFINIBAND_ADDR_TRANS`.  Note, even in standard virtualized environments, RDS is fully reachable with `CONFIG_RDS` and `CONFIG_RDS_TCP`.

Now, regarding the code modifications, the focus here is on *specific* feature logic and entry points, avoiding ubiquitous functions.  Here's a breakdown of the *key* changes, focusing on refcounting, connection lifecycle, and transport layer:

*   **`rds_conn_destroy`**: This function is central. It's heavily modified, indicating significant changes to connection teardown. This looks to be the place where we manage the "destroy in progress" flag, and release the passive references.
*   **`rds_conn_put`**: A *new* exported function. This is critical because it's the counterpart to `rds_conn_get` for releasing references.  We need to pay very close attention to how this is used and where.
*   **`rds_conn_get`**: Also a *new* exported function.  This indicates an addition in how connections are acquired. We'll need to understand all the call sites to ensure correct refcount management.
*   **`rds_conn_destroy_fini`**: This is a key function because, while it is static, it's passed as a function pointer to `kref_put`.  It's the callback for dropping the kref, where we clean up the connection path and free memory via `kmem_cache_free`. This is where the actual memory release happens.
*   **`rds_sendmsg`**: This function is modified to deal with connection caching and refcounting under the `rs_lock`.
*   **`rds_release`**: Modified to drop the reference to `rs->rs_conn` with `rds_conn_put`.
*   **`rds_ioctl`**: This sees a locking change.
*   **`rds_send_queue_rm`**: This is an important function.  It checks for `rds_destroy_pending` under `cp_lock` and also handles refcount manipulation.
*   **`rds_tcp_accept_one`**: Another crucial function, this is the TCP accept handler, which is now checking `rds_destroy_pending` and also handling refcounts.
*   **`rds_inc_put`**: This function is used to drop the connection refcount in the incoming path.
*   **`rds_recv_incoming`**: Further demonstrates refcount handling within the receive path.
*   **`rds_message_put`**: This calls `rds_message_free`, which in turn drops the connection refcount.
*   **`rds_conn_wait_conns_freed`**: This is a *new* exported function that is critical for waiting for transport connections to be freed. This is likely very important for ensuring proper shutdown and resource cleanup.



12/2 2026/09/14 04:29 tool grepper
0m Args:
{
  "Expression": "__rds_conn_create",
  "PathPrefix": "net/rds/"
}

Results:
{
  "Output": "net/rds/connection.c=190=static void rds_conn_free_transport_data(struct rds_connection *conn,\n--\nnet/rds/connection.c-210- */\nnet/rds/connection.c:211:static struct rds_connection *__rds_conn_create(struct net *net,\nnet/rds/connection.c-212-\t\t\t\t\t\tconst struct in6_addr *laddr,\n--\nnet/rds/connection.c=452=struct rds_connection *rds_conn_create(struct net *net,\n--\nnet/rds/connection.c-457-{\nnet/rds/connection.c:458:\treturn __rds_conn_create(net, laddr, faddr, trans, gfp, tos, 0, dev_if);\nnet/rds/connection.c-459-}\n--\nnet/rds/connection.c=462=struct rds_connection *rds_conn_create_outgoing(struct net *net,\n--\nnet/rds/connection.c-467-{\nnet/rds/connection.c:468:\treturn __rds_conn_create(net, laddr, faddr, trans, gfp, tos, 1, dev_if);\nnet/rds/connection.c-469-}\n--\nnet/rds/connection.c=744=void rds_conn_destroy(struct rds_connection *conn)\n--\nnet/rds/connection.c-781-\t/* Snatch c_passive while holding the lock:\nnet/rds/connection.c:782:\t * __rds_conn_create() dereferences it under rcu_read_lock()\nnet/rds/connection.c-783-\t * (and refuses to install a new one once c_destroy_in_prog is\n--\nnet/rds/connection.c-791-\t/* If we are a parent's passive twin, invalidate its pointer to\nnet/rds/connection.c:792:\t * us as well, so that __rds_conn_create() cannot hand out a\nnet/rds/connection.c-793-\t * connection whose teardown has begun.  The parent is the\n--\nnet/rds/ib_cm.c=1283=void rds_ib_conn_free(void *arg)\n--\nnet/rds/ib_cm.c-1296-\t * Callers may hold rds_conn_lock with interrupts disabled\nnet/rds/ib_cm.c:1297:\t * (__rds_conn_create() undoing a lost creation race), so do not\nnet/rds/ib_cm.c-1298-\t * re-enable interrupts unconditionally here.\n"
}

13/2 2026/09/14 04:29 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 28115
							output: 224
							thoughts: 797
						

Thoughts:
**My Analysis of the RDS Patch**

Okay, let's break this down. My initial assessment reveals a deep dive into the RDS networking stack. I see a clear delineation of functions, all prefixed with `rds_`, indicating their core association with the AF_RDS socket family. Functions like `rds_conn_create`, `rds_conn_destroy`, and others are all specific to RDS, confirming its distinct nature. I'm focusing on the connection lifecycle, which seems central to this patch's modifications.

I've zeroed in on the key symbols, focusing on connection management and data flow: `rds_conn_destroy`, `rds_conn_put`, `rds_conn_get`, `rds_conn_destroy_fini`, `rds_conn_create`, `rds_conn_create_outgoing`, along with `rds_sendmsg`, `rds_send_queue_rm`, `rds_release`, and `rds_tcp_accept_one`.

The core reasoning aligns with my understanding: This patch introduces reference counting (`kref`) for `struct rds_connection`. This is a significant change, as it defers connection freeing until all references are dropped, making the synchronization and connection lifecycle management critical across RDS core and transport layers. This spans multiple files, including `connection.c`, `af_rds.c`, `send.c`, `recv.c`, `message.c`, and the transport-specific code in `tcp.c`. The potential for races, memory leaks, and use-after-free bugs is definitely something I need to keep in mind.

Now, considering the configuration aspects, I'm noting the relevant Kconfig options: `RDS`, `RDS_TCP`, `RDS_RDMA`, `INFINIBAND`, and `INFINIBAND_ADDR_TRANS`. I've confirmed that `RDS_TCP` and `RDS_RDMA` are tristate, meaning they can be compiled as modules or built-in, which influences how we approach the fuzzing. `RDS` has no further dependencies to consider.

I'm now ready to call `set-results`.



14/2 2026/09/14 04:29 tool set-results
0m Args:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND",
    "INFINIBAND_ADDR_TRANS"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_destroy_fini",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_conn_create_outgoing",
    "rds_sendmsg",
    "rds_send_queue_rm",
    "rds_release",
    "rds_tcp_accept_one"
  ],
  "Reasoning": "The patch substantially refactors connection lifetime and reference counting (using kref) in the RDS subsystem (net/rds), deferring connection cleanup until all references from messages, incoming queues, sockets, and transports are dropped. These changes touch core networking logic reachable via AF_RDS sockets and could introduce subtle race conditions, refcounting leaks, use-after-free, or deadlocks.",
  "WorthFuzzing": true
}

Results:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND",
    "INFINIBAND_ADDR_TRANS"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_destroy_fini",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_conn_create_outgoing",
    "rds_sendmsg",
    "rds_send_queue_rm",
    "rds_release",
    "rds_tcp_accept_one"
  ],
  "Reasoning": "The patch substantially refactors connection lifetime and reference counting (using kref) in the RDS subsystem (net/rds), deferring connection cleanup until all references from messages, incoming queues, sockets, and transports are dropped. These changes touch core networking logic reachable via AF_RDS sockets and could introduce subtle race conditions, refcounting leaks, use-after-free, or deadlocks.",
  "WorthFuzzing": true
}

15/1 2026/09/14 04:29 action If
0m Args:
{
  "WorthFuzzing": true
}

Results:
null

16/2 2026/09/14 04:29 agent kmsan-evaluator
0m Model:
core

Args:
null

Results:
{
  "KMSANReasoning": "The patch introduces reference counting (kref) and synchronization fixes for RDS connections to prevent use-after-free, races, and premature frees during connection teardown and module unloading. All newly added or modified data structures (e.g. struct rds_connection, rds_conn_path) continue to be allocated using zero-initializing allocators (kmem_cache_zalloc, kzalloc_objs), and all local variables are properly initialized. The only user-copy path touched is SIOCRDSGETTOS, which merely updates the spinlock protecting rs_tos (a single __u8 value). The risks associated with these changes are strictly related to object lifetimes, concurrent access, reference leaks, or locking deadlocks, which are effectively covered by KASAN and LOCKDEP. No uninitialized memory reads or information leak hazards are introduced or exposed, so a dedicated KMSAN fuzzing session is not warranted.",
  "NeedsKMSAN": false
}

Instruction:
You are an expert Linux kernel security engineer specializing in kernel memory
error detectors (KASAN and KMSAN). Your job is to review the provided patch series and
determine if the code changes justify spawning a dedicated KMSAN (KernelMemorySanitizer)
fuzzing session in addition to standard KASAN fuzzing.

CRITICAL DISTINCTION BETWEEN KASAN AND KMSAN:
- Standard KASAN kernel builds (upstream-apparmor-kasan.config) already enable
  a comprehensive suite of debugging tools and sanitizers, including KASAN
  (out-of-bounds accesses, use-after-free, double free, invalid free), LOCKDEP
  (locking bugs and deadlocks), UB-sanitizers, and memory corruption checks.
- KMSAN (KernelMemorySanitizer) detects reads of UNINITIALIZED memory (stack, heap,
  or page allocations) and kernel-to-user memory info-leaks.

Rule: THERE IS NO SENSE IN RUNNING A KMSAN SESSION IF A BUG CAN BE CAUGHT BY KASAN,
LOCKDEP, OR OTHER STANDARD BUG DETECTORS.
A dedicated KMSAN fuzzing session incurs significant resource costs. You must ONLY
set NeedsKMSAN=true if the code changes introduce or expose UNINITIALIZED MEMORY risks
that are detected ONLY by KMSAN.

Look holistically at the patch series and surrounding code. Even if no direct
uninitialized field accesses or new buffer allocations are added in the diff itself,
a patch may alter control flow, bounds checking, or data length calculations in ways
that change how the rest of the code operates on existing buffers (e.g. allowing
uninitialized stack/heap memory to be read, copied to user space, or used in control
flow). Do not hesitate to use your code access tools to inspect the surrounding code,
called functions, and callers.

Set NeedsKMSAN=true ONLY IF the patch introduces or modifies:
1. Kernel structures sent to user space (via copy_to_user, put_user, netlink skb
   attributes, ioctl output arguments, socket options, or BPF buffers) where fields
   or structure padding might not be fully initialized/zeroed.
2. Conditional logic or branching that depends on potentially uninitialized variables
   or struct fields.
3. Allocation or initialization of complex data structures where uninitialized fields
   could be read later in reachable code paths.
4. Bounds checks, lengths, or logic in a way that allows surrounding code to access
   uninitialized bytes of existing buffers.

Set NeedsKMSAN=false IF:
- The code changes primarily risk out-of-bounds access, array overflows, NULL pointer
  dereferences, locking deadlocks, or use-after-free bugs (these are already caught
  by KASAN, LOCKDEP, or standard bug detectors).
- All stack/heap structures touched or introduced by the patch are fully zeroed
  or initialized (e.g. using = {0}, memset, kzalloc) before being read or copied.
- The patch does not introduce any risk of uninitialized memory usage or info-leaks.

Use your code access tools to inspect the surrounding code if necessary, then provide
detailed KMSANReasoning contrasting KASAN vs KMSAN applicability for this patch.
Prefer calling several tools at the same time to save round-trips.


Use set-results tool to provide results of the analysis.
It must be called exactly once before the final reply.
Ignore results of this tool.

Prompt:
Target architecture: amd64

For your convenience, here is the diff of the changes:
commit fe0d65dad5bf3cdb3ebccbae9d834aa98bc8f6e0
Author: syz-cluster <triage@syzkaller.com>
Date:   Mon Sep 14 04:28:33 2026 +0000

    syz-cluster: applied patch under review

diff --git a/net/rds/af_rds.c b/net/rds/af_rds.c
index d5defe9172e36..1cc20b5cfd213 100644
--- a/net/rds/af_rds.c
+++ b/net/rds/af_rds.c
@@ -80,6 +80,14 @@ static int rds_release(struct socket *sock)
 	rds_notify_queue_get(rs, NULL);
 	rds_notify_msg_zcopy_purge(&rs->rs_zcookie_queue);
 
+	/* drop the cached connection reference; no sendmsg can race
+	 * with us here, the socket is going away
+	 */
+	if (rs->rs_conn) {
+		rds_conn_put(rs->rs_conn);
+		rs->rs_conn = NULL;
+	}
+
 	spin_lock_bh(&rds_sock_lock);
 	list_del_init(&rs->rs_item);
 	spin_unlock_bh(&rds_sock_lock);
@@ -255,6 +263,7 @@ static int rds_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
 {
 	struct rds_sock *rs = rds_sk_to_rs(sock->sk);
 	rds_tos_t utos, tos = 0;
+	unsigned long flags;
 
 	switch (cmd) {
 	case SIOCRDSSETTOS:
@@ -267,18 +276,23 @@ static int rds_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
 		else
 			return -ENOIOCTLCMD;
 
-		spin_lock_bh(&rds_sock_lock);
+		/* rs_conn is serialized by rs_lock (see rds_sendmsg());
+		 * hold it across the "no connection yet" check and the
+		 * rs_tos store so a racing sendmsg cannot cache a conn
+		 * whose c_tos then disagrees with rs_tos.
+		 */
+		spin_lock_irqsave(&rs->rs_lock, flags);
 		if (rs->rs_tos || rs->rs_conn) {
-			spin_unlock_bh(&rds_sock_lock);
+			spin_unlock_irqrestore(&rs->rs_lock, flags);
 			return -EINVAL;
 		}
 		rs->rs_tos = tos;
-		spin_unlock_bh(&rds_sock_lock);
+		spin_unlock_irqrestore(&rs->rs_lock, flags);
 		break;
 	case SIOCRDSGETTOS:
-		spin_lock_bh(&rds_sock_lock);
+		spin_lock_irqsave(&rs->rs_lock, flags);
 		tos = rs->rs_tos;
-		spin_unlock_bh(&rds_sock_lock);
+		spin_unlock_irqrestore(&rs->rs_lock, flags);
 		if (put_user(tos, (rds_tos_t __user *)arg))
 			return -EFAULT;
 		break;
diff --git a/net/rds/connection.c b/net/rds/connection.c
index b6c4beb50eaf0..f5f25e4a53b97 100644
--- a/net/rds/connection.c
+++ b/net/rds/connection.c
@@ -47,7 +47,8 @@
 
 /* converting this to RCU is a chore for another day.. */
 static DEFINE_SPINLOCK(rds_conn_lock);
-static unsigned long rds_conn_count;
+/* woken whenever a transport's t_conn_count drops to zero */
+static DECLARE_WAIT_QUEUE_HEAD(rds_conn_freed_waitq);
 static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];
 static struct kmem_cache *rds_conn_slab;
 
@@ -79,7 +80,18 @@ static struct hlist_head *rds_conn_bucket(const struct in6_addr *laddr,
 		var |= RDS_INFO_CONNECTION_FLAG_##suffix;	\
 } while (0)
 
-/* rcu read lock must be held or the connection spinlock */
+/* rcu read lock must be held or the connection spinlock.
+ * On success a reference is taken on the returned connection; the
+ * caller must drop it with rds_conn_put().
+ */
+/* c_passive is written under rds_conn_lock and read under RCU */
+static struct rds_connection *
+rds_conn_passive_locked(struct rds_connection *conn)
+{
+	return rcu_dereference_protected(conn->c_passive,
+					 lockdep_is_held(&rds_conn_lock));
+}
+
 static struct rds_connection *rds_conn_lookup(struct net *net,
 					      struct hlist_head *head,
 					      const struct in6_addr *laddr,
@@ -96,6 +108,17 @@ static struct rds_connection *rds_conn_lookup(struct net *net,
 		    conn->c_tos == tos &&
 		    net == rds_conn_net(conn) &&
 		    conn->c_dev_if == dev_if) {
+			/* Only ever hand out a live reference.
+			 * rds_conn_destroy() unhashes under
+			 * rds_conn_lock and waits a grace period
+			 * before dropping the initial reference, so
+			 * an entry this traversal reaches still holds
+			 * at least that one; the conditional get
+			 * documents the contract rather than
+			 * papering over a zero-refcount entry.
+			 */
+			if (!kref_get_unless_zero(&conn->c_refcount))
+				continue;
 			ret = conn;
 			break;
 		}
@@ -161,6 +184,22 @@ static void __rds_conn_path_init(struct rds_connection *conn,
 	cp->cp_flags = 0;
 }
 
+/* Undo trans->conn_alloc(): it may have allocated transport data for
+ * every path of a multipath connection, not just for path 0.
+ */
+static void rds_conn_free_transport_data(struct rds_connection *conn,
+					 int npaths)
+{
+	struct rds_conn_path *cp;
+	int i;
+
+	for (i = 0; i < npaths; i++) {
+		cp = &conn->c_path[i];
+		if (cp->cp_transport_data)
+			conn->c_trans->conn_free(cp->cp_transport_data);
+	}
+}
+
 /*
  * There is only every one 'conn' for a given pair of addresses in the
  * system at a time.  They contain messages to be retransmitted and so
@@ -197,7 +236,20 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		 * We need a second connection object into which we
 		 * can stick the other QP. */
 		parent = conn;
-		conn = parent->c_passive;
+		/* The c_passive pointer holds a reference which is only
+		 * dropped one synchronize_rcu() after the pointer is
+		 * cleared, so within this RCU section a fetched pointer
+		 * is always safe to take a reference on.  A passive conn
+		 * whose own destroy has begun is not handed out, though:
+		 * it is quiesced and about to clear the parent's pointer
+		 * itself, and reusing it would re-arm a connection that
+		 * nothing will tear down again.
+		 */
+		conn = rcu_dereference(parent->c_passive);
+		if (conn && READ_ONCE(conn->c_destroy_in_prog))
+			conn = NULL;
+		if (conn)
+			rds_conn_get(conn);
 	}
 	rcu_read_unlock();
 	if (conn)
@@ -215,6 +267,7 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		goto out;
 	}
 
+	kref_init(&conn->c_refcount);
 	INIT_HLIST_NODE(&conn->c_hash_node);
 	conn->c_laddr = *laddr;
 	conn->c_isv6 = !ipv6_addr_v4mapped(laddr);
@@ -278,12 +331,13 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 
 	init_waitqueue_head(&conn->c_hs_waitq);
 	for (i = 0; i < npaths; i++) {
+		int seq = atomic_read(&trans->t_conn_count);
+
 		__rds_conn_path_init(conn, &conn->c_path[i],
 				     is_outgoing);
 		conn->c_path[i].cp_index = i;
 		conn->c_path[i].cp_wq =
-			alloc_ordered_workqueue("krds_cp_wq#%lu/%d", 0,
-						rds_conn_count, i);
+			alloc_ordered_workqueue("krds_cp_wq#%d/%d", 0, seq, i);
 		if (!conn->c_path[i].cp_wq)
 			conn->c_path[i].cp_wq = rds_wq;
 	}
@@ -315,15 +369,46 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 	spin_lock_irqsave(&rds_conn_lock, flags);
 	if (parent) {
 		/* Creating passive conn */
-		if (parent->c_passive) {
-			trans->conn_free(conn->c_path[0].cp_transport_data);
+		if (READ_ONCE(parent->c_destroy_in_prog)) {
+			/* The parent's destroy has begun (it sets the
+			 * flag and snatches c_passive under this
+			 * lock); do not install a new passive conn
+			 * that nothing would ever destroy.
+			 */
+			rds_conn_free_transport_data(conn, npaths);
+			free_cp = conn->c_path;
+			kmem_cache_free(rds_conn_slab, conn);
+			conn = ERR_PTR(-ENETDOWN);
+		} else if (rcu_access_pointer(parent->c_passive)) {
+			struct rds_connection *passive;
+
+			passive = rds_conn_passive_locked(parent);
+			rds_conn_free_transport_data(conn, npaths);
 			free_cp = conn->c_path;
 			kmem_cache_free(rds_conn_slab, conn);
-			conn = parent->c_passive;
+			if (READ_ONCE(passive->c_destroy_in_prog)) {
+				/* Its destroy will clear the parent's
+				 * pointer under this lock shortly; until
+				 * then there is no usable passive conn.
+				 */
+				conn = ERR_PTR(-ENETDOWN);
+			} else {
+				rds_conn_get(passive);
+				conn = passive;
+			}
 		} else {
-			parent->c_passive = conn;
+			/* The initial reference belongs to whoever
+			 * destroys the conn (the transport's conn
+			 * lists, as for any other conn).  Take one
+			 * for the c_passive pointer - dropped when
+			 * the parent is destroyed - and one for our
+			 * caller.
+			 */
+			rds_conn_get(conn);	/* c_passive */
+			rds_conn_get(conn);	/* caller */
+			rcu_assign_pointer(parent->c_passive, conn);
 			rds_cong_add_conn(conn);
-			rds_conn_count++;
+			atomic_inc(&conn->c_trans->t_conn_count);
 		}
 	} else {
 		/* Creating normal conn */
@@ -332,33 +417,28 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		found = rds_conn_lookup(net, head, laddr, faddr, trans,
 					tos, dev_if);
 		if (found) {
-			struct rds_conn_path *cp;
-			int i;
-
-			for (i = 0; i < npaths; i++) {
-				cp = &conn->c_path[i];
-				/* The ->conn_alloc invocation may have
-				 * allocated resource for all paths, so all
-				 * of them may have to be freed here.
-				 */
-				if (cp->cp_transport_data)
-					trans->conn_free(cp->cp_transport_data);
-			}
+			rds_conn_free_transport_data(conn, npaths);
 			free_cp = conn->c_path;
 			kmem_cache_free(rds_conn_slab, conn);
 			conn = found;
 		} else {
 			conn->c_my_gen_num = rds_gen_num;
 			conn->c_peer_gen_num = 0;
+			/* the initial reference belongs to whoever
+			 * destroys the conn; take one for our caller
+			 */
+			rds_conn_get(conn);
 			hlist_add_head_rcu(&conn->c_hash_node, head);
 			rds_cong_add_conn(conn);
-			rds_conn_count++;
+			atomic_inc(&conn->c_trans->t_conn_count);
 		}
 	}
 	spin_unlock_irqrestore(&rds_conn_lock, flags);
 	rcu_read_unlock();
 
 out:
+	if (parent)
+		rds_conn_put(parent);
 	if (free_cp) {
 		for (i = 0; i < npaths; i++)
 			if (free_cp[i].cp_wq != rds_wq)
@@ -466,9 +546,10 @@ void rds_conn_shutdown(struct rds_conn_path *cp)
 			 * Quiesce the reconnect timer before bailing
 			 * out, though.  When a pending destroy did
 			 * suppress the queue, no later pass runs, and
-			 * rds_conn_path_destroy() is about to flush
-			 * cp_down_w and free the path: it must not
-			 * find cp_conn_w still armed.  A successor
+			 * rds_conn_path_quiesce() is about to flush
+			 * cp_down_w, ahead of the path's deferred
+			 * free: it must not find cp_conn_w still
+			 * armed.  A successor
 			 * pass, when there is one, re-arms the
 			 * reconnect from its own tail.
 			 */
@@ -515,12 +596,16 @@ void rds_conn_shutdown(struct rds_conn_path *cp)
 		conn->c_trans->conn_slots_available(conn, false);
 }
 
-/* destroy a single rds_conn_path. rds_conn_destroy() iterates over
- * all paths using rds_conn_path_destroy()
+/* quiesce a single rds_conn_path: shut it down and tear down any
+ * queued messages.  rds_conn_destroy() iterates over all paths using
+ * rds_conn_path_quiesce(); the transport state and the workqueue are
+ * freed later, from rds_conn_path_free().
  */
-static void rds_conn_path_destroy(struct rds_conn_path *cp)
+static void rds_conn_path_quiesce(struct rds_conn_path *cp)
 {
 	struct rds_message *rm, *rtmp;
+	unsigned long flags;
+	LIST_HEAD(purge);
 
 	if (!cp->cp_transport_data)
 		return;
@@ -532,10 +617,16 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)
 	rds_conn_path_drop(cp, true);
 	flush_work(&cp->cp_down_w);
 
-	/* tear down queued messages */
-	list_for_each_entry_safe(rm, rtmp,
-				 &cp->cp_send_queue,
-				 m_conn_item) {
+	/* Tear down queued messages.  Take the queue under cp_lock:
+	 * a sender that still holds a reference can be inside
+	 * rds_send_queue_rm() right now, and it tests
+	 * rds_destroy_pending() under the same lock, so after this
+	 * splice nothing is added behind our back.
+	 */
+	spin_lock_irqsave(&cp->cp_lock, flags);
+	list_splice_init(&cp->cp_send_queue, &purge);
+	spin_unlock_irqrestore(&cp->cp_lock, flags);
+	list_for_each_entry_safe(rm, rtmp, &purge, m_conn_item) {
 		list_del_init(&rm->m_conn_item);
 		BUG_ON(!list_empty(&rm->m_sock_item));
 		rds_message_put(rm);
@@ -547,6 +638,16 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)
 	WARN_ON(delayed_work_pending(&cp->cp_recv_w));
 	WARN_ON(delayed_work_pending(&cp->cp_conn_w));
 	WARN_ON(work_pending(&cp->cp_down_w));
+}
+
+/* free a quiesced rds_conn_path's transport state and workqueue; runs
+ * from rds_conn_destroy_fini() once the last connection reference is
+ * dropped.
+ */
+static void rds_conn_path_free(struct rds_conn_path *cp)
+{
+	if (!cp->cp_transport_data)
+		return;
 
 	if (cp->cp_wq != rds_wq) {
 		destroy_workqueue(cp->cp_wq);
@@ -556,17 +657,96 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp)
 	cp->cp_conn->c_trans->conn_free(cp->cp_transport_data);
 }
 
+/* Free a connection.  This runs from rds_conn_put() when the last
+ * reference is dropped, after rds_conn_destroy() has quiesced the
+ * connection and dropped the initial reference.
+ */
+static void rds_conn_destroy_fini(struct kref *kref)
+{
+	struct rds_connection *conn = container_of(kref, struct rds_connection,
+						   c_refcount);
+	int npaths = (conn->c_trans->t_mp_capable ? RDS_MPATH_WORKERS : 1);
+	struct rds_transport *trans = conn->c_trans;
+	int i;
+
+	for (i = 0; i < npaths; i++)
+		rds_conn_path_free(&conn->c_path[i]);
+
+	kfree(conn->c_path);
+	kmem_cache_free(rds_conn_slab, conn);
+
+	/* only after everything the transport module owns has been
+	 * freed above may its unload proceed
+	 */
+	if (!atomic_dec_return(&trans->t_conn_count))
+		wake_up_all(&rds_conn_freed_waitq);
+}
+
+/* Wait for all of @trans's connections to be freed; the free runs
+ * asynchronously once rds_conn_destroy() has quiesced a connection.
+ * Called on transport module unload, after the transport has destroyed
+ * all of its connections.  A connection reference can be held for an
+ * application-controlled time - an unread datagram pins the inc that
+ * carries it, and thus the connection - so the wait is unbounded: the
+ * frees that run after unload call into this module's text (conn_free,
+ * inc_free) and free into its slabs, so proceeding while any remain
+ * would be a use-after-free, not a leak.  Warn periodically so a stuck
+ * count is diagnosable, but never stop waiting.  This matches the
+ * historical RDS contract that teardown does not discard queued data.
+ */
+void rds_conn_wait_conns_freed(struct rds_transport *trans,
+			       void (*resweep)(void))
+{
+	unsigned long warn_interval =
+			msecs_to_jiffies(RDS_CONN_FREE_WARN_INTERVAL_MS);
+	unsigned long warn_at = jiffies + warn_interval;
+
+	while (!wait_event_timeout(rds_conn_freed_waitq,
+				   !atomic_read(&trans->t_conn_count),
+				   msecs_to_jiffies(RDS_CONN_FREE_POLL_MS))) {
+		/* A transport whose teardown is asynchronous (IB moves a
+		 * connection off its device from the shutdown work) gives
+		 * us a resweep to destroy what has arrived since.
+		 */
+		if (resweep)
+			resweep();
+		if (time_after_eq(jiffies, warn_at)) {
+			pr_warn("RDS/%s: still waiting for %d connection(s) to be freed before unload\n",
+				trans->t_name,
+				atomic_read(&trans->t_conn_count));
+			warn_at = jiffies + warn_interval;
+		}
+	}
+}
+EXPORT_SYMBOL_GPL(rds_conn_wait_conns_freed);
+
+void rds_conn_get(struct rds_connection *conn)
+{
+	kref_get(&conn->c_refcount);
+}
+EXPORT_SYMBOL_GPL(rds_conn_get);
+
+void rds_conn_put(struct rds_connection *conn)
+{
+	kref_put(&conn->c_refcount, rds_conn_destroy_fini);
+}
+EXPORT_SYMBOL_GPL(rds_conn_put);
+
 /*
  * Stop and free a connection.
  *
- * This can only be used in very limited circumstances.  It assumes that once
- * the conn has been shutdown that no one else is referencing the connection.
- * We can only ensure this in the rmmod path in the current code.
+ * Quiesces the connection synchronously (workers cancelled, transport
+ * connections shut down, queued messages dropped) and drops the
+ * initial reference.  The memory - including the transport's
+ * per-connection state and the path workqueues - is freed once the
+ * last rds_conn_put() runs, which may be after this returns.
  */
 void rds_conn_destroy(struct rds_connection *conn)
 {
-	unsigned long flags;
 	int i;
+	struct rds_connection *passive, *parent;
+	struct hlist_head *head;
+	bool was_passive = false;
 	struct rds_conn_path *cp;
 	int npaths = (conn->c_trans->t_mp_capable ? RDS_MPATH_WORKERS : 1);
 
@@ -574,16 +754,67 @@ void rds_conn_destroy(struct rds_connection *conn)
 		 "%pI4\n", conn, &conn->c_laddr,
 		 &conn->c_faddr);
 
-	/* Ensure conn will not be scheduled for reconnect */
+	/* Make rds_destroy_pending() true for this conn.  Together with
+	 * the synchronize_rcu() below this stops the work-requeueing
+	 * sites (which all test rds_destroy_pending() under
+	 * rcu_read_lock()) from queueing new work on the path
+	 * workqueues once we start cancelling and destroying them.
+	 *
+	 * Now that the transport state stays discoverable (e.g. on the
+	 * transports' connection lists) until the final rds_conn_put(),
+	 * a conn can be handed to rds_conn_destroy() more than once -
+	 * e.g. dropped for a protocol version mismatch and then found
+	 * again at module unload.  Only the first caller proceeds; the
+	 * unhash also happens under rds_conn_lock, so a looked-up conn
+	 * can never be quiesced twice.
+	 */
 	spin_lock_irq(&rds_conn_lock);
+	if (conn->c_destroy_in_prog) {
+		spin_unlock_irq(&rds_conn_lock);
+		return;
+	}
+	WRITE_ONCE(conn->c_destroy_in_prog, true);
+
+	/* Ensure conn will not be scheduled for reconnect */
 	hlist_del_init_rcu(&conn->c_hash_node);
+
+	/* Snatch c_passive while holding the lock:
+	 * __rds_conn_create() dereferences it under rcu_read_lock()
+	 * (and refuses to install a new one once c_destroy_in_prog is
+	 * set, which it checks under this lock).  After the
+	 * synchronize_rcu() below no one can pick the pointer up any
+	 * more and its reference can be dropped.
+	 */
+	passive = rds_conn_passive_locked(conn);
+	RCU_INIT_POINTER(conn->c_passive, NULL);
+
+	/* If we are a parent's passive twin, invalidate its pointer to
+	 * us as well, so that __rds_conn_create() cannot hand out a
+	 * connection whose teardown has begun.  The parent is the
+	 * hashed connection for our key (a passive conn is never
+	 * hashed, and we unhashed ourselves above); it holds its
+	 * initial reference for as long as it is hashed, so the lookup
+	 * reference dropped below cannot be its last.
+	 */
+	head = rds_conn_bucket(&conn->c_laddr, &conn->c_faddr);
+	rcu_read_lock();
+	parent = rds_conn_lookup(rds_conn_net(conn), head, &conn->c_laddr,
+				 &conn->c_faddr, conn->c_trans, conn->c_tos,
+				 conn->c_dev_if);
+	rcu_read_unlock();
+	if (parent && rds_conn_passive_locked(parent) == conn) {
+		RCU_INIT_POINTER(parent->c_passive, NULL);
+		was_passive = true;
+	}
 	spin_unlock_irq(&rds_conn_lock);
+	if (parent)
+		rds_conn_put(parent);
 	synchronize_rcu();
 
 	/* shut the connection down */
 	for (i = 0; i < npaths; i++) {
 		cp = &conn->c_path[i];
-		rds_conn_path_destroy(cp);
+		rds_conn_path_quiesce(cp);
 		BUG_ON(!list_empty(&cp->cp_retrans));
 	}
 
@@ -594,12 +825,19 @@ void rds_conn_destroy(struct rds_connection *conn)
 	 */
 	rds_cong_remove_conn(conn);
 
-	kfree(conn->c_path);
-	kmem_cache_free(rds_conn_slab, conn);
+	/* drop the reference our c_passive pointer held, if any, and
+	 * the one a parent's c_passive pointer held on us; neither can
+	 * be the last, since the initial reference is dropped below
+	 */
+	if (passive)
+		rds_conn_put(passive);
+	if (was_passive)
+		rds_conn_put(conn);
 
-	spin_lock_irqsave(&rds_conn_lock, flags);
-	rds_conn_count--;
-	spin_unlock_irqrestore(&rds_conn_lock, flags);
+	/* drop the initial reference; the connection is freed from
+	 * rds_conn_destroy_fini() once every holder has dropped theirs
+	 */
+	rds_conn_put(conn);
 }
 EXPORT_SYMBOL_GPL(rds_conn_destroy);
 
diff --git a/net/rds/ib.c b/net/rds/ib.c
index 786f39169bc14..3fc2de9d19d55 100644
--- a/net/rds/ib.c
+++ b/net/rds/ib.c
@@ -525,10 +525,7 @@ static void rds_ib_set_unloading(void)
 
 static bool rds_ib_is_unloading(struct rds_connection *conn)
 {
-	struct rds_conn_path *cp = &conn->c_path[0];
-
-	return (test_bit(RDS_DESTROY_PENDING, &cp->cp_flags) ||
-		atomic_read(&rds_ib_unloading) != 0);
+	return atomic_read(&rds_ib_unloading) != 0;
 }
 
 void rds_ib_exit(void)
@@ -540,7 +537,24 @@ void rds_ib_exit(void)
 	rds_info_deregister_func(RDS6_INFO_IB_CONNECTIONS, rds6_ib_ic_info);
 #endif
 	rds_ib_unregister_client();
+
+	/* rds_ib_dev_shutdown() only dropped the connections still
+	 * attached to a device; each moves itself to ib_nodev_conns
+	 * from its shutdown work.  Destroy what is there now and keep
+	 * sweeping the list while the wait sees connections outstanding,
+	 * so a late arrival is destroyed rather than waited on forever.
+	 */
 	rds_ib_destroy_nodev_conns();
+	rds_conn_wait_conns_freed(&rds_ib_transport,
+				  rds_ib_destroy_nodev_conns);
+
+	/* Tearing down the last connection may have dropped the final
+	 * reference on a device, deferring rds_ib_dev_free() to rds_wq.
+	 * Drain it before the module goes away; it queues nothing
+	 * further on rds_wq.
+	 */
+	flush_workqueue(rds_wq);
+
 	rds_ib_sysctl_exit();
 	rds_ib_recv_exit();
 	rds_trans_unregister(&rds_ib_transport);
diff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c
index 4feb0edc360c8..257e03ed48618 100644
--- a/net/rds/ib_cm.c
+++ b/net/rds/ib_cm.c
@@ -874,6 +874,13 @@ int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id,
 	 * see the comment above rds_queue_reconnect()
 	 */
 	mutex_lock(&conn->c_cm_lock);
+	/* A destroy that has already quiesced this conn leaves it in
+	 * RDS_CONN_DOWN with no cm_id, exactly what the transition
+	 * below would happily claim; nothing would tear the new cm_id
+	 * and QP down again before the conn is freed.  Reject instead.
+	 */
+	if (rds_destroy_pending(conn))
+		goto out;
 	if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_CONNECTING)) {
 		if (rds_conn_state(conn) == RDS_CONN_UP) {
 			rdsdebug("incoming connect while connecting\n");
@@ -924,8 +931,14 @@ int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id,
 		rds_ib_conn_error(conn, "rdma_accept failed\n");
 
 out:
-	if (conn)
+	if (conn) {
 		mutex_unlock(&conn->c_cm_lock);
+		/* The conn stays reachable through cm_id->context
+		 * without a reference of its own: connection destroy
+		 * shuts the cm_id down before the conn is freed.
+		 */
+		rds_conn_put(conn);
+	}
 	if (err)
 		rdma_reject(cm_id, &err, sizeof(int),
 			    IB_CM_REJ_CONSUMER_DEFINED);
@@ -1271,6 +1284,7 @@ void rds_ib_conn_free(void *arg)
 {
 	struct rds_ib_connection *ic = arg;
 	spinlock_t	*lock_ptr;
+	unsigned long flags;
 
 	rdsdebug("ic %p\n", ic);
 
@@ -1278,12 +1292,18 @@ void rds_ib_conn_free(void *arg)
 	 * Conn is either on a dev's list or on the nodev list.
 	 * A race with shutdown() or connect() would cause problems
 	 * (since rds_ibdev would change) but that should never happen.
+	 *
+	 * Callers may hold rds_conn_lock with interrupts disabled
+	 * (__rds_conn_create() undoing a lost creation race), so do not
+	 * re-enable interrupts unconditionally here.
 	 */
 	lock_ptr = ic->rds_ibdev ? &ic->rds_ibdev->spinlock : &ib_nodev_conns_lock;
 
-	spin_lock_irq(lock_ptr);
-	list_del(&ic->ib_node);
-	spin_unlock_irq(lock_ptr);
+	spin_lock_irqsave(lock_ptr, flags);
+	/* already unlinked if a transport teardown gathered us first */
+	if (!list_empty(&ic->ib_node))
+		list_del(&ic->ib_node);
+	spin_unlock_irqrestore(lock_ptr, flags);
 
 	rds_ib_recv_free_caches(ic);
 
diff --git a/net/rds/ib_rdma.c b/net/rds/ib_rdma.c
index db7e92e7bd29f..91db43a0e7d73 100644
--- a/net/rds/ib_rdma.c
+++ b/net/rds/ib_rdma.c
@@ -165,11 +165,21 @@ void rds_ib_destroy_nodev_conns(void)
 
 	/* avoid calling conn_destroy with irqs off */
 	spin_lock_irq(&ib_nodev_conns_lock);
-	list_splice(&ib_nodev_conns, &tmp_list);
+	list_splice_init(&ib_nodev_conns, &tmp_list);
 	spin_unlock_irq(&ib_nodev_conns_lock);
 
-	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node)
+	/* rds_conn_destroy() can return before the connection is freed,
+	 * and it is the free - rds_ib_conn_free() - that unlinks ib_node.
+	 * tmp_list lives on this stack frame, so unlink each node before
+	 * its destroy; the free then finds it empty and leaves it alone.
+	 */
+	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node) {
+		spin_lock_irq(&ib_nodev_conns_lock);
+		list_del_init(&ic->ib_node);
+		spin_unlock_irq(&ib_nodev_conns_lock);
+
 		rds_conn_destroy(ic->conn);
+	}
 }
 
 void rds_ib_get_mr_info(struct rds_ib_device *rds_ibdev, struct rds_info_rdma_connection *iinfo)
diff --git a/net/rds/ib_recv.c b/net/rds/ib_recv.c
index bd6cb3ffaa571..7d45808544a0d 100644
--- a/net/rds/ib_recv.c
+++ b/net/rds/ib_recv.c
@@ -458,7 +458,11 @@ void rds_ib_recv_refill(struct rds_connection *conn, int prefill, gfp_t gfp)
 	    (must_wake ||
 	    (can_wait && rds_ib_ring_low(&ic->i_recv_ring)) ||
 	    rds_ib_ring_empty(&ic->i_recv_ring))) {
-		queue_delayed_work(conn->c_path->cp_wq, &conn->c_recv_w, 1);
+		rcu_read_lock();
+		if (!rds_destroy_pending(conn))
+			queue_delayed_work(conn->c_path->cp_wq,
+					   &conn->c_recv_w, 1);
+		rcu_read_unlock();
 	}
 	if (can_wait)
 		cond_resched();
diff --git a/net/rds/ib_send.c b/net/rds/ib_send.c
index d6be95542119f..bc411e96ad12d 100644
--- a/net/rds/ib_send.c
+++ b/net/rds/ib_send.c
@@ -298,8 +298,13 @@ void rds_ib_send_cqe_handler(struct rds_ib_connection *ic, struct ib_wc *wc)
 	rds_ib_sub_signaled(ic, nr_sig);
 
 	if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags) ||
-	    test_bit(0, &conn->c_map_queued))
-		queue_delayed_work(conn->c_path->cp_wq, &conn->c_send_w, 0);
+	    test_bit(0, &conn->c_map_queued)) {
+		rcu_read_lock();
+		if (!rds_destroy_pending(conn))
+			queue_delayed_work(conn->c_path->cp_wq,
+					   &conn->c_send_w, 0);
+		rcu_read_unlock();
+	}
 
 	/* We expect errors as the qp is drained during shutdown */
 	if (wc->status != IB_WC_SUCCESS && rds_conn_up(conn)) {
@@ -420,8 +425,13 @@ void rds_ib_send_add_credits(struct rds_connection *conn, unsigned int credits)
 			test_bit(RDS_LL_SEND_FULL, &conn->c_flags) ? ", ll_send_full" : "");
 
 	atomic_add(IB_SET_SEND_CREDITS(credits), &ic->i_credits);
-	if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags))
-		queue_delayed_work(conn->c_path->cp_wq, &conn->c_send_w, 0);
+	if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags)) {
+		rcu_read_lock();
+		if (!rds_destroy_pending(conn))
+			queue_delayed_work(conn->c_path->cp_wq,
+					   &conn->c_send_w, 0);
+		rcu_read_unlock();
+	}
 
 	WARN_ON(IB_GET_SEND_CREDITS(credits) >= 16384);
 
diff --git a/net/rds/loop.c b/net/rds/loop.c
index e6b0750bbedab..42e6b841b42c6 100644
--- a/net/rds/loop.c
+++ b/net/rds/loop.c
@@ -156,6 +156,28 @@ static int rds_loop_conn_alloc(struct rds_connection *conn, gfp_t gfp)
 	return 0;
 }
 
+/* Destroy the connections whose nodes were gathered on @tmp_list.
+ *
+ * rds_conn_destroy() can return before the connection is freed, and
+ * it is the free - rds_loop_conn_free() - that unlinks loop_node.
+ * @tmp_list lives on the caller's stack, so unlink each node before
+ * its destroy; the free then finds it empty and leaves it alone.
+ */
+static void rds_loop_destroy_gathered_conns(struct list_head *tmp_list)
+{
+	struct rds_loop_connection *lc, *_lc;
+
+	list_for_each_entry_safe(lc, _lc, tmp_list, loop_node) {
+		WARN_ON(rcu_access_pointer(lc->conn->c_passive));
+
+		spin_lock_irq(&loop_conns_lock);
+		list_del_init(&lc->loop_node);
+		spin_unlock_irq(&loop_conns_lock);
+
+		rds_conn_destroy(lc->conn);
+	}
+}
+
 static void rds_loop_conn_free(void *arg)
 {
 	struct rds_loop_connection *lc = arg;
@@ -163,7 +185,9 @@ static void rds_loop_conn_free(void *arg)
 
 	rdsdebug("lc %p\n", lc);
 	spin_lock_irqsave(&loop_conns_lock, flags);
-	list_del(&lc->loop_node);
+	/* already unlinked if a transport teardown gathered us first */
+	if (!list_empty(&lc->loop_node))
+		list_del(&lc->loop_node);
 	spin_unlock_irqrestore(&loop_conns_lock, flags);
 	kfree(lc);
 }
@@ -180,7 +204,6 @@ static void rds_loop_conn_path_shutdown(struct rds_conn_path *cp)
 
 void rds_loop_exit(void)
 {
-	struct rds_loop_connection *lc, *_lc;
 	LIST_HEAD(tmp_list);
 
 	rds_loop_set_unloading();
@@ -191,10 +214,9 @@ void rds_loop_exit(void)
 	INIT_LIST_HEAD(&loop_conns);
 	spin_unlock_irq(&loop_conns_lock);
 
-	list_for_each_entry_safe(lc, _lc, &tmp_list, loop_node) {
-		WARN_ON(lc->conn->c_passive);
-		rds_conn_destroy(lc->conn);
-	}
+	rds_loop_destroy_gathered_conns(&tmp_list);
+
+	rds_conn_wait_conns_freed(&rds_loop_transport, NULL);
 }
 
 static void rds_loop_kill_conns(struct net *net)
@@ -212,10 +234,7 @@ static void rds_loop_kill_conns(struct net *net)
 	}
 	spin_unlock_irq(&loop_conns_lock);
 
-	list_for_each_entry_safe(lc, _lc, &tmp_list, loop_node) {
-		WARN_ON(lc->conn->c_passive);
-		rds_conn_destroy(lc->conn);
-	}
+	rds_loop_destroy_gathered_conns(&tmp_list);
 }
 
 static void __net_exit rds_loop_exit_net(struct net *net)
diff --git a/net/rds/message.c b/net/rds/message.c
index f25f2592586f7..29e95028e61e4 100644
--- a/net/rds/message.c
+++ b/net/rds/message.c
@@ -182,6 +182,18 @@ static void rds_message_purge(struct rds_message *rm)
 		kref_put(&rm->atomic.op_rdma_mr->r_kref, __rds_put_mr_final);
 }
 
+static void rds_message_free(struct rds_message *rm)
+{
+	/* get in rds_send_queue_rm(), rds_send_probe() or the congestion
+	 * map path of rds_send_xmit().  Messages that were never queued on
+	 * a connection have no reference to drop.
+	 */
+	if (rm->m_inc.i_conn)
+		rds_conn_put(rm->m_inc.i_conn);
+
+	kfree(rm);
+}
+
 static void rds_message_unpin_worker(struct work_struct *work)
 {
 	struct rds_message *rm = container_of(work, struct rds_message,
@@ -192,7 +204,7 @@ static void rds_message_unpin_worker(struct work_struct *work)
 	if (rm->atomic.op_unpin_deferred)
 		rds_atomic_op_unpin_page(&rm->atomic);
 
-	kfree(rm);
+	rds_message_free(rm);
 }
 
 void rds_message_put(struct rds_message *rm)
@@ -217,7 +229,7 @@ void rds_message_put(struct rds_message *rm)
 			return;
 		}
 
-		kfree(rm);
+		rds_message_free(rm);
 	}
 }
 EXPORT_SYMBOL_GPL(rds_message_put);
diff --git a/net/rds/rdma_transport.c b/net/rds/rdma_transport.c
index b15cf316b23a2..584e9867810f4 100644
--- a/net/rds/rdma_transport.c
+++ b/net/rds/rdma_transport.c
@@ -63,6 +63,18 @@ static int rds_rdma_cm_event_handler_cmn(struct rdma_cm_id *cm_id,
 	if (cm_id->device->node_type == RDMA_NODE_IB_CA)
 		trans = &rds_ib_transport;
 
+	/* cm_id->context carries no reference of its own.  Pin the
+	 * connection for the duration of the handler: what the callbacks
+	 * below do may drop the last reference other than ours, and the
+	 * mutex released at out: lives in the connection's path array.
+	 * A connection already being freed gets no events handled.
+	 */
+	if (conn && !rds_conn_get_unless_zero(conn)) {
+		rdsdebug("conn %p id %p is being freed, ignoring event\n",
+			 conn, cm_id);
+		return 0;
+	}
+
 	/* Prevent shutdown from tearing down the connection
 	 * while we're executing. */
 	if (conn) {
@@ -171,8 +183,10 @@ static int rds_rdma_cm_event_handler_cmn(struct rdma_cm_id *cm_id,
 	}
 
 out:
-	if (conn)
+	if (conn) {
 		mutex_unlock(&conn->c_cm_lock);
+		rds_conn_put(conn);
+	}
 
 	rdsdebug("id %p event %u (%s) handling ret %d\n", cm_id, event->event,
 		 rdma_event_msg(event->event), ret);
diff --git a/net/rds/rds.h b/net/rds/rds.h
index 2db49573dacd5..06d48c2821efd 100644
--- a/net/rds/rds.h
+++ b/net/rds/rds.h
@@ -89,7 +89,6 @@ enum {
 #define RDS_RECONNECT_PENDING	1
 #define RDS_IN_XMIT		2
 #define RDS_RECV_REFILL		3
-#define	RDS_DESTROY_PENDING	4
 
 /* Max number of multipaths per RDS connection. Must be a power of 2 */
 #define	RDS_MPATH_WORKERS	8
@@ -138,6 +137,12 @@ struct rds_conn_path {
 /* One rds_connection per RDS address pair */
 struct rds_connection {
 	struct hlist_node	c_hash_node;
+	/* Free of the connection memory (not the teardown of its
+	 * transport state - that stays synchronous in
+	 * rds_conn_destroy()) is deferred until the last reference is
+	 * dropped via rds_conn_put().
+	 */
+	struct kref		c_refcount;
 	struct in6_addr		c_laddr;
 	struct in6_addr		c_faddr;
 	int			c_dev_if; /* ifindex used for this conn */
@@ -148,7 +153,15 @@ struct rds_connection {
 				c_pad_to_32:29;
 	int			c_npaths;
 	bool			c_with_sport_idx;
-	struct rds_connection	*c_passive;
+	/* Set once, by rds_conn_destroy(), before it cancels the path
+	 * works; read through rds_destroy_pending().  A site that arms
+	 * a path work must test the predicate and queue the work inside
+	 * one rcu_read_lock() section: the synchronize_rcu() that
+	 * follows the store is what keeps a queue issued after the
+	 * cancellation from landing on a destroyed workqueue.
+	 */
+	bool			c_destroy_in_prog;
+	struct rds_connection __rcu *c_passive;
 	struct rds_transport	*c_trans;
 
 	struct rds_cong_map	*c_lcong;
@@ -544,6 +557,12 @@ struct rds_transport {
 	unsigned int		t_prefer_loopback:1,
 				t_mp_capable:1;
 	unsigned int		t_type;
+	/* Connections of this transport not yet freed; freeing runs
+	 * asynchronously once rds_conn_destroy() has quiesced a
+	 * connection, so transport module unload has to wait for this
+	 * to reach zero (rds_conn_wait_conns_freed()).
+	 */
+	atomic_t		t_conn_count;
 
 	int (*laddr_check)(struct net *net, const struct in6_addr *addr,
 			   __u32 scope_id);
@@ -819,6 +838,20 @@ struct rds_connection *rds_conn_create_outgoing(struct net *net,
 						u8 tos, gfp_t gfp, int dev_if);
 void rds_conn_shutdown(struct rds_conn_path *cpath);
 void rds_conn_destroy(struct rds_connection *conn);
+void rds_conn_get(struct rds_connection *conn);
+void rds_conn_put(struct rds_connection *conn);
+/* take a reference unless the connection is already being freed */
+static inline bool rds_conn_get_unless_zero(struct rds_connection *conn)
+{
+	return kref_get_unless_zero(&conn->c_refcount);
+}
+/* transport unload waits for its connections to be freed, polling at
+ * the first interval and warning at the second
+ */
+#define RDS_CONN_FREE_POLL_MS		100
+#define RDS_CONN_FREE_WARN_INTERVAL_MS	10000
+void rds_conn_wait_conns_freed(struct rds_transport *trans,
+			       void (*resweep)(void));
 void rds_conn_drop(struct rds_connection *conn);
 void rds_conn_path_drop(struct rds_conn_path *cpath, bool destroy);
 void rds_conn_connect_if_down(struct rds_connection *conn);
@@ -994,7 +1027,8 @@ void __rds_put_mr_final(struct kref *kref);
 
 static inline bool rds_destroy_pending(struct rds_connection *conn)
 {
-	return !check_net(rds_conn_net(conn)) ||
+	return READ_ONCE(conn->c_destroy_in_prog) ||
+	       !check_net(rds_conn_net(conn)) ||
 	       (conn->c_trans->t_unloading && conn->c_trans->t_unloading(conn));
 }
 
diff --git a/net/rds/recv.c b/net/rds/recv.c
index 6204e577a90ae..b031c0b43af83 100644
--- a/net/rds/recv.c
+++ b/net/rds/recv.c
@@ -46,6 +46,7 @@ void rds_inc_init(struct rds_incoming *inc, struct rds_connection *conn,
 {
 	refcount_set(&inc->i_refcount, 1);
 	INIT_LIST_HEAD(&inc->i_item);
+	rds_conn_get(conn);	/* put in rds_inc_put() */
 	inc->i_conn = conn;
 	inc->i_conn_path = NULL;
 	inc->i_saddr = *saddr;
@@ -61,6 +62,7 @@ void rds_inc_path_init(struct rds_incoming *inc, struct rds_conn_path *cp,
 {
 	refcount_set(&inc->i_refcount, 1);
 	INIT_LIST_HEAD(&inc->i_item);
+	rds_conn_get(cp->cp_conn);	/* put in rds_inc_put() */
 	inc->i_conn = cp->cp_conn;
 	inc->i_conn_path = cp;
 	inc->i_saddr = *saddr;
@@ -81,9 +83,19 @@ void rds_inc_put(struct rds_incoming *inc)
 {
 	rdsdebug("put inc %p ref %d\n", inc, refcount_read(&inc->i_refcount));
 	if (refcount_dec_and_test(&inc->i_refcount)) {
+		struct rds_connection *conn = inc->i_conn;
+
 		BUG_ON(!list_empty(&inc->i_item));
 
-		inc->i_conn->c_trans->inc_free(inc);
+		/* inc_free() can free the memory @inc lives in, so the
+		 * connection reference has to be dropped through the
+		 * copy taken above.
+		 */
+		conn->c_trans->inc_free(inc);
+		/* get in rds_inc_init(), rds_inc_path_init() or
+		 * rds_recv_incoming()
+		 */
+		rds_conn_put(conn);
 	}
 }
 EXPORT_SYMBOL_GPL(rds_inc_put);
@@ -325,6 +337,13 @@ void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,
 	unsigned long flags;
 	struct rds_conn_path *cp;
 
+	/* every caller initialized @inc with rds_inc_init() or
+	 * rds_inc_path_init() first, so i_conn already holds a reference.
+	 * Take the new one before dropping the old, so that re-pointing an
+	 * inc at the connection it already refers to cannot free it.
+	 */
+	rds_conn_get(conn);
+	rds_conn_put(inc->i_conn);
 	inc->i_conn = conn;
 	inc->i_rx_jiffies = jiffies;
 	if (conn->c_trans->t_mp_capable)
diff --git a/net/rds/send.c b/net/rds/send.c
index 1afa981e5c06d..f59a487b079d5 100644
--- a/net/rds/send.c
+++ b/net/rds/send.c
@@ -290,6 +290,8 @@ int rds_send_xmit(struct rds_conn_path *cp)
 			}
 			rm->data.op_active = 1;
 			rm->m_inc.i_conn_path = cp;
+			/* put in rds_message_put() */
+			rds_conn_get(cp->cp_conn);
 			rm->m_inc.i_conn = cp->cp_conn;
 
 			cp->cp_xmit_rm = rm;
@@ -928,6 +930,19 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
 	 * and poll() now knows no more data can be sent.
 	 */
 	if (rs->rs_snd_bytes < rds_sk_sndbuf(rs)) {
+		/* rds_conn_path_quiesce() empties cp_send_queue under
+		 * cp_lock once the connection's destroy has begun.  Test
+		 * for that under the same lock, before touching either
+		 * queue: a message added after the purge would hold a
+		 * connection reference nothing ever drops.
+		 */
+		spin_lock(&cp->cp_lock);
+		if (rds_destroy_pending(conn)) {
+			spin_unlock(&cp->cp_lock);
+			*queued = -EAGAIN;
+			goto unlock;
+		}
+
 		rs->rs_snd_bytes += len;
 
 		/* let recv side know we are close to send space exhaustion.
@@ -947,11 +962,11 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
 		/* The code ordering is a little weird, but we're
 		   trying to minimize the time we hold c_lock */
 		rds_message_populate_header(&rm->m_inc.i_hdr, sport, dport, 0);
+		rds_conn_get(conn);	/* put in rds_message_put() */
 		rm->m_inc.i_conn = conn;
 		rm->m_inc.i_conn_path = cp;
 		rds_message_addref(rm);
 
-		spin_lock(&cp->cp_lock);
 		rm->m_inc.i_hdr.h_sequence = cpu_to_be64(cp->cp_next_tx_seq++);
 		list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
 		set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
@@ -964,6 +979,7 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,
 		*queued = 1;
 	}
 
+unlock:
 	spin_unlock_irqrestore(&rs->rs_lock, flags);
 out:
 	return *queued;
@@ -1159,13 +1175,14 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	DECLARE_SOCKADDR(struct sockaddr_in *, usin, msg->msg_name);
 	__be16 dport;
 	struct rds_message *rm = NULL;
-	struct rds_connection *conn;
+	struct rds_connection *conn = NULL;
 	int ret = 0;
 	int queued = 0, allocated_mr = 0;
 	int nonblock = msg->msg_flags & MSG_DONTWAIT;
 	long timeo = sock_sndtimeo(sk, nonblock);
 	struct rds_conn_path *cpath;
 	struct in6_addr daddr;
+	unsigned long flags;
 	__u32 scope_id = 0;
 	size_t rdma_payload_len = 0;
 	bool zcopy = ((msg->msg_flags & MSG_ZEROCOPY) &&
@@ -1340,11 +1357,29 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	rm->m_daddr = daddr;
 
 	/* rds_conn_create has a spinlock that runs with IRQ off.
-	 * Caching the conn in the socket helps a lot. */
-	if (rs->rs_conn && ipv6_addr_equal(&rs->rs_conn->c_faddr, &daddr) &&
-	    rs->rs_tos == rs->rs_conn->c_tos) {
-		conn = rs->rs_conn;
+	 * Caching the conn in the socket helps a lot.
+	 *
+	 * The cached rs_conn holds a connection reference; take one of
+	 * our own for the duration of this call (dropped on both exit
+	 * paths), so that neither a concurrent sender replacing the
+	 * cache nor rds_conn_destroy() can free the connection under
+	 * us.  A cached connection whose destruction has begun is not
+	 * reused: dropping it here lets the next sendmsg look up or
+	 * create a live one instead of returning -EAGAIN forever.
+	 */
+	spin_lock_irqsave(&rs->rs_lock, flags);
+	conn = rs->rs_conn;
+	if (conn && ipv6_addr_equal(&conn->c_faddr, &daddr) &&
+	    rs->rs_tos == conn->c_tos && !rds_destroy_pending(conn)) {
+		rds_conn_get(conn);
 	} else {
+		conn = NULL;
+	}
+	spin_unlock_irqrestore(&rs->rs_lock, flags);
+
+	if (!conn) {
+		struct rds_connection *old;
+
 		conn = rds_conn_create_outgoing(sock_net(sock->sk),
 						&rs->rs_bound_addr, &daddr,
 						rs->rs_transport, rs->rs_tos,
@@ -1352,9 +1387,17 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 						scope_id);
 		if (IS_ERR(conn)) {
 			ret = PTR_ERR(conn);
+			conn = NULL;
 			goto out;
 		}
+		/* hand the cache its own reference */
+		rds_conn_get(conn);
+		spin_lock_irqsave(&rs->rs_lock, flags);
+		old = rs->rs_conn;
 		rs->rs_conn = conn;
+		spin_unlock_irqrestore(&rs->rs_lock, flags);
+		if (old)
+			rds_conn_put(old);
 	}
 
 	if (conn->c_trans->t_mp_capable) {
@@ -1378,9 +1421,14 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 		 * outstanding.
 		 */
 		if (!test_and_set_bit(RDS_RECONNECT_PENDING,
-				      &conn->c_path[0].cp_flags))
-			queue_delayed_work(conn->c_path[0].cp_wq,
-					   &conn->c_path[0].cp_conn_w, 0);
+				      &conn->c_path[0].cp_flags)) {
+			rcu_read_lock();
+			if (!rds_destroy_pending(conn))
+				queue_delayed_work(conn->c_path[0].cp_wq,
+						   &conn->c_path[0].cp_conn_w,
+						   0);
+			rcu_read_unlock();
+		}
 		rds_send_ping(conn, 0);
 	}
 
@@ -1442,6 +1490,11 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 			ret = -ETIMEDOUT;
 		goto out;
 	}
+	/* rds_send_queue_rm() refused: the connection is being destroyed */
+	if (queued < 0) {
+		ret = queued;
+		goto out;
+	}
 
 	/*
 	 * By now we've committed to the send.  We reuse rds_send_worker()
@@ -1469,6 +1522,8 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 		kfree(vct.vec[ind].iov);
 	kfree(vct.vec);
 
+	rds_conn_put(conn);
+
 	return payload_len;
 
 out:
@@ -1476,6 +1531,9 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 		kfree(vct.vec[ind].iov);
 	kfree(vct.vec);
 
+	if (conn)
+		rds_conn_put(conn);
+
 	/* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
 	 * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
 	 * or in any other way, we need to destroy the MR again */
@@ -1522,6 +1580,7 @@ rds_send_probe(struct rds_conn_path *cp, __be16 sport,
 	list_add_tail(&rm->m_conn_item, &cp->cp_send_queue);
 	set_bit(RDS_MSG_ON_CONN, &rm->m_flags);
 	rds_message_addref(rm);
+	rds_conn_get(cp->cp_conn);	/* put in rds_message_put() */
 	rm->m_inc.i_conn = cp->cp_conn;
 	rm->m_inc.i_conn_path = cp;
 
diff --git a/net/rds/tcp.c b/net/rds/tcp.c
index 774a71f88d375..a71d6a4f0939c 100644
--- a/net/rds/tcp.c
+++ b/net/rds/tcp.c
@@ -502,6 +502,28 @@ static bool rds_tcp_is_unloading(struct rds_connection *conn)
 	return atomic_read(&rds_tcp_unloading) != 0;
 }
 
+/* Destroy the connections whose nodes were gathered on @tmp_list.
+ *
+ * rds_conn_destroy() can return before the connection is freed, and
+ * it is the free - rds_tcp_conn_free() - that unlinks t_tcp_node.
+ * Since @tmp_list lives on the caller's stack, unlink each node here
+ * and mark it detached before its destroy, so that a free that runs
+ * after the caller has returned does not write into a dead frame.
+ */
+static void rds_tcp_destroy_gathered_conns(struct list_head *tmp_list)
+{
+	struct rds_tcp_connection *tc, *_tc;
+
+	list_for_each_entry_safe(tc, _tc, tmp_list, t_tcp_node) {
+		spin_lock_irq(&rds_tcp_conn_lock);
+		list_del_init(&tc->t_tcp_node);
+		tc->t_tcp_node_detached = true;
+		spin_unlock_irq(&rds_tcp_conn_lock);
+
+		rds_conn_destroy(tc->t_cpath->cp_conn);
+	}
+}
+
 static void rds_tcp_destroy_conns(void)
 {
 	struct rds_tcp_connection *tc, *_tc;
@@ -515,8 +537,7 @@ static void rds_tcp_destroy_conns(void)
 	}
 	spin_unlock_irq(&rds_tcp_conn_lock);
 
-	list_for_each_entry_safe(tc, _tc, &tmp_list, t_tcp_node)
-		rds_conn_destroy(tc->t_cpath->cp_conn);
+	rds_tcp_destroy_gathered_conns(&tmp_list);
 }
 
 static void rds_tcp_exit(void);
@@ -698,8 +719,7 @@ static void rds_tcp_kill_sock(struct net *net)
 		}
 	}
 	spin_unlock_irq(&rds_tcp_conn_lock);
-	list_for_each_entry_safe(tc, _tc, &tmp_list, t_tcp_node)
-		rds_conn_destroy(tc->t_cpath->cp_conn);
+	rds_tcp_destroy_gathered_conns(&tmp_list);
 }
 
 static void __net_exit rds_tcp_exit_net(struct net *net)
@@ -805,6 +825,7 @@ static void rds_tcp_exit(void)
 #endif
 	unregister_pernet_device(&rds_tcp_net_ops);
 	rds_tcp_destroy_conns();
+	rds_conn_wait_conns_freed(&rds_tcp_transport, NULL);
 	rds_trans_unregister(&rds_tcp_transport);
 	rds_tcp_recv_exit();
 	kmem_cache_destroy(rds_tcp_conn_slab);
diff --git a/net/rds/tcp_listen.c b/net/rds/tcp_listen.c
index 13fa60c1985bb..e22ea9ca8c1ce 100644
--- a/net/rds/tcp_listen.c
+++ b/net/rds/tcp_listen.c
@@ -153,7 +153,7 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 {
 	struct socket *listen_sock = rtn->rds_tcp_listen_sock;
 	struct socket *new_sock = NULL;
-	struct rds_connection *conn;
+	struct rds_connection *conn = NULL;
 	int ret;
 	struct inet_sock *inet;
 	struct rds_tcp_connection *rs_tcp = NULL;
@@ -229,6 +229,7 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 
 	if (IS_ERR(conn)) {
 		ret = PTR_ERR(conn);
+		conn = NULL;
 		goto out;
 	}
 	/* An incoming SYN request came in, and TCP just accepted it.
@@ -277,7 +278,15 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 	cp = rs_tcp->t_cpath;
 	conn_state = rds_conn_path_state(cp);
 	WARN_ON(conn_state == RDS_CONN_UP);
-	if (conn_state != RDS_CONN_CONNECTING && conn_state != RDS_CONN_ERROR) {
+	/* A connection whose destroy has begun has been quiesced and is
+	 * only waiting for its last reference: its paths sit in
+	 * RDS_CONN_DOWN, which rds_tcp_accept_one_path() happily claims.
+	 * Installing a socket on it would leave sk_user_data pointing
+	 * at a path that is about to be freed.
+	 */
+	if (rds_destroy_pending(conn) ||
+	    (conn_state != RDS_CONN_CONNECTING &&
+	     conn_state != RDS_CONN_ERROR)) {
 		rds_conn_path_drop(cp, 0);
 		goto rst_nsk;
 	}
@@ -316,10 +325,14 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 	 */
 	if (READ_ONCE(sk->sk_state) == TCP_CLOSE_WAIT ||
 	    READ_ONCE(sk->sk_state) == TCP_LAST_ACK ||
-	    READ_ONCE(sk->sk_state) == TCP_CLOSE)
+	    READ_ONCE(sk->sk_state) == TCP_CLOSE) {
 		rds_conn_path_drop(cp, 0);
-	else
-		queue_delayed_work(cp->cp_wq, &cp->cp_recv_w, 0);
+	} else {
+		rcu_read_lock();
+		if (!rds_destroy_pending(cp->cp_conn))
+			queue_delayed_work(cp->cp_wq, &cp->cp_recv_w, 0);
+		rcu_read_unlock();
+	}
 
 	sock_put(sk);
 
@@ -343,6 +356,8 @@ int rds_tcp_accept_one(struct rds_tcp_net *rtn)
 		mutex_unlock(&rs_tcp->t_conn_path_lock);
 	if (new_sock)
 		sock_release(new_sock);
+	if (conn)
+		rds_conn_put(conn);
 
 	mutex_unlock(&rtn->rds_tcp_accept_lock);
 

17/3 2026/09/14 04:29 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 20612
							output: 19
							thoughts: 1211
						

Thoughts:
**KMSAN Evaluation for `net/rds/` Patch Series**

Okay, I need to determine if this patch series in `net/rds/` necessitates a dedicated KMSAN session. My primary goal is to assess whether it introduces or exposes uninitialized memory risks that *only* KMSAN would detect, as per the established guidelines. Standard KASAN, LOCKDEP, and other bug detectors should catch anything else. I'm going to carefully dissect each file changed.

First, `net/rds/af_rds.c`. The changes to `rds_release()` seem straightforward – just dropping the connection reference, nothing that screams uninitialized memory. Then, `rds_ioctl()`. The `SIOCRDSGETTOS` case caught my eye.  The code reads `rs->rs_tos`. My initial thought was: is `rs->rs_tos` initialized?  I need to check the allocation and initialization of `rds_sock` to be sure. Alright, I've checked the before and after states of `rds_ioctl`. The code *already* used `rs->rs_tos` and put it to user space.  Also, `tos` is initialized on the stack, and the `rds_sock` structure is zeroed out by `sk_alloc`. So, no immediate KMSAN concerns here. The patch simply modifies locking, which is unrelated to this concern.

Next, `net/rds/connection.c`. This file seems focused on connection lifetime management through krefs and RCU.  Introduced refcounting, proper quiescing and workqueue synchronization, and all that. It's refactoring for better resource management, which makes sense. I see no obvious spots for uninitialized reads. It's all about ensuring that references are handled correctly with `rds_conn_get()`, `rds_conn_put()`, and friends, and that's generally good practice.

Finally, a slew of files: `net/rds/ib.c`, `net/rds/ib_cm.c`, `net/rds/ib_rdma.c`, etc. They are clearly addressing issues with refcounting to prevent use-after-free and race conditions during teardown, unload, disconnect, and destroy. It's guarding workqueue queuing and handling list unlinking safely. All of this is about preventing memory corruption bugs, but not necessarily uninitialized memory reads. This patch series is all about preventing *already* initialized memory from being misused or corrupted via various races and UAF issues.

I'm now going to carefully read the patch changes, and if there is a commit log, I will check it. This might reveal the bug it is addressing, and will confirm if it is something KASAN or LOCKDEP could catch, as KMSAN should only be used in specific cases where it is needed. I'll also use grep, or other tools to carefully investigate these functions, and confirm my initial assumptions.

Based on my initial inspection, I do not see any code that introduces *new* uninitialized memory risks. It looks like the patch is preventing existing bugs, not introducing new ones.



18/3 2026/09/14 04:29 tool grepper
0m Args:
{
  "Expression": "t_conn_count"
}

Results:
{
  "Output": "net/rds/connection.c=49=static DEFINE_SPINLOCK(rds_conn_lock);\nnet/rds/connection.c:50:/* woken whenever a transport's t_conn_count drops to zero */\nnet/rds/connection.c-51-static DECLARE_WAIT_QUEUE_HEAD(rds_conn_freed_waitq);\n--\nnet/rds/connection.c=211=static struct rds_connection *__rds_conn_create(struct net *net,\n--\nnet/rds/connection.c-333-\tfor (i = 0; i \u003c npaths; i++) {\nnet/rds/connection.c:334:\t\tint seq = atomic_read(\u0026trans-\u003et_conn_count);\nnet/rds/connection.c-335-\n--\nnet/rds/connection.c-410-\t\t\trds_cong_add_conn(conn);\nnet/rds/connection.c:411:\t\t\tatomic_inc(\u0026conn-\u003ec_trans-\u003et_conn_count);\nnet/rds/connection.c-412-\t\t}\n--\nnet/rds/connection.c-432-\t\t\trds_cong_add_conn(conn);\nnet/rds/connection.c:433:\t\t\tatomic_inc(\u0026conn-\u003ec_trans-\u003et_conn_count);\nnet/rds/connection.c-434-\t\t}\n--\nnet/rds/connection.c=664=static void rds_conn_destroy_fini(struct kref *kref)\n--\nnet/rds/connection.c-680-\t */\nnet/rds/connection.c:681:\tif (!atomic_dec_return(\u0026trans-\u003et_conn_count))\nnet/rds/connection.c-682-\t\twake_up_all(\u0026rds_conn_freed_waitq);\n--\nnet/rds/connection.c=697=void rds_conn_wait_conns_freed(struct rds_transport *trans,\n--\nnet/rds/connection.c-704-\twhile (!wait_event_timeout(rds_conn_freed_waitq,\nnet/rds/connection.c:705:\t\t\t\t   !atomic_read(\u0026trans-\u003et_conn_count),\nnet/rds/connection.c-706-\t\t\t\t   msecs_to_jiffies(RDS_CONN_FREE_POLL_MS))) {\n--\nnet/rds/connection.c-715-\t\t\t\ttrans-\u003et_name,\nnet/rds/connection.c:716:\t\t\t\tatomic_read(\u0026trans-\u003et_conn_count));\nnet/rds/connection.c-717-\t\t\twarn_at = jiffies + warn_interval;\n--\nnet/rds/rds.h=553=struct rds_transport {\n--\nnet/rds/rds.h-564-\t */\nnet/rds/rds.h:565:\tatomic_t\t\tt_conn_count;\nnet/rds/rds.h-566-\n"
}

19/3 2026/09/14 04:29 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 23238
							output: 35
							thoughts: 29
						

Thoughts:
Here's my take on the thought process, as if it were my own:

**Investigating a Network Connection Issue in RDS**

Okay, time to dive in. I need to pinpoint where the connection object (`conn`) is being allocated within the Remote Direct Memory Access over TCP (RDS) network stack. That usually means starting with the connection creation process. Let's head straight to the source code: `net/rds/connection.c`. Specifically, I need to zero in on the `__rds_conn_create` function. That function is the likely candidate for allocating the `conn` struct. Understanding the allocation process is fundamental to tracking down potential memory leaks, connection problems, or issues during initialization, so this is the logical place to start digging.



20/3 2026/09/14 04:29 tool read-file
0m Args:
{
  "File": "net/rds/connection.c",
  "FirstLine": 250,
  "LineCount": 50
}

Results:
{
  "Contents": " 250:\t\t\t\tconn = NULL;\n 251:\t\t\tif (conn)\n 252:\t\t\t\trds_conn_get(conn);\n 253:\t\t}\n 254:\t\trcu_read_unlock();\n 255:\t\tif (conn)\n 256:\t\t\tgoto out;\n 257:\t\n 258:\t\tconn = kmem_cache_zalloc(rds_conn_slab, gfp);\n 259:\t\tif (!conn) {\n 260:\t\t\tconn = ERR_PTR(-ENOMEM);\n 261:\t\t\tgoto out;\n 262:\t\t}\n 263:\t\tconn-\u003ec_path = kzalloc_objs(struct rds_conn_path, npaths, gfp);\n 264:\t\tif (!conn-\u003ec_path) {\n 265:\t\t\tkmem_cache_free(rds_conn_slab, conn);\n 266:\t\t\tconn = ERR_PTR(-ENOMEM);\n 267:\t\t\tgoto out;\n 268:\t\t}\n 269:\t\n 270:\t\tkref_init(\u0026conn-\u003ec_refcount);\n 271:\t\tINIT_HLIST_NODE(\u0026conn-\u003ec_hash_node);\n 272:\t\tconn-\u003ec_laddr = *laddr;\n 273:\t\tconn-\u003ec_isv6 = !ipv6_addr_v4mapped(laddr);\n 274:\t\tconn-\u003ec_faddr = *faddr;\n 275:\t\tconn-\u003ec_dev_if = dev_if;\n 276:\t\tconn-\u003ec_tos = tos;\n 277:\t\n 278:\t#if IS_ENABLED(CONFIG_IPV6)\n 279:\t\t/* If the local address is link local, set c_bound_if to be the\n 280:\t\t * index used for this connection.  Otherwise, set it to 0 as\n 281:\t\t * the socket is not bound to an interface.  c_bound_if is used\n 282:\t\t * to look up a socket when a packet is received\n 283:\t\t */\n 284:\t\tif (ipv6_addr_type(laddr) \u0026 IPV6_ADDR_LINKLOCAL)\n 285:\t\t\tconn-\u003ec_bound_if = dev_if;\n 286:\t\telse\n 287:\t#endif\n 288:\t\t\tconn-\u003ec_bound_if = 0;\n 289:\t\n 290:\t\trds_conn_net_set(conn, net);\n 291:\t\n 292:\t\tret = rds_cong_get_maps(conn);\n 293:\t\tif (ret) {\n 294:\t\t\tkfree(conn-\u003ec_path);\n 295:\t\t\tkmem_cache_free(rds_conn_slab, conn);\n 296:\t\t\tconn = ERR_PTR(ret);\n 297:\t\t\tgoto out;\n 298:\t\t}\n 299:\t\n"
}

21/3 2026/09/14 04:29 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 24234
							output: 21
							thoughts: 385
						

Thoughts:
**Initial Code Review: Memory Allocation and Field Initialization**

Okay, so I'm looking at this patch and immediately, I'm noting the memory allocation. The `conn` struct is allocated with `kmem_cache_zalloc`, and the `c_path` member is allocated using `kzalloc_objs`. This is good; we're ensuring zero-initialization which is critical for preventing memory leaks and undefined behavior, especially in a network context. I always appreciate this kind of diligence.  I also see that any new fields added to the `rds_connection` and `rds_transport` structs are also properly initialized. This is exactly what I would expect, and confirms that the author is likely thinking about concurrent access. I’m satisfied that the basics of memory management are sound here.

Now, let's dive into the specifics of the changes and see what else they've touched.

First up: `af_rds.c`

The `rds_release` function is simple: it decrements the connection's reference count. Standard stuff. Nothing unusual here.

Then, there's `rds_ioctl`. Ah, here's where things get interesting. The change of lock from `rds_sock_lock` to the socket's individual lock (`rs->rs_lock`) for `SIOCRDSSETTOS` and `SIOCRDSGETTOS` catches my eye. This is probably a good move; a per-socket lock should improve concurrency, as opposed to a global lock, assuming the old implementation used `rds_sock_lock` everywhere.

In `SIOCRDSGETTOS`, the `tos` variable is initialized to 0 at declaration, before being populated from `rs->rs_tos` under lock protection, and then passed back to the user via `put_user`. The locking seems correct, and the use of the `put_user` function here indicates that the data is being sent to userspace safely. They are using an `rds_tos_t` type, so I should double-check that this is a safe, single-byte or integral type - let me go look up its definition quickly... yes, it looks fine. Good! That's a reasonable change.



22/3 2026/09/14 04:29 tool grepper
0m Args:
{
  "Expression": "typedef .*rds_tos_t"
}

Results:
{
  "Output": "include/uapi/linux/rds.h-77-\ninclude/uapi/linux/rds.h:78:typedef __u8\trds_tos_t;\ninclude/uapi/linux/rds.h-79-\n"
}

23/3 2026/09/14 04:29 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 25178
							output: 214
							thoughts: 825
						

Thoughts:
**KMSAN Assessment: RDS Connection Refcounting Patch**

Okay, let's break this down. `rds_tos_t` is a simple `__u8`, completely initialized, so that's a non-issue. The core of this patch is refactoring the `struct rds_connection` lifecycle management. It's moving from direct deallocation to a refcounted approach with `kref`, `rds_conn_get()`, and `rds_conn_put()`. This is all about fixing classic memory safety issues: use-after-free, double frees, race conditions during teardown, reconnects, unloads, and concurrent operations. Things that KASAN and LOCKDEP are *built* to catch, like use-after-free, deadlocks, and null dereferences.

Now, let's be thorough. I need to make sure this doesn't *introduce* any new vulnerabilities, especially around user-space interaction. Scanning `af_rds.c`, the only `put_user` call already existed for `tos`, and the only change was to the locking – from `rds_sock_lock` to `rs->rs_lock`. No new syscalls, ioctls, socket options, or netlink attributes. Nothing that touches packet parsing lengths or calculations that could lead to out-of-bounds reads or uninitialized memory exposure. I've checked for uninitialized variables and structs – there aren't any. All heap allocations are zeroed (`kzalloc`, `kmem_cache_zalloc`), and all stack variables in the modified code are either initialized directly or used correctly within macros like `spin_lock_irqsave` or are assigned values before use.

The whole patch is about refcounting, concurrency control with locks, RCU, waitqueues, atomic counters, and lifecycle management to fix race conditions and use-after-free issues. Those are precisely the kinds of bugs KASAN and LOCKDEP are designed to find!

So, the conclusion is clear: `NeedsKMSAN=false`.

Now, let's write a clear, contrastive reasoning for the `KMSANReasoning` field, highlighting why KMSAN isn't necessary here:

*   This patch focuses on connection lifecycle and race conditions in RDS, specifically refcounting `struct rds_connection` using `kref`, ensuring safe teardown via RCU, workqueue flushing, and synchronized locking in `af_rds`, `connection`, `ib`, `loop`, `tcp`, `send`, and `recv`.
*   The bugs this patch addresses, and the regressions it *could* introduce, are use-after-free, race conditions, double frees, invalid frees, and locking deadlocks. These are all caught by KASAN and LOCKDEP, which are standard debugging tools.
*   The patch doesn't introduce any new uninitialized memory reads or allocations. All structures are zeroed during allocation, all stack variables are correctly initialized, and there are no copies to user-space that expose uninitialized fields. The existing `put_user` of `rds_tos_t` (a single byte) is properly locked.
*   Therefore, dedicated KMSAN fuzzing isn't required here; standard KASAN/LOCKDEP builds are sufficient.

Okay, that's solid. Let's call `set-results`.



24/3 2026/09/14 04:30 tool set-results
0m Args:
{
  "KMSANReasoning": "The patch introduces reference counting (kref) and synchronization fixes for RDS connections to prevent use-after-free, races, and premature frees during connection teardown and module unloading. All newly added or modified data structures (e.g. struct rds_connection, rds_conn_path) continue to be allocated using zero-initializing allocators (kmem_cache_zalloc, kzalloc_objs), and all local variables are properly initialized. The only user-copy path touched is SIOCRDSGETTOS, which merely updates the spinlock protecting rs_tos (a single __u8 value). The risks associated with these changes are strictly related to object lifetimes, concurrent access, reference leaks, or locking deadlocks, which are effectively covered by KASAN and LOCKDEP. No uninitialized memory reads or information leak hazards are introduced or exposed, so a dedicated KMSAN fuzzing session is not warranted.",
  "NeedsKMSAN": false
}

Results:
{
  "KMSANReasoning": "The patch introduces reference counting (kref) and synchronization fixes for RDS connections to prevent use-after-free, races, and premature frees during connection teardown and module unloading. All newly added or modified data structures (e.g. struct rds_connection, rds_conn_path) continue to be allocated using zero-initializing allocators (kmem_cache_zalloc, kzalloc_objs), and all local variables are properly initialized. The only user-copy path touched is SIOCRDSGETTOS, which merely updates the spinlock protecting rs_tos (a single __u8 value). The risks associated with these changes are strictly related to object lifetimes, concurrent access, reference leaks, or locking deadlocks, which are effectively covered by KASAN and LOCKDEP. No uninitialized memory reads or information leak hazards are introduced or exposed, so a dedicated KMSAN fuzzing session is not warranted.",
  "NeedsKMSAN": false
}

LLM Calls Summary:
Total Calls Total Tokens Avg Tokens Total Duration (Seconds) Avg Duration (Seconds)
Tool Calls Summary:
Total Calls Total Duration (Seconds) Avg Duration (Seconds)