AI Trajectory


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

Results:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_sendmsg",
    "rds_release",
    "rds_ioctl",
    "rds_send_queue_rm",
    "rds_recv_incoming",
    "rds_tcp_accept_one"
  ],
  "KMSANReasoning": "The patch introduces reference counting (kref) for 'struct rds_connection' to manage connection lifetimes safely and prevent use-after-free and concurrency races during connection teardown, module unload, and asynchronous operations (across RDS core, TCP, Loop, and IB transports).\n\nKASAN vs KMSAN applicability:\n- The primary failure modes and risks associated with these changes are use-after-free, double-free, reference leaks, and locking issues/deadlocks. These are covered by KASAN, KASAN-quarantine, and LOCKDEP in standard debug builds.\n- All touched structures (including newly added fields such as 'c_refcount', 'c_destroy_in_prog', 'i_ib_node_detached', and 't_tcp_node_detached') are allocated using zeroing allocators (kmem_cache_zalloc, kzalloc_obj) or are static/atomic variables initialized to zero.\n- No uninitialized stack/heap memory or structure padding is read or exposed across security boundaries (e.g. copied to userspace or passed uninitialized to kernel subsystems).\n\nConsequently, there are no uninitialized memory hazards that would necessitate a dedicated KMSAN fuzzing session.",
  "NeedsKMSAN": false,
  "Reasoning": "The patch refactors connection lifecycle and reference counting in the RDS subsystem (net/rds). It introduces kref-based reference counting on rds_connection, alters locking schemes in rds_ioctl and rds_sendmsg, manages cached socket connection references, and adjusts connection teardown across core RDS, loopback, TCP, and IB transports. These code paths are directly reachable via standard AF_RDS socket operations and can introduce subtle concurrency bugs, refcount imbalances, or use-after-free conditions under fuzzing.",
  "WorthFuzzing": true
}

1/1 2026/09/21 06:14 action read-patch-diff
0m Args:
null

Results:
{
  "PatchDiff": "commit 7e7965e091176ceeb5b5cdfd3144694b6bfcb59d\nAuthor: syz-cluster \u003ctriage@syzkaller.com\u003e\nDate:   Mon Sep 21 06:14:23 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..546961bb2e127 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,10 @@ 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 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 +100,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 +176,30 @@ static void __rds_conn_path_init(struct rds_connection *conn,\n \tcp-\u003ecp_flags = 0;\n }\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+/* 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,97 @@ 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 - once incs hold one, an unread datagram\n+ * pins the inc that carries it, and thus the connection - so the wait\n+ * 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 +755,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 +826,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.h b/net/rds/ib.h\nindex 5ff346a1e8baa..cb410c3ae8d85 100644\n--- a/net/rds/ib.h\n+++ b/net/rds/ib.h\n@@ -143,6 +143,10 @@ struct rds_ib_device;\n struct rds_ib_connection {\n \n \tstruct list_head\tib_node;\n+\t/* set under ib_nodev_conns_lock once a transport teardown has\n+\t * claimed ib_node; from then on only the teardown touches it\n+\t */\n+\tbool\t\t\ti_ib_node_detached;\n \tstruct rds_ib_device\t*rds_ibdev;\n \tstruct rds_connection\t*conn;\n \ndiff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c\nindex 4feb0edc360c8..cbeedfc4fabd0 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,15 @@ 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/* Drop the reference rds_conn_create() handed us.  The\n+\t\t * conn stays reachable through cm_id-\u003econtext without a\n+\t\t * reference of its own for now; the CM event handler is\n+\t\t * given one of its own by a following patch.\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 +1285,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 +1293,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/* a transport teardown that gathered us first owns the node */\n+\tif (!ic-\u003ei_ib_node_detached)\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..1548e5be0e559 100644\n--- a/net/rds/ib_rdma.c\n+++ b/net/rds/ib_rdma.c\n@@ -123,15 +123,18 @@ void rds_ib_add_conn(struct rds_ib_device *rds_ibdev, struct rds_connection *con\n {\n \tstruct rds_ib_connection *ic = conn-\u003ec_transport_data;\n \n-\t/* conn was previously on the nodev_conns_list */\n+\t/* conn was previously on the nodev_conns_list, unless a teardown\n+\t * sweep has claimed it ahead of destroying it: then it is on its\n+\t * way out, and its node belongs to the sweep.\n+\t */\n \tspin_lock_irq(\u0026ib_nodev_conns_lock);\n-\tBUG_ON(list_empty(\u0026ib_nodev_conns));\n-\tBUG_ON(list_empty(\u0026ic-\u003eib_node));\n-\tlist_del(\u0026ic-\u003eib_node);\n+\tif (!ic-\u003ei_ib_node_detached) {\n+\t\tlist_del(\u0026ic-\u003eib_node);\n \n-\tspin_lock(\u0026rds_ibdev-\u003espinlock);\n-\tlist_add_tail(\u0026ic-\u003eib_node, \u0026rds_ibdev-\u003econn_list);\n-\tspin_unlock(\u0026rds_ibdev-\u003espinlock);\n+\t\tspin_lock(\u0026rds_ibdev-\u003espinlock);\n+\t\tlist_add_tail(\u0026ic-\u003eib_node, \u0026rds_ibdev-\u003econn_list);\n+\t\tspin_unlock(\u0026rds_ibdev-\u003espinlock);\n+\t}\n \tspin_unlock_irq(\u0026ib_nodev_conns_lock);\n \n \tic-\u003erds_ibdev = rds_ibdev;\n@@ -142,15 +145,22 @@ void rds_ib_remove_conn(struct rds_ib_device *rds_ibdev, struct rds_connection *\n {\n \tstruct rds_ib_connection *ic = conn-\u003ec_transport_data;\n \n-\t/* place conn on nodev_conns_list */\n+\tbool detached;\n+\n+\t/* place conn on nodev_conns_list - unless a teardown sweep has\n+\t * claimed it ahead of destroying it, in which case its node\n+\t * belongs to the sweep\n+\t */\n \tspin_lock(\u0026ib_nodev_conns_lock);\n \n \tspin_lock_irq(\u0026rds_ibdev-\u003espinlock);\n-\tBUG_ON(list_empty(\u0026ic-\u003eib_node));\n-\tlist_del(\u0026ic-\u003eib_node);\n+\tdetached = ic-\u003ei_ib_node_detached;\n+\tif (!detached)\n+\t\tlist_del(\u0026ic-\u003eib_node);\n \tspin_unlock_irq(\u0026rds_ibdev-\u003espinlock);\n \n-\tlist_add_tail(\u0026ic-\u003eib_node, \u0026ib_nodev_conns);\n+\tif (!detached)\n+\t\tlist_add_tail(\u0026ic-\u003eib_node, \u0026ib_nodev_conns);\n \n \tspin_unlock(\u0026ib_nodev_conns_lock);\n \n@@ -163,13 +173,40 @@ void rds_ib_destroy_nodev_conns(void)\n \tstruct rds_ib_connection *ic, *_ic;\n \tLIST_HEAD(tmp_list);\n \n-\t/* avoid calling conn_destroy with irqs off */\n+\tstruct rds_connection *conn;\n+\n+\t/* Gather the connections and take a reference on each, so that\n+\t * none is freed under the walk below (a connection destroyed\n+\t * earlier, for a protocol version mismatch, can be on this list\n+\t * with only a socket's reference still pending).  One whose free\n+\t * is already running gets no reference: its free unlinks the\n+\t * node itself, under this lock, once we drop it.  Marking the\n+\t * node detached claims it for this sweep: rds_ib_add_conn(),\n+\t * rds_ib_remove_conn() and rds_ib_conn_free() leave a claimed\n+\t * node alone, so the walk over tmp_list below needs no lock.\n+\t * Avoid calling conn_destroy with irqs off.\n+\t */\n \tspin_lock_irq(\u0026ib_nodev_conns_lock);\n-\tlist_splice(\u0026ib_nodev_conns, \u0026tmp_list);\n+\tlist_for_each_entry_safe(ic, _ic, \u0026ib_nodev_conns, ib_node) {\n+\t\tif (rds_conn_get_unless_zero(ic-\u003econn)) {\n+\t\t\tic-\u003ei_ib_node_detached = true;\n+\t\t\tlist_move_tail(\u0026ic-\u003eib_node, \u0026tmp_list);\n+\t\t}\n+\t}\n \tspin_unlock_irq(\u0026ib_nodev_conns_lock);\n \n-\tlist_for_each_entry_safe(ic, _ic, \u0026tmp_list, ib_node)\n-\t\trds_conn_destroy(ic-\u003econn);\n+\t/* rds_conn_destroy() can return before the connection is freed,\n+\t * and it is the free - rds_ib_conn_free() - that would unlink\n+\t * ib_node.  tmp_list lives on this stack frame, so take each node\n+\t * off it before its destroy; the free then leaves it alone.\n+\t */\n+\tlist_for_each_entry_safe(ic, _ic, \u0026tmp_list, ib_node) {\n+\t\tconn = ic-\u003econn;\n+\t\tlist_del_init(\u0026ic-\u003eib_node);\n+\n+\t\trds_conn_destroy(conn);\n+\t\trds_conn_put(conn);\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..5bac858df87ef 100644\n--- a/net/rds/loop.c\n+++ b/net/rds/loop.c\n@@ -156,6 +156,45 @@ 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+\tstruct rds_connection *conn;\n+\n+\tlist_for_each_entry_safe(lc, _lc, tmp_list, loop_node) {\n+\t\tconn = lc-\u003econn;\n+\t\tWARN_ON(rcu_access_pointer(conn-\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(conn);\n+\t\trds_conn_put(conn);\n+\t}\n+}\n+\n+/* Gather @lc's connection for destruction: move the node to the\n+ * caller's @tmp_list and take a reference that keeps the connection,\n+ * and so the node, alive until rds_loop_destroy_gathered_conns() has\n+ * dealt with it.  Called with loop_conns_lock held.  A connection\n+ * whose free is already running gets no reference; its free unlinks\n+ * the node itself, under the same lock, once we drop it.\n+ */\n+static void rds_loop_gather_conn(struct rds_loop_connection *lc,\n+\t\t\t\t struct list_head *tmp_list)\n+{\n+\tif (rds_conn_get_unless_zero(lc-\u003econn))\n+\t\tlist_move_tail(\u0026lc-\u003eloop_node, tmp_list);\n+}\n+\n static void rds_loop_conn_free(void *arg)\n {\n \tstruct rds_loop_connection *lc = arg;\n@@ -163,7 +202,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@@ -187,14 +228,13 @@ void rds_loop_exit(void)\n \tsynchronize_rcu();\n \t/* avoid calling conn_destroy with irqs off */\n \tspin_lock_irq(\u0026loop_conns_lock);\n-\tlist_splice(\u0026loop_conns, \u0026tmp_list);\n-\tINIT_LIST_HEAD(\u0026loop_conns);\n+\tlist_for_each_entry_safe(lc, _lc, \u0026loop_conns, loop_node)\n+\t\trds_loop_gather_conn(lc, \u0026tmp_list);\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@@ -208,14 +248,11 @@ static void rds_loop_kill_conns(struct net *net)\n \n \t\tif (net != c_net)\n \t\t\tcontinue;\n-\t\tlist_move_tail(\u0026lc-\u003eloop_node, \u0026tmp_list);\n+\t\trds_loop_gather_conn(lc, \u0026tmp_list);\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 47d5e9ab9b109..cea48ad745ca7 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..321f2da9e76d9 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/* rds_conn_destroy() quiesces the connection synchronously;\n+\t * freeing it - the connection memory, the path workqueues and\n+\t * the transport's per-connection state - is deferred until the\n+\t * last reference is 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@@ -650,7 +669,9 @@ struct rds_sock {\n \n \t/*\n \t * rds_sendmsg caches the conn it used the last time around.\n-\t * This helps avoid costly lookups.\n+\t * This helps avoid costly lookups.  The cache owns a connection\n+\t * reference, dropped when it is replaced or the socket is\n+\t * released, and is read and written under rs_lock.\n \t */\n \tstruct rds_connection\t*rs_conn;\n \n@@ -819,6 +840,21 @@ 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+\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 +1030,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..1fcd4483d3be4 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)\n@@ -406,8 +425,9 @@ void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,\n \t * rds_find_bound() uses a global (netns-agnostic) hash table.\n \t * An RDS connection created in netns A can match a socket bound\n \t * in the init netns, delivering inc cross-netns with inc-\u003ei_conn\n-\t * pointing into netns A.  When cleanup_net() then frees that conn,\n-\t * any subsequent dereference of inc-\u003ei_conn is a use-after-free.\n+\t * pointing into netns A.  Cross-netns delivery is wrong on its\n+\t * own, and the inc's connection reference would keep a\n+\t * connection of a dead netns around, with a stale c_net.\n \t * Drop the inc if the receiving socket lives in a different netns.\n \t */\n \tif (!net_eq(sock_net(rds_rs_to_sk(rs)), rds_conn_net(conn))) {\ndiff --git a/net/rds/send.c b/net/rds/send.c\nindex 1afa981e5c06d..afb22779ff977 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,28 @@ 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/* rs_tos was sampled without rs_lock for the create above,\n+\t\t * and SIOCRDSSETTOS only refuses a change once rs_conn is\n+\t\t * set, so it can have changed underneath us.  Do not\n+\t\t * install - or send on - a connection whose ToS no longer\n+\t\t * matches the socket's; the retry uses the new one.\n+\t\t */\n+\t\tspin_lock_irqsave(\u0026rs-\u003ers_lock, flags);\n+\t\tif (conn-\u003ec_tos != rs-\u003ers_tos) {\n+\t\t\tspin_unlock_irqrestore(\u0026rs-\u003ers_lock, flags);\n+\t\t\tret = -EAGAIN;\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\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 +1432,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 +1501,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 +1533,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 +1542,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@@ -1519,9 +1588,18 @@ rds_send_probe(struct rds_conn_path *cp, __be16 sport,\n \t\tgoto out;\n \n \tspin_lock_irqsave(\u0026cp-\u003ecp_lock, flags);\n+\t/* Same rule as rds_send_queue_rm(): once the destroy has purged\n+\t * cp_send_queue under this lock, nothing may be added behind it.\n+\t */\n+\tif (rds_destroy_pending(cp-\u003ecp_conn)) {\n+\t\tspin_unlock_irqrestore(\u0026cp-\u003ecp_lock, flags);\n+\t\tret = -EAGAIN;\n+\t\tgoto out;\n+\t}\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..552b32278e30d 100644\n--- a/net/rds/tcp.c\n+++ b/net/rds/tcp.c\n@@ -502,6 +502,48 @@ static bool rds_tcp_is_unloading(struct rds_connection *conn)\n \treturn atomic_read(\u0026rds_tcp_unloading) != 0;\n }\n \n+/* Gather @tc's connection for destruction: move the node to the\n+ * caller's @tmp_list and take a reference that keeps the connection,\n+ * and so the node, alive until rds_tcp_destroy_gathered_conns() has\n+ * dealt with it.  Called with rds_tcp_conn_lock held.  A connection\n+ * whose free is already running gets no reference; its free unlinks\n+ * the node itself, under the same lock, once we drop it.\n+ */\n+static void rds_tcp_gather_conn(struct rds_tcp_connection *tc,\n+\t\t\t\tstruct list_head *tmp_list)\n+{\n+\tif (rds_conn_get_unless_zero(tc-\u003et_cpath-\u003ecp_conn))\n+\t\tlist_move_tail(\u0026tc-\u003et_tcp_node, tmp_list);\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+ * Every entry holds a reference taken by rds_tcp_gather_conn(), so\n+ * none can be freed under the walk; each is dropped after its destroy.\n+ */\n+static void rds_tcp_destroy_gathered_conns(struct list_head *tmp_list)\n+{\n+\tstruct rds_tcp_connection *tc, *_tc;\n+\tstruct rds_connection *conn;\n+\n+\tlist_for_each_entry_safe(tc, _tc, tmp_list, t_tcp_node) {\n+\t\tconn = tc-\u003et_cpath-\u003ecp_conn;\n+\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(conn);\n+\t\trds_conn_put(conn);\n+\t}\n+}\n+\n static void rds_tcp_destroy_conns(void)\n {\n \tstruct rds_tcp_connection *tc, *_tc;\n@@ -511,12 +553,11 @@ static void rds_tcp_destroy_conns(void)\n \tspin_lock_irq(\u0026rds_tcp_conn_lock);\n \tlist_for_each_entry_safe(tc, _tc, \u0026rds_tcp_conn_list, t_tcp_node) {\n \t\tif (!list_has_conn(\u0026tmp_list, tc-\u003et_cpath-\u003ecp_conn))\n-\t\t\tlist_move_tail(\u0026tc-\u003et_tcp_node, \u0026tmp_list);\n+\t\t\trds_tcp_gather_conn(tc, \u0026tmp_list);\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@@ -691,15 +732,14 @@ static void rds_tcp_kill_sock(struct net *net)\n \t\tif (net != c_net)\n \t\t\tcontinue;\n \t\tif (!list_has_conn(\u0026tmp_list, tc-\u003et_cpath-\u003ecp_conn)) {\n-\t\t\tlist_move_tail(\u0026tc-\u003et_tcp_node, \u0026tmp_list);\n+\t\t\trds_tcp_gather_conn(tc, \u0026tmp_list);\n \t\t} else {\n \t\t\tlist_del(\u0026tc-\u003et_tcp_node);\n \t\t\ttc-\u003et_tcp_node_detached = true;\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 +845,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..7fea5501d7567 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,6 +278,12 @@ 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+\t/* A connection whose destroy has begun cannot be found here:\n+\t * TCP connections are only destroyed on netns teardown and on\n+\t * module unload, and both run rds_tcp_listen_stop() - which\n+\t * flushes this work and clears the listen socket that the top\n+\t * of this function tests - before any connection is destroyed.\n+\t */\n \tif (conn_state != RDS_CONN_CONNECTING \u0026\u0026 conn_state != RDS_CONN_ERROR) {\n \t\trds_conn_path_drop(cp, 0);\n \t\tgoto rst_nsk;\n@@ -316,10 +323,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 +354,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/21 06:14 agent patch-evaluator
0m Model:
core

Args:
null

Results:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_sendmsg",
    "rds_release",
    "rds_ioctl",
    "rds_send_queue_rm",
    "rds_recv_incoming",
    "rds_tcp_accept_one"
  ],
  "Reasoning": "The patch refactors connection lifecycle and reference counting in the RDS subsystem (net/rds). It introduces kref-based reference counting on rds_connection, alters locking schemes in rds_ioctl and rds_sendmsg, manages cached socket connection references, and adjusts connection teardown across core RDS, loopback, TCP, and IB transports. These code paths are directly reachable via standard AF_RDS socket operations and can introduce subtle concurrency bugs, refcount imbalances, or use-after-free conditions under fuzzing.",
  "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 7e7965e091176ceeb5b5cdfd3144694b6bfcb59d
Author: syz-cluster <triage@syzkaller.com>
Date:   Mon Sep 21 06:14:23 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..546961bb2e127 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,10 @@ 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().
+ */
 static struct rds_connection *rds_conn_lookup(struct net *net,
 					      struct hlist_head *head,
 					      const struct in6_addr *laddr,
@@ -96,6 +100,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 +176,30 @@ static void __rds_conn_path_init(struct rds_connection *conn,
 	cp->cp_flags = 0;
 }
 
+/* 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));
+}
+
+/* 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,97 @@ 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 - once incs hold one, 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 +755,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 +826,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.h b/net/rds/ib.h
index 5ff346a1e8baa..cb410c3ae8d85 100644
--- a/net/rds/ib.h
+++ b/net/rds/ib.h
@@ -143,6 +143,10 @@ struct rds_ib_device;
 struct rds_ib_connection {
 
 	struct list_head	ib_node;
+	/* set under ib_nodev_conns_lock once a transport teardown has
+	 * claimed ib_node; from then on only the teardown touches it
+	 */
+	bool			i_ib_node_detached;
 	struct rds_ib_device	*rds_ibdev;
 	struct rds_connection	*conn;
 
diff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c
index 4feb0edc360c8..cbeedfc4fabd0 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,15 @@ 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);
+		/* Drop the reference rds_conn_create() handed us.  The
+		 * conn stays reachable through cm_id->context without a
+		 * reference of its own for now; the CM event handler is
+		 * given one of its own by a following patch.
+		 */
+		rds_conn_put(conn);
+	}
 	if (err)
 		rdma_reject(cm_id, &err, sizeof(int),
 			    IB_CM_REJ_CONSUMER_DEFINED);
@@ -1271,6 +1285,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 +1293,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);
+	/* a transport teardown that gathered us first owns the node */
+	if (!ic->i_ib_node_detached)
+		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..1548e5be0e559 100644
--- a/net/rds/ib_rdma.c
+++ b/net/rds/ib_rdma.c
@@ -123,15 +123,18 @@ void rds_ib_add_conn(struct rds_ib_device *rds_ibdev, struct rds_connection *con
 {
 	struct rds_ib_connection *ic = conn->c_transport_data;
 
-	/* conn was previously on the nodev_conns_list */
+	/* conn was previously on the nodev_conns_list, unless a teardown
+	 * sweep has claimed it ahead of destroying it: then it is on its
+	 * way out, and its node belongs to the sweep.
+	 */
 	spin_lock_irq(&ib_nodev_conns_lock);
-	BUG_ON(list_empty(&ib_nodev_conns));
-	BUG_ON(list_empty(&ic->ib_node));
-	list_del(&ic->ib_node);
+	if (!ic->i_ib_node_detached) {
+		list_del(&ic->ib_node);
 
-	spin_lock(&rds_ibdev->spinlock);
-	list_add_tail(&ic->ib_node, &rds_ibdev->conn_list);
-	spin_unlock(&rds_ibdev->spinlock);
+		spin_lock(&rds_ibdev->spinlock);
+		list_add_tail(&ic->ib_node, &rds_ibdev->conn_list);
+		spin_unlock(&rds_ibdev->spinlock);
+	}
 	spin_unlock_irq(&ib_nodev_conns_lock);
 
 	ic->rds_ibdev = rds_ibdev;
@@ -142,15 +145,22 @@ void rds_ib_remove_conn(struct rds_ib_device *rds_ibdev, struct rds_connection *
 {
 	struct rds_ib_connection *ic = conn->c_transport_data;
 
-	/* place conn on nodev_conns_list */
+	bool detached;
+
+	/* place conn on nodev_conns_list - unless a teardown sweep has
+	 * claimed it ahead of destroying it, in which case its node
+	 * belongs to the sweep
+	 */
 	spin_lock(&ib_nodev_conns_lock);
 
 	spin_lock_irq(&rds_ibdev->spinlock);
-	BUG_ON(list_empty(&ic->ib_node));
-	list_del(&ic->ib_node);
+	detached = ic->i_ib_node_detached;
+	if (!detached)
+		list_del(&ic->ib_node);
 	spin_unlock_irq(&rds_ibdev->spinlock);
 
-	list_add_tail(&ic->ib_node, &ib_nodev_conns);
+	if (!detached)
+		list_add_tail(&ic->ib_node, &ib_nodev_conns);
 
 	spin_unlock(&ib_nodev_conns_lock);
 
@@ -163,13 +173,40 @@ void rds_ib_destroy_nodev_conns(void)
 	struct rds_ib_connection *ic, *_ic;
 	LIST_HEAD(tmp_list);
 
-	/* avoid calling conn_destroy with irqs off */
+	struct rds_connection *conn;
+
+	/* Gather the connections and take a reference on each, so that
+	 * none is freed under the walk below (a connection destroyed
+	 * earlier, for a protocol version mismatch, can be on this list
+	 * with only a socket's reference still pending).  One whose free
+	 * is already running gets no reference: its free unlinks the
+	 * node itself, under this lock, once we drop it.  Marking the
+	 * node detached claims it for this sweep: rds_ib_add_conn(),
+	 * rds_ib_remove_conn() and rds_ib_conn_free() leave a claimed
+	 * node alone, so the walk over tmp_list below needs no lock.
+	 * Avoid calling conn_destroy with irqs off.
+	 */
 	spin_lock_irq(&ib_nodev_conns_lock);
-	list_splice(&ib_nodev_conns, &tmp_list);
+	list_for_each_entry_safe(ic, _ic, &ib_nodev_conns, ib_node) {
+		if (rds_conn_get_unless_zero(ic->conn)) {
+			ic->i_ib_node_detached = true;
+			list_move_tail(&ic->ib_node, &tmp_list);
+		}
+	}
 	spin_unlock_irq(&ib_nodev_conns_lock);
 
-	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node)
-		rds_conn_destroy(ic->conn);
+	/* rds_conn_destroy() can return before the connection is freed,
+	 * and it is the free - rds_ib_conn_free() - that would unlink
+	 * ib_node.  tmp_list lives on this stack frame, so take each node
+	 * off it before its destroy; the free then leaves it alone.
+	 */
+	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node) {
+		conn = ic->conn;
+		list_del_init(&ic->ib_node);
+
+		rds_conn_destroy(conn);
+		rds_conn_put(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..5bac858df87ef 100644
--- a/net/rds/loop.c
+++ b/net/rds/loop.c
@@ -156,6 +156,45 @@ 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;
+	struct rds_connection *conn;
+
+	list_for_each_entry_safe(lc, _lc, tmp_list, loop_node) {
+		conn = lc->conn;
+		WARN_ON(rcu_access_pointer(conn->c_passive));
+
+		spin_lock_irq(&loop_conns_lock);
+		list_del_init(&lc->loop_node);
+		spin_unlock_irq(&loop_conns_lock);
+
+		rds_conn_destroy(conn);
+		rds_conn_put(conn);
+	}
+}
+
+/* Gather @lc's connection for destruction: move the node to the
+ * caller's @tmp_list and take a reference that keeps the connection,
+ * and so the node, alive until rds_loop_destroy_gathered_conns() has
+ * dealt with it.  Called with loop_conns_lock held.  A connection
+ * whose free is already running gets no reference; its free unlinks
+ * the node itself, under the same lock, once we drop it.
+ */
+static void rds_loop_gather_conn(struct rds_loop_connection *lc,
+				 struct list_head *tmp_list)
+{
+	if (rds_conn_get_unless_zero(lc->conn))
+		list_move_tail(&lc->loop_node, tmp_list);
+}
+
 static void rds_loop_conn_free(void *arg)
 {
 	struct rds_loop_connection *lc = arg;
@@ -163,7 +202,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);
 }
@@ -187,14 +228,13 @@ void rds_loop_exit(void)
 	synchronize_rcu();
 	/* avoid calling conn_destroy with irqs off */
 	spin_lock_irq(&loop_conns_lock);
-	list_splice(&loop_conns, &tmp_list);
-	INIT_LIST_HEAD(&loop_conns);
+	list_for_each_entry_safe(lc, _lc, &loop_conns, loop_node)
+		rds_loop_gather_conn(lc, &tmp_list);
 	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)
@@ -208,14 +248,11 @@ static void rds_loop_kill_conns(struct net *net)
 
 		if (net != c_net)
 			continue;
-		list_move_tail(&lc->loop_node, &tmp_list);
+		rds_loop_gather_conn(lc, &tmp_list);
 	}
 	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 47d5e9ab9b109..cea48ad745ca7 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..321f2da9e76d9 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;
+	/* rds_conn_destroy() quiesces the connection synchronously;
+	 * freeing it - the connection memory, the path workqueues and
+	 * the transport's per-connection state - 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);
@@ -650,7 +669,9 @@ struct rds_sock {
 
 	/*
 	 * rds_sendmsg caches the conn it used the last time around.
-	 * This helps avoid costly lookups.
+	 * This helps avoid costly lookups.  The cache owns a connection
+	 * reference, dropped when it is replaced or the socket is
+	 * released, and is read and written under rs_lock.
 	 */
 	struct rds_connection	*rs_conn;
 
@@ -819,6 +840,21 @@ 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 +1030,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..1fcd4483d3be4 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)
@@ -406,8 +425,9 @@ void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,
 	 * rds_find_bound() uses a global (netns-agnostic) hash table.
 	 * An RDS connection created in netns A can match a socket bound
 	 * in the init netns, delivering inc cross-netns with inc->i_conn
-	 * pointing into netns A.  When cleanup_net() then frees that conn,
-	 * any subsequent dereference of inc->i_conn is a use-after-free.
+	 * pointing into netns A.  Cross-netns delivery is wrong on its
+	 * own, and the inc's connection reference would keep a
+	 * connection of a dead netns around, with a stale c_net.
 	 * Drop the inc if the receiving socket lives in a different netns.
 	 */
 	if (!net_eq(sock_net(rds_rs_to_sk(rs)), rds_conn_net(conn))) {
diff --git a/net/rds/send.c b/net/rds/send.c
index 1afa981e5c06d..afb22779ff977 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,28 @@ 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;
 		}
+		/* rs_tos was sampled without rs_lock for the create above,
+		 * and SIOCRDSSETTOS only refuses a change once rs_conn is
+		 * set, so it can have changed underneath us.  Do not
+		 * install - or send on - a connection whose ToS no longer
+		 * matches the socket's; the retry uses the new one.
+		 */
+		spin_lock_irqsave(&rs->rs_lock, flags);
+		if (conn->c_tos != rs->rs_tos) {
+			spin_unlock_irqrestore(&rs->rs_lock, flags);
+			ret = -EAGAIN;
+			goto out;
+		}
+		/* hand the cache its own reference */
+		rds_conn_get(conn);
+		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 +1432,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 +1501,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 +1533,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 +1542,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 */
@@ -1519,9 +1588,18 @@ rds_send_probe(struct rds_conn_path *cp, __be16 sport,
 		goto out;
 
 	spin_lock_irqsave(&cp->cp_lock, flags);
+	/* Same rule as rds_send_queue_rm(): once the destroy has purged
+	 * cp_send_queue under this lock, nothing may be added behind it.
+	 */
+	if (rds_destroy_pending(cp->cp_conn)) {
+		spin_unlock_irqrestore(&cp->cp_lock, flags);
+		ret = -EAGAIN;
+		goto out;
+	}
 	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..552b32278e30d 100644
--- a/net/rds/tcp.c
+++ b/net/rds/tcp.c
@@ -502,6 +502,48 @@ static bool rds_tcp_is_unloading(struct rds_connection *conn)
 	return atomic_read(&rds_tcp_unloading) != 0;
 }
 
+/* Gather @tc's connection for destruction: move the node to the
+ * caller's @tmp_list and take a reference that keeps the connection,
+ * and so the node, alive until rds_tcp_destroy_gathered_conns() has
+ * dealt with it.  Called with rds_tcp_conn_lock held.  A connection
+ * whose free is already running gets no reference; its free unlinks
+ * the node itself, under the same lock, once we drop it.
+ */
+static void rds_tcp_gather_conn(struct rds_tcp_connection *tc,
+				struct list_head *tmp_list)
+{
+	if (rds_conn_get_unless_zero(tc->t_cpath->cp_conn))
+		list_move_tail(&tc->t_tcp_node, tmp_list);
+}
+
+/* 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.
+ * Every entry holds a reference taken by rds_tcp_gather_conn(), so
+ * none can be freed under the walk; each is dropped after its destroy.
+ */
+static void rds_tcp_destroy_gathered_conns(struct list_head *tmp_list)
+{
+	struct rds_tcp_connection *tc, *_tc;
+	struct rds_connection *conn;
+
+	list_for_each_entry_safe(tc, _tc, tmp_list, t_tcp_node) {
+		conn = tc->t_cpath->cp_conn;
+
+		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(conn);
+		rds_conn_put(conn);
+	}
+}
+
 static void rds_tcp_destroy_conns(void)
 {
 	struct rds_tcp_connection *tc, *_tc;
@@ -511,12 +553,11 @@ static void rds_tcp_destroy_conns(void)
 	spin_lock_irq(&rds_tcp_conn_lock);
 	list_for_each_entry_safe(tc, _tc, &rds_tcp_conn_list, t_tcp_node) {
 		if (!list_has_conn(&tmp_list, tc->t_cpath->cp_conn))
-			list_move_tail(&tc->t_tcp_node, &tmp_list);
+			rds_tcp_gather_conn(tc, &tmp_list);
 	}
 	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);
@@ -691,15 +732,14 @@ static void rds_tcp_kill_sock(struct net *net)
 		if (net != c_net)
 			continue;
 		if (!list_has_conn(&tmp_list, tc->t_cpath->cp_conn)) {
-			list_move_tail(&tc->t_tcp_node, &tmp_list);
+			rds_tcp_gather_conn(tc, &tmp_list);
 		} else {
 			list_del(&tc->t_tcp_node);
 			tc->t_tcp_node_detached = true;
 		}
 	}
 	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 +845,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..7fea5501d7567 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,6 +278,12 @@ 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);
+	/* A connection whose destroy has begun cannot be found here:
+	 * TCP connections are only destroyed on netns teardown and on
+	 * module unload, and both run rds_tcp_listen_stop() - which
+	 * flushes this work and clears the listen socket that the top
+	 * of this function tests - before any connection is destroyed.
+	 */
 	if (conn_state != RDS_CONN_CONNECTING && conn_state != RDS_CONN_ERROR) {
 		rds_conn_path_drop(cp, 0);
 		goto rst_nsk;
@@ -316,10 +323,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 +354,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/21 06:14 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 24078
							output: 26
							thoughts: 723
						

Thoughts:
**Analyzing the RDS Refactoring Patch Series**

Okay, so I've got a patch series here focused on refactoring the connection lifecycle and refcounting within the Reliable Datagram Sockets (RDS) subsystem in the Linux kernel. My initial instinct is to go straight to `git log` to understand the context, but since I don't have that tool at hand, I'll leverage the diff provided. No problem, I can work with that.

First, I see the changes touch a wide range of files within `net/rds/`, which indicates a pretty comprehensive modification. Specifically, I see changes to `af_rds.c`, `connection.c`, `ib.c`, `ib.h`, `ib_cm.c`, `ib_rdma.c`, `ib_recv.c`, `ib_send.c`, `loop.c`, `message.c`, `rdma_transport.c`, `rds.h`, `recv.c`, `send.c`, `tcp.c`, and `tcp_listen.c`. That's a lot.

Looking at the core of the patch, it seems like they're introducing a `c_refcount` (using `kref`) to `struct rds_connection`. This is a pretty common pattern for managing the lifetime of kernel objects, so I'd expect associated functions like `rds_conn_get()`, `rds_conn_put()`, and `rds_conn_destroy_fini()` to appear. And yes, there they are. Also, they're updating the create, lookup, and destroy functions.

I can see some modifications in `af_rds.c`, where they're handling connection reference counting and locking with `rs_lock` now, which is expected. Same with refcount changes in `rds_sendmsg()`. The message management and transport drivers are also being updated to work with the connection refcounts. They're making changes to RDS loopback, RDS TCP, and RDS IB. Good to see this is impacting the different transports.

The fact that RDS loopback and TCP are software-based and testable in standard environments is important. That means it should be possible to easily test and fuzz these changes. Since the patch series also changes how RDS connections work, that probably also implies that syzkaller can easily fuzz the code too.

Now I need to check how reachable RDS is in standard QEMU and, crucially, within syzkaller. I know RDS is a standard network protocol socket family (`AF_RDS = 21`). That's promising. Syzkaller's documentation would be a good place to double-check. I'll search for "rds.txt". I need to search the `net/rds/Kconfig` file.



4/2 2026/09/21 06:14 tool grepper
0m Args:
{
  "Expression": "config RDS",
  "PathPrefix": "net/rds/"
}

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-9-\nnet/rds/Kconfig:10:config RDS_RDMA\nnet/rds/Kconfig-11-\ttristate \"RDS over Infiniband\"\n--\nnet/rds/Kconfig-16-\nnet/rds/Kconfig:17:config RDS_TCP\nnet/rds/Kconfig-18-\ttristate \"RDS over TCP\"\n--\nnet/rds/Kconfig-24-\nnet/rds/Kconfig:25:config RDS_DEBUG\nnet/rds/Kconfig-26-\tbool \"RDS debugging messages\"\n"
}

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

Args:
null

Results:
null

Tokens:
							input: 25613
							output: 32
							thoughts: 12
						

Thoughts:
Okay, let's dive into this. Here's my take on that `net/rds/Kconfig` file:

**Diving into the RDS Configuration**

Alright, so the first thing I'm doing is opening up `net/rds/Kconfig`. Classic. Kconfig files are the lifeblood of Linux kernel configuration, and this one specifically deals with the Reliable Datagram Sockets (RDS) protocol, which is a niche but important protocol for high-performance networking, especially in clusters and HPC environments.

My immediate focus will be on understanding the available options, their dependencies, and any default settings. I'm looking for the core RDS functionality – the enabling and disabling of the protocol itself (`CONFIG_RDS`), which is usually the top-level option. Then, I'll be scanning for the related features, such as the transport layer, and various configuration parameters, looking for areas I can configure.

I'll be paying close attention to any dependencies on other kernel subsystems or features. Is it reliant on RDMA? What about specific hardware drivers? Any inter-dependencies that are complex will become a focus. I might also be looking at the documentation, if it's available, to better understand what this protocol is meant to do.

Ultimately, I'm trying to gain a high-level understanding of what's configurable, the design choices the developers made, and how to potentially customize RDS for a specific environment. If this were a real-world scenario, I'd also be thinking about any potential performance implications of the settings.



6/2 2026/09/21 06:14 tool read-file
0m Args:
{
  "File": "net/rds/Kconfig",
  "FirstLine": 1,
  "LineCount": 50
}

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

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

Args:
null

Results:
null

Tokens:
							input: 26430
							output: 31
							thoughts: 14
						

Thoughts:
Here's my take on those thoughts, tailored for an expert audience:

**Diving into the Network RDS Makefile**

Alright, time to dive into the `Makefile` within `net/rds/`. I need to understand how the Remote Direct Memory Access (RDMA) over Sockets (RDS) stack is being built and managed. I'm looking for the build process, dependencies, and any custom flags or configurations being employed. This `Makefile` is the blueprint, the key to understanding how these crucial networking components are orchestrated. I'll pay close attention to the target definitions, particularly how the kernel modules (`rds.ko`, etc.) are compiled, linked, and installed. I also want to see how this interacts with the overall kernel build system. Any clever hacks or optimizations I should know about? Let's get to work.



8/2 2026/09/21 06:14 tool read-file
0m Args:
{
  "File": "net/rds/Makefile",
  "FirstLine": 1,
  "LineCount": 50
}

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  21:\tGCOV_PROFILE := y\n  22:\tendif\n"
}

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

Args:
null

Results:
null

Tokens:
							input: 27041
							output: 32
							thoughts: 1639
						

Thoughts:
**Fuzzing Recommendation: WorthFuzzing=true - RDS Connection and Lifetime Management**

Okay, let's break this down. My analysis indicates a high likelihood of finding vulnerabilities, making this a prime candidate for fuzzing. I'm seeing significant modifications to the RDS (Reliable Datagram Sockets) core networking/protocol stack, specifically within `net/rds/`. We're talking about a comprehensive overhaul of connection refcounting and lifetime management. This includes critical areas like `rds_release`, `rds_ioctl`, connection lookup/creation/destruction (`rds_conn_*`), message handling, and send/receive paths, along with the TCP implementation.

The dependencies are clear: `CONFIG_RDS` enables the core `rds.o` module, with `CONFIG_RDS_TCP` and `CONFIG_RDS_RDMA` adding TCP and RDMA support, respectively. Importantly, `CONFIG_RDS` relies on `INET`, ensuring that the code is reachable through standard networking interfaces, making it testable.

The changes involve a large number of functions, including the following:

-   `rds_release`, `rds_ioctl`
-   `rds_conn_lookup`, `rds_conn_free_transport_data`, `__rds_conn_create`, `rds_conn_create`, `rds_conn_create_outgoing` `rds_conn_path_quiesce`, `rds_conn_path_free`, `rds_conn_destroy_fini`, `rds_conn_wait_conns_freed`, `rds_conn_get`, `rds_conn_put`, `rds_conn_destroy`
-   `rds_message_free`, `rds_message_put`
-   `rds_inc_init`, `rds_inc_path_init`, `rds_inc_put`, `rds_recv_incoming`
-   `rds_send_xmit`, `rds_send_queue_rm`, `rds_sendmsg`, `rds_send_probe`
-   `rds_tcp_accept_one`, `rds_tcp_destroy_gathered_conns`, `rds_tcp_destroy_conns`, `rds_tcp_kill_sock`
-   `rds_loop_destroy_gathered_conns`, `rds_loop_gather_conn`

This is not a trivial patch! This is a major change touching on core aspects of connection management. These changes are likely addressing race conditions that may exist between sendmsg, ioctl, release, connection teardown, and netns exit. This kind of modification screams potential for subtle race conditions, refcounting bugs (memory leaks, use-after-free, double frees), lock inversions, or deadlocks. These types of vulnerabilities are prime targets for fuzzing.

My analysis aligns perfectly with the "WorthFuzzing=true" criteria, specifically the sections covering "Core Subsystems" (core networking) and "Refactorings and Code Cleanups" (restructuring of data structures and control flow). RDS sockets are definitively reachable through standard socket syscalls, for example using `socket(AF_RDS, ...)` calls.

Now, let's pick some specific "FocusSymbols." We want functions that are modified or newly added that are specific to RDS (not general-purpose kernel routines). These focus symbols should be reachable and observable by a fuzzer such as Syzkaller. After careful consideration, the focus symbols I recommend are:

-   `rds_conn_destroy`
-   `rds_conn_put`
-   `rds_conn_create` (or `__rds_conn_create`)
-   `rds_sendmsg`
-   `rds_send_queue_rm`
-   `rds_recv_incoming`
-   `rds_ioctl`
-   `rds_release`

These functions are specific to the RDS protocol and are not ubiquitous kernel functions. Fuzzing them offers a high potential for uncovering significant vulnerabilities. Therefore, I conclude this patch should be considered a high-priority fuzzing target and thus `WorthFuzzing = true`.



10/2 2026/09/21 06:14 tool grepper
0m Args:
{
  "Expression": "rds_conn_create",
  "PathPrefix": "net/rds/connection.c"
}

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-451-\nnet/rds/connection.c:452:struct rds_connection *rds_conn_create(struct net *net,\nnet/rds/connection.c-453-\t\t\t\t       const struct in6_addr *laddr,\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-}\nnet/rds/connection.c:460:EXPORT_SYMBOL_GPL(rds_conn_create);\nnet/rds/connection.c-461-\nnet/rds/connection.c:462:struct rds_connection *rds_conn_create_outgoing(struct net *net,\nnet/rds/connection.c-463-\t\t\t\t\t\tconst struct in6_addr *laddr,\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-}\nnet/rds/connection.c:470:EXPORT_SYMBOL_GPL(rds_conn_create_outgoing);\nnet/rds/connection.c-471-\n--\nnet/rds/connection.c=745=void rds_conn_destroy(struct rds_connection *conn)\n--\nnet/rds/connection.c-782-\t/* Snatch c_passive while holding the lock:\nnet/rds/connection.c:783:\t * __rds_conn_create() dereferences it under rcu_read_lock()\nnet/rds/connection.c-784-\t * (and refuses to install a new one once c_destroy_in_prog is\n--\nnet/rds/connection.c-792-\t/* If we are a parent's passive twin, invalidate its pointer to\nnet/rds/connection.c:793:\t * us as well, so that __rds_conn_create() cannot hand out a\nnet/rds/connection.c-794-\t * connection whose teardown has begun.  The parent is the\n"
}

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

Args:
null

Results:
null

Tokens:
							input: 30268
							output: 35
							thoughts: 669
						

Thoughts:
**Analyzing RDS Configuration Requirements**

Okay, so I'm looking at these RDS changes. The core connection creation function, `__rds_conn_create`, is static and called by both `rds_conn_create` and `rds_conn_create_outgoing`. The compiler could potentially inline these, depending on optimization. The listing of the various `rds_*` functions seems comprehensive; `rds_conn_create`, `rds_conn_destroy`, `rds_conn_put`, `rds_conn_get`, `rds_sendmsg`, `rds_release`, `rds_ioctl`, `rds_send_queue_rm`, `rds_recv_incoming` should cover the primary interfaces.

Now, let's get into the EnableConfigs. I need to make sure I have all the necessary options set.

From `net/rds/Kconfig`, I can see that `CONFIG_RDS` is obviously required. Since these changes touch `net/rds/tcp.c` and `net/rds/tcp_listen.c`, `CONFIG_RDS_TCP` is also needed. `CONFIG_RDS` also depends on `CONFIG_INET`, so that's in.

And, regarding `CONFIG_RDS_RDMA`, the presence of `net/rds/ib.c`, `net/rds/ib_cm.c`, etc. makes it clear that it's related. I see that `CONFIG_RDS_RDMA` depends on `RDS && INFINIBAND && INFINIBAND_ADDR_TRANS`. I'm curious if Infiniband software emulation, like `rdma_rxe` or `siw`, is supported in QEMU. Given what I've seen in the past, I'm pretty sure `CONFIG_RDMA_RXE` or `CONFIG_RDMA_SIW` would handle that.

Now, let's get down to brass tacks: Are `RDS`, `RDS_TCP`, and `RDS_RDMA` standard, or are we dealing with some custom options? Let's just strip the "CONFIG_" prefix and see: "RDS", "RDS_TCP", "RDS_RDMA", "INET". Got it.

Wait. A quick thought: Is `CONFIG_RDS` alone enough to get the core changes to compile? Let's dissect this.

Yep, `RDS` compiles `rds.o` (covering af_rds.c, connection.c, message.c, recv.c, send.c, loop.c). `RDS_TCP` compiles `rds_tcp.o` (tcp.c, tcp_listen.c). `RDS_RDMA` compiles `rds_rdma.o` (ib.c, ib_cm.c, ib_rdma.c, ib_recv.c, ib_send.c, rdma_transport.c). So enabling `RDS`, `RDS_TCP`, and `RDS_RDMA` will enable all the modified files that I'm looking at. Excellent.

Just to be absolutely certain, I should double-check dependencies within the Kconfig file for `RDS_RDMA`. I see `RDS_RDMA` depends on `RDS && INFINIBAND && INFINIBAND_ADDR_TRANS`, as expected. `RDS_TCP` depends on `RDS`, and `RDS` depends on `INET`. So, the minimal set of configurations to get this to compile and function correctly is: `RDS`, `RDS_TCP`, `RDS_RDMA`, `INFINIBAND`, and `INFINIBAND_ADDR_TRANS`. Perfect.

Now, let me confirm what `INFINIBAND_ADDR_TRANS` depends on.



12/2 2026/09/21 06:15 tool grepper
0m Args:
{
  "Expression": "config INFINIBAND_ADDR_TRANS",
  "PathPrefix": "drivers/infiniband/"
}

Results:
{
  "Output": "drivers/infiniband/Kconfig=51=config INFINIBAND_ON_DEMAND_PAGING\n--\ndrivers/infiniband/Kconfig-63-\ndrivers/infiniband/Kconfig:64:config INFINIBAND_ADDR_TRANS\ndrivers/infiniband/Kconfig-65-\tbool \"RDMA/CM\"\n--\ndrivers/infiniband/Kconfig-71-\ndrivers/infiniband/Kconfig:72:config INFINIBAND_ADDR_TRANS_CONFIGFS\ndrivers/infiniband/Kconfig-73-\tbool\n"
}

13/2 2026/09/21 06:15 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 31909
							output: 36
							thoughts: 21
						

Thoughts:
Here's my summary, as if I were thinking these thoughts:

**Examining the Infiniband Kconfig**

Okay, time to dig into this. I need to refresh my memory on the Infiniband configuration options. Let's start with the `drivers/infiniband/Kconfig` file. I'll focus around line 64, because that's often where the core or key dependencies are defined. I'm expecting to find the central configuration symbol, maybe dependencies on specific hardware or other kernel subsystems, or perhaps some branching logic based on the underlying architecture or feature set. This Kconfig file is the bedrock for Infiniband configuration, so understanding the dependencies defined around that line is critical for the task at hand. Hopefully, I'll quickly spot what I'm looking for.



14/2 2026/09/21 06:15 tool read-file
0m Args:
{
  "File": "drivers/infiniband/Kconfig",
  "FirstLine": 60,
  "LineCount": 20
}

Results:
{
  "Contents": "  60:\t\t  Together with driver support this allows registration of\n  61:\t\t  memory regions without pinning their pages, fetching the\n  62:\t\t  pages on demand instead.\n  63:\t\n  64:\tconfig INFINIBAND_ADDR_TRANS\n  65:\t\tbool \"RDMA/CM\"\n  66:\t\tdepends on INFINIBAND\n  67:\t\tdefault y\n  68:\t\thelp\n  69:\t\t  Support for RDMA communication manager (CM).\n  70:\t\t  This allows for a generic connection abstraction over RDMA.\n  71:\t\n  72:\tconfig INFINIBAND_ADDR_TRANS_CONFIGFS\n  73:\t\tbool\n  74:\t\tdepends on INFINIBAND_ADDR_TRANS \u0026\u0026 CONFIGFS_FS \u0026\u0026 !(INFINIBAND=y \u0026\u0026 CONFIGFS_FS=m)\n  75:\t\tdefault y\n  76:\t\thelp\n  77:\t\t  ConfigFS support for RDMA communication manager (CM).\n  78:\t\t  This allows the user to config the default GID type that the CM\n  79:\t\t  uses for each device, when initiaing new connections.\n"
}

15/2 2026/09/21 06:15 llm patch-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 32423
							output: 237
							thoughts: 836
						

Thoughts:
**Patch Analysis for Fuzzing Readiness**

Okay, so `INFINIBAND_ADDR_TRANS` defaults to 'y' if `INFINIBAND` is enabled. That means including `RDS`, `RDS_TCP`, `RDS_RDMA`, and `INFINIBAND` is the most comprehensive way to enable the relevant code paths touched by this patch. Makes sense.

Now, let's look at the actual patch logic. I need to make sure there's nothing that's dead code or unreachable. Right, so we have a bunch of functions being touched:

*   `af_rds.c`: `rds_release` is dropping `rs_conn`, `rds_ioctl` is handling `SIOCRDSSETTOS` and `SIOCRDSGETTOS` with `rs_lock` which sounds like setting/getting ToS values.
*   `connection.c`: Dealing with connection lookup, creation, quiescing, freeing, destruction and reference counting: `rds_conn_lookup`, `__rds_conn_create`, `rds_conn_path_quiesce`, `rds_conn_path_free`, `rds_conn_destroy_fini`, `rds_conn_destroy`, `rds_conn_get`, `rds_conn_put`.
*   `message.c`:  Freeing and putting messages: `rds_message_free`, `rds_message_put`.
*   `recv.c`: Initialization and handling of incoming data `rds_inc_init`, `rds_inc_path_init`, `rds_inc_put`, `rds_recv_incoming`.
*   `send.c`:  Sending data: `rds_sendmsg`, `rds_send_queue_rm`, `rds_send_probe`, `rds_send_xmit`.
*   `loop.c`: Connection gathering and destruction on the loopback transport: `rds_loop_gather_conn`, `rds_loop_destroy_gathered_conns`, `rds_loop_exit`, `rds_loop_kill_conns`.
*   `tcp.c`:  The TCP transport, dealing with connections: `rds_tcp_gather_conn`, `rds_tcp_destroy_gathered_conns`, `rds_tcp_destroy_conns`, `rds_tcp_kill_sock`.
*   `tcp_listen.c`: Handling TCP accepts: `rds_tcp_accept_one`.
*   `ib*.c`, `rdma_transport.c`: This covers RDMA and InfiniBand-related code.

This covers the whole lifecycle of the socket: core connection management, IOCTLs, send/receive paths, connection setup/teardown, and reference counting, and it crosses RDS, the loopback, TCP, and RDMA transports.

Alright, these are all reachable through userspace syscalls, using `socket(AF_RDS, ...)` and friends, like `sendmsg`, `ioctl`, `close`, network namespaces and so on. No obvious restrictions there.

Now, let's see about *ubiquitous* symbols. Are any of these used throughout the kernel, or are they specific to RDS?

Okay, the candidate symbols are: `rds_conn_destroy`, `rds_conn_put`, `rds_conn_get`, `rds_conn_create`, `rds_sendmsg`, `rds_release`, `rds_ioctl`, `rds_send_queue_rm`, `rds_recv_incoming`, and `rds_tcp_accept_one`.

Nope. They're all specific to the RDS subsystem. That makes me feel better.

Okay, let's verify parameters for the `set-results` call.

*   `WorthFuzzing`: Definitely true.
*   `Reasoning`: The patch rework RDS connection lifecycle and reference counting by introducing kref-based refcounting for `rds_connection`, changing locking in `rds_ioctl` and `rds_sendmsg`, managing cached connection lifespans, and altering connection teardown across core RDS, loopback, TCP, and RDMA transports. All of this is reachable through AF_RDS sockets and could introduce race conditions, UAFs, or refcount leaks.
*   `EnableConfigs`: `["RDS", "RDS_TCP", "RDS_RDMA", "INFINIBAND"]` - Got it. Covers the bases.
*   `FocusSymbols`: `["rds_conn_destroy", "rds_conn_put", "rds_conn_get", "rds_conn_create", "rds_sendmsg", "rds_release", "rds_ioctl", "rds_send_queue_rm", "rds_recv_incoming", "rds_tcp_accept_one"]` - These are the key functions to watch


16/2 2026/09/21 06:15 tool set-results
0m Args:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_sendmsg",
    "rds_release",
    "rds_ioctl",
    "rds_send_queue_rm",
    "rds_recv_incoming",
    "rds_tcp_accept_one"
  ],
  "Reasoning": "The patch refactors connection lifecycle and reference counting in the RDS subsystem (net/rds). It introduces kref-based reference counting on rds_connection, alters locking schemes in rds_ioctl and rds_sendmsg, manages cached socket connection references, and adjusts connection teardown across core RDS, loopback, TCP, and IB transports. These code paths are directly reachable via standard AF_RDS socket operations and can introduce subtle concurrency bugs, refcount imbalances, or use-after-free conditions under fuzzing.",
  "WorthFuzzing": true
}

Results:
{
  "EnableConfigs": [
    "RDS",
    "RDS_TCP",
    "RDS_RDMA",
    "INFINIBAND"
  ],
  "FocusSymbols": [
    "rds_conn_destroy",
    "rds_conn_put",
    "rds_conn_get",
    "rds_conn_create",
    "rds_sendmsg",
    "rds_release",
    "rds_ioctl",
    "rds_send_queue_rm",
    "rds_recv_incoming",
    "rds_tcp_accept_one"
  ],
  "Reasoning": "The patch refactors connection lifecycle and reference counting in the RDS subsystem (net/rds). It introduces kref-based reference counting on rds_connection, alters locking schemes in rds_ioctl and rds_sendmsg, manages cached socket connection references, and adjusts connection teardown across core RDS, loopback, TCP, and IB transports. These code paths are directly reachable via standard AF_RDS socket operations and can introduce subtle concurrency bugs, refcount imbalances, or use-after-free conditions under fuzzing.",
  "WorthFuzzing": true
}

17/1 2026/09/21 06:15 action If
0m Args:
{
  "WorthFuzzing": true
}

Results:
null

18/2 2026/09/21 06:15 agent kmsan-evaluator
0m Model:
core

Args:
null

Results:
{
  "KMSANReasoning": "The patch introduces reference counting (kref) for 'struct rds_connection' to manage connection lifetimes safely and prevent use-after-free and concurrency races during connection teardown, module unload, and asynchronous operations (across RDS core, TCP, Loop, and IB transports).\n\nKASAN vs KMSAN applicability:\n- The primary failure modes and risks associated with these changes are use-after-free, double-free, reference leaks, and locking issues/deadlocks. These are covered by KASAN, KASAN-quarantine, and LOCKDEP in standard debug builds.\n- All touched structures (including newly added fields such as 'c_refcount', 'c_destroy_in_prog', 'i_ib_node_detached', and 't_tcp_node_detached') are allocated using zeroing allocators (kmem_cache_zalloc, kzalloc_obj) or are static/atomic variables initialized to zero.\n- No uninitialized stack/heap memory or structure padding is read or exposed across security boundaries (e.g. copied to userspace or passed uninitialized to kernel subsystems).\n\nConsequently, there are no uninitialized memory hazards that would necessitate a dedicated KMSAN fuzzing session.",
  "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 7e7965e091176ceeb5b5cdfd3144694b6bfcb59d
Author: syz-cluster <triage@syzkaller.com>
Date:   Mon Sep 21 06:14:23 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..546961bb2e127 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,10 @@ 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().
+ */
 static struct rds_connection *rds_conn_lookup(struct net *net,
 					      struct hlist_head *head,
 					      const struct in6_addr *laddr,
@@ -96,6 +100,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 +176,30 @@ static void __rds_conn_path_init(struct rds_connection *conn,
 	cp->cp_flags = 0;
 }
 
+/* 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));
+}
+
+/* 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,97 @@ 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 - once incs hold one, 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 +755,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 +826,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.h b/net/rds/ib.h
index 5ff346a1e8baa..cb410c3ae8d85 100644
--- a/net/rds/ib.h
+++ b/net/rds/ib.h
@@ -143,6 +143,10 @@ struct rds_ib_device;
 struct rds_ib_connection {
 
 	struct list_head	ib_node;
+	/* set under ib_nodev_conns_lock once a transport teardown has
+	 * claimed ib_node; from then on only the teardown touches it
+	 */
+	bool			i_ib_node_detached;
 	struct rds_ib_device	*rds_ibdev;
 	struct rds_connection	*conn;
 
diff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c
index 4feb0edc360c8..cbeedfc4fabd0 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,15 @@ 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);
+		/* Drop the reference rds_conn_create() handed us.  The
+		 * conn stays reachable through cm_id->context without a
+		 * reference of its own for now; the CM event handler is
+		 * given one of its own by a following patch.
+		 */
+		rds_conn_put(conn);
+	}
 	if (err)
 		rdma_reject(cm_id, &err, sizeof(int),
 			    IB_CM_REJ_CONSUMER_DEFINED);
@@ -1271,6 +1285,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 +1293,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);
+	/* a transport teardown that gathered us first owns the node */
+	if (!ic->i_ib_node_detached)
+		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..1548e5be0e559 100644
--- a/net/rds/ib_rdma.c
+++ b/net/rds/ib_rdma.c
@@ -123,15 +123,18 @@ void rds_ib_add_conn(struct rds_ib_device *rds_ibdev, struct rds_connection *con
 {
 	struct rds_ib_connection *ic = conn->c_transport_data;
 
-	/* conn was previously on the nodev_conns_list */
+	/* conn was previously on the nodev_conns_list, unless a teardown
+	 * sweep has claimed it ahead of destroying it: then it is on its
+	 * way out, and its node belongs to the sweep.
+	 */
 	spin_lock_irq(&ib_nodev_conns_lock);
-	BUG_ON(list_empty(&ib_nodev_conns));
-	BUG_ON(list_empty(&ic->ib_node));
-	list_del(&ic->ib_node);
+	if (!ic->i_ib_node_detached) {
+		list_del(&ic->ib_node);
 
-	spin_lock(&rds_ibdev->spinlock);
-	list_add_tail(&ic->ib_node, &rds_ibdev->conn_list);
-	spin_unlock(&rds_ibdev->spinlock);
+		spin_lock(&rds_ibdev->spinlock);
+		list_add_tail(&ic->ib_node, &rds_ibdev->conn_list);
+		spin_unlock(&rds_ibdev->spinlock);
+	}
 	spin_unlock_irq(&ib_nodev_conns_lock);
 
 	ic->rds_ibdev = rds_ibdev;
@@ -142,15 +145,22 @@ void rds_ib_remove_conn(struct rds_ib_device *rds_ibdev, struct rds_connection *
 {
 	struct rds_ib_connection *ic = conn->c_transport_data;
 
-	/* place conn on nodev_conns_list */
+	bool detached;
+
+	/* place conn on nodev_conns_list - unless a teardown sweep has
+	 * claimed it ahead of destroying it, in which case its node
+	 * belongs to the sweep
+	 */
 	spin_lock(&ib_nodev_conns_lock);
 
 	spin_lock_irq(&rds_ibdev->spinlock);
-	BUG_ON(list_empty(&ic->ib_node));
-	list_del(&ic->ib_node);
+	detached = ic->i_ib_node_detached;
+	if (!detached)
+		list_del(&ic->ib_node);
 	spin_unlock_irq(&rds_ibdev->spinlock);
 
-	list_add_tail(&ic->ib_node, &ib_nodev_conns);
+	if (!detached)
+		list_add_tail(&ic->ib_node, &ib_nodev_conns);
 
 	spin_unlock(&ib_nodev_conns_lock);
 
@@ -163,13 +173,40 @@ void rds_ib_destroy_nodev_conns(void)
 	struct rds_ib_connection *ic, *_ic;
 	LIST_HEAD(tmp_list);
 
-	/* avoid calling conn_destroy with irqs off */
+	struct rds_connection *conn;
+
+	/* Gather the connections and take a reference on each, so that
+	 * none is freed under the walk below (a connection destroyed
+	 * earlier, for a protocol version mismatch, can be on this list
+	 * with only a socket's reference still pending).  One whose free
+	 * is already running gets no reference: its free unlinks the
+	 * node itself, under this lock, once we drop it.  Marking the
+	 * node detached claims it for this sweep: rds_ib_add_conn(),
+	 * rds_ib_remove_conn() and rds_ib_conn_free() leave a claimed
+	 * node alone, so the walk over tmp_list below needs no lock.
+	 * Avoid calling conn_destroy with irqs off.
+	 */
 	spin_lock_irq(&ib_nodev_conns_lock);
-	list_splice(&ib_nodev_conns, &tmp_list);
+	list_for_each_entry_safe(ic, _ic, &ib_nodev_conns, ib_node) {
+		if (rds_conn_get_unless_zero(ic->conn)) {
+			ic->i_ib_node_detached = true;
+			list_move_tail(&ic->ib_node, &tmp_list);
+		}
+	}
 	spin_unlock_irq(&ib_nodev_conns_lock);
 
-	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node)
-		rds_conn_destroy(ic->conn);
+	/* rds_conn_destroy() can return before the connection is freed,
+	 * and it is the free - rds_ib_conn_free() - that would unlink
+	 * ib_node.  tmp_list lives on this stack frame, so take each node
+	 * off it before its destroy; the free then leaves it alone.
+	 */
+	list_for_each_entry_safe(ic, _ic, &tmp_list, ib_node) {
+		conn = ic->conn;
+		list_del_init(&ic->ib_node);
+
+		rds_conn_destroy(conn);
+		rds_conn_put(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..5bac858df87ef 100644
--- a/net/rds/loop.c
+++ b/net/rds/loop.c
@@ -156,6 +156,45 @@ 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;
+	struct rds_connection *conn;
+
+	list_for_each_entry_safe(lc, _lc, tmp_list, loop_node) {
+		conn = lc->conn;
+		WARN_ON(rcu_access_pointer(conn->c_passive));
+
+		spin_lock_irq(&loop_conns_lock);
+		list_del_init(&lc->loop_node);
+		spin_unlock_irq(&loop_conns_lock);
+
+		rds_conn_destroy(conn);
+		rds_conn_put(conn);
+	}
+}
+
+/* Gather @lc's connection for destruction: move the node to the
+ * caller's @tmp_list and take a reference that keeps the connection,
+ * and so the node, alive until rds_loop_destroy_gathered_conns() has
+ * dealt with it.  Called with loop_conns_lock held.  A connection
+ * whose free is already running gets no reference; its free unlinks
+ * the node itself, under the same lock, once we drop it.
+ */
+static void rds_loop_gather_conn(struct rds_loop_connection *lc,
+				 struct list_head *tmp_list)
+{
+	if (rds_conn_get_unless_zero(lc->conn))
+		list_move_tail(&lc->loop_node, tmp_list);
+}
+
 static void rds_loop_conn_free(void *arg)
 {
 	struct rds_loop_connection *lc = arg;
@@ -163,7 +202,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);
 }
@@ -187,14 +228,13 @@ void rds_loop_exit(void)
 	synchronize_rcu();
 	/* avoid calling conn_destroy with irqs off */
 	spin_lock_irq(&loop_conns_lock);
-	list_splice(&loop_conns, &tmp_list);
-	INIT_LIST_HEAD(&loop_conns);
+	list_for_each_entry_safe(lc, _lc, &loop_conns, loop_node)
+		rds_loop_gather_conn(lc, &tmp_list);
 	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)
@@ -208,14 +248,11 @@ static void rds_loop_kill_conns(struct net *net)
 
 		if (net != c_net)
 			continue;
-		list_move_tail(&lc->loop_node, &tmp_list);
+		rds_loop_gather_conn(lc, &tmp_list);
 	}
 	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 47d5e9ab9b109..cea48ad745ca7 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..321f2da9e76d9 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;
+	/* rds_conn_destroy() quiesces the connection synchronously;
+	 * freeing it - the connection memory, the path workqueues and
+	 * the transport's per-connection state - 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);
@@ -650,7 +669,9 @@ struct rds_sock {
 
 	/*
 	 * rds_sendmsg caches the conn it used the last time around.
-	 * This helps avoid costly lookups.
+	 * This helps avoid costly lookups.  The cache owns a connection
+	 * reference, dropped when it is replaced or the socket is
+	 * released, and is read and written under rs_lock.
 	 */
 	struct rds_connection	*rs_conn;
 
@@ -819,6 +840,21 @@ 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 +1030,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..1fcd4483d3be4 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)
@@ -406,8 +425,9 @@ void rds_recv_incoming(struct rds_connection *conn, struct in6_addr *saddr,
 	 * rds_find_bound() uses a global (netns-agnostic) hash table.
 	 * An RDS connection created in netns A can match a socket bound
 	 * in the init netns, delivering inc cross-netns with inc->i_conn
-	 * pointing into netns A.  When cleanup_net() then frees that conn,
-	 * any subsequent dereference of inc->i_conn is a use-after-free.
+	 * pointing into netns A.  Cross-netns delivery is wrong on its
+	 * own, and the inc's connection reference would keep a
+	 * connection of a dead netns around, with a stale c_net.
 	 * Drop the inc if the receiving socket lives in a different netns.
 	 */
 	if (!net_eq(sock_net(rds_rs_to_sk(rs)), rds_conn_net(conn))) {
diff --git a/net/rds/send.c b/net/rds/send.c
index 1afa981e5c06d..afb22779ff977 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,28 @@ 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;
 		}
+		/* rs_tos was sampled without rs_lock for the create above,
+		 * and SIOCRDSSETTOS only refuses a change once rs_conn is
+		 * set, so it can have changed underneath us.  Do not
+		 * install - or send on - a connection whose ToS no longer
+		 * matches the socket's; the retry uses the new one.
+		 */
+		spin_lock_irqsave(&rs->rs_lock, flags);
+		if (conn->c_tos != rs->rs_tos) {
+			spin_unlock_irqrestore(&rs->rs_lock, flags);
+			ret = -EAGAIN;
+			goto out;
+		}
+		/* hand the cache its own reference */
+		rds_conn_get(conn);
+		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 +1432,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 +1501,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 +1533,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 +1542,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 */
@@ -1519,9 +1588,18 @@ rds_send_probe(struct rds_conn_path *cp, __be16 sport,
 		goto out;
 
 	spin_lock_irqsave(&cp->cp_lock, flags);
+	/* Same rule as rds_send_queue_rm(): once the destroy has purged
+	 * cp_send_queue under this lock, nothing may be added behind it.
+	 */
+	if (rds_destroy_pending(cp->cp_conn)) {
+		spin_unlock_irqrestore(&cp->cp_lock, flags);
+		ret = -EAGAIN;
+		goto out;
+	}
 	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..552b32278e30d 100644
--- a/net/rds/tcp.c
+++ b/net/rds/tcp.c
@@ -502,6 +502,48 @@ static bool rds_tcp_is_unloading(struct rds_connection *conn)
 	return atomic_read(&rds_tcp_unloading) != 0;
 }
 
+/* Gather @tc's connection for destruction: move the node to the
+ * caller's @tmp_list and take a reference that keeps the connection,
+ * and so the node, alive until rds_tcp_destroy_gathered_conns() has
+ * dealt with it.  Called with rds_tcp_conn_lock held.  A connection
+ * whose free is already running gets no reference; its free unlinks
+ * the node itself, under the same lock, once we drop it.
+ */
+static void rds_tcp_gather_conn(struct rds_tcp_connection *tc,
+				struct list_head *tmp_list)
+{
+	if (rds_conn_get_unless_zero(tc->t_cpath->cp_conn))
+		list_move_tail(&tc->t_tcp_node, tmp_list);
+}
+
+/* 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.
+ * Every entry holds a reference taken by rds_tcp_gather_conn(), so
+ * none can be freed under the walk; each is dropped after its destroy.
+ */
+static void rds_tcp_destroy_gathered_conns(struct list_head *tmp_list)
+{
+	struct rds_tcp_connection *tc, *_tc;
+	struct rds_connection *conn;
+
+	list_for_each_entry_safe(tc, _tc, tmp_list, t_tcp_node) {
+		conn = tc->t_cpath->cp_conn;
+
+		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(conn);
+		rds_conn_put(conn);
+	}
+}
+
 static void rds_tcp_destroy_conns(void)
 {
 	struct rds_tcp_connection *tc, *_tc;
@@ -511,12 +553,11 @@ static void rds_tcp_destroy_conns(void)
 	spin_lock_irq(&rds_tcp_conn_lock);
 	list_for_each_entry_safe(tc, _tc, &rds_tcp_conn_list, t_tcp_node) {
 		if (!list_has_conn(&tmp_list, tc->t_cpath->cp_conn))
-			list_move_tail(&tc->t_tcp_node, &tmp_list);
+			rds_tcp_gather_conn(tc, &tmp_list);
 	}
 	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);
@@ -691,15 +732,14 @@ static void rds_tcp_kill_sock(struct net *net)
 		if (net != c_net)
 			continue;
 		if (!list_has_conn(&tmp_list, tc->t_cpath->cp_conn)) {
-			list_move_tail(&tc->t_tcp_node, &tmp_list);
+			rds_tcp_gather_conn(tc, &tmp_list);
 		} else {
 			list_del(&tc->t_tcp_node);
 			tc->t_tcp_node_detached = true;
 		}
 	}
 	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 +845,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..7fea5501d7567 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,6 +278,12 @@ 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);
+	/* A connection whose destroy has begun cannot be found here:
+	 * TCP connections are only destroyed on netns teardown and on
+	 * module unload, and both run rds_tcp_listen_stop() - which
+	 * flushes this work and clears the listen socket that the top
+	 * of this function tests - before any connection is destroyed.
+	 */
 	if (conn_state != RDS_CONN_CONNECTING && conn_state != RDS_CONN_ERROR) {
 		rds_conn_path_drop(cp, 0);
 		goto rst_nsk;
@@ -316,10 +323,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 +354,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);
 

19/3 2026/09/21 06:15 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 23290
							output: 29
							thoughts: 1285
						

Thoughts:
**KMSAN Fuzzing Evaluation for RDS Patch Series**

Okay, let's break this down. My goal is to figure out if this RDS patch series warrants a dedicated KMSAN fuzzing session, beyond the usual KASAN coverage. I'm going to carefully analyze the diff, looking for potential areas where KMSAN might be especially useful.

First, the patch touches a bunch of RDS code, specifically: `net/rds/af_rds.c`, `net/rds/connection.c`, and several files related to RDMA, TCP, and the core RDS implementation. That's a wide area, suggesting fairly significant changes.

Reading through it, it seems this is mostly focused on improving memory management, particularly related to the lifetime of `struct rds_connection`. The patch is introducing reference counting (`kref`) to handle connections more robustly. They're implementing `rds_conn_get`, `rds_conn_put`, `rds_conn_destroy`, `rds_conn_destroy_fini` functions to manage the refcounting. This also includes `rds_conn_freed_waitq` and `rds_conn_wait_conns_freed` for safe module unloading. This is crucial for fixing those use-after-free and race condition bugs that are a nightmare to debug. They're trying to prevent issues during teardown or unload with ongoing traffic. The core idea is to defer freeing of `rds_connection` until the reference count goes to zero.

I spot changes to locking (replacing `rds_sock_lock` with `spin_lock_irqsave(&rs->rs_lock, flags)`), ensuring that the updates to `rs->rs_tos` and `rs->rs_conn` are properly protected. It doesn't seem like there are any new writes of kernel structures to userspace, which is a good thing as this will simplify the analysis. `rds_ioctl` seems to be the only place where data from the kernel is being sent to userspace. In `rds_ioctl`, `tos` is being initialized with 0. So no obvious uninitialized memory issues that KMSAN would catch there. After I did a quick `copy_to_user` and `put_user` search, I couldn't find anything else.

Now, the important part: Does this introduce any potential KMSAN targets? The main things I'm looking for are:

1.  **Kernel structures to userspace:** Anything going out to user space via `copy_to_user`, `put_user`, netlink, etc. where we might have uninitialized fields or padding. So I looked at `rds_ioctl`, and I didn't see any obvious issues.
2.  **Conditional logic based on potentially uninitialized variables:** This includes any new fields that might be used in control flow. I did a quick check on the newly introduced fields (`c_refcount`, `c_destroy_in_prog`) and they seem to be initialized properly. They're being allocated via `kmem_cache_zalloc`, meaning the memory is zeroed. I didn't see anything obvious that would lead to this type of issue.

Given this analysis, my *initial* assessment is that this patch series doesn't strongly suggest a *dedicated* KMSAN fuzzing session. The focus on refcounting and connection lifetime is great for KASAN, and this will probably be sufficient. However, it's always worth keeping an eye on the fuzzing results, as these refcounting changes are complex and might indirectly reveal issues that KMSAN can help with.

I'll keep a close watch during the standard KASAN fuzzing and reassess if something pops up that screams KMSAN.



20/3 2026/09/21 06:15 tool grepper
0m Args:
{
  "Expression": "rds_conn_slab",
  "PathPrefix": "net/rds/"
}

Results:
{
  "Output": "net/rds/connection.c=52=static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];\nnet/rds/connection.c:53:static struct kmem_cache *rds_conn_slab;\nnet/rds/connection.c-54-\n--\nnet/rds/connection.c=211=static struct rds_connection *__rds_conn_create(struct net *net,\n--\nnet/rds/connection.c-257-\nnet/rds/connection.c:258:\tconn = kmem_cache_zalloc(rds_conn_slab, gfp);\nnet/rds/connection.c-259-\tif (!conn) {\n--\nnet/rds/connection.c-264-\tif (!conn-\u003ec_path) {\nnet/rds/connection.c:265:\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-266-\t\tconn = ERR_PTR(-ENOMEM);\n--\nnet/rds/connection.c-294-\t\tkfree(conn-\u003ec_path);\nnet/rds/connection.c:295:\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-296-\t\tconn = ERR_PTR(ret);\n--\nnet/rds/connection.c-322-\t\t\t\tkfree(conn-\u003ec_path);\nnet/rds/connection.c:323:\t\t\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-324-\t\t\t\tconn = ERR_PTR(-EOPNOTSUPP);\n--\nnet/rds/connection.c-351-\t\tfree_cp = conn-\u003ec_path;\nnet/rds/connection.c:352:\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-353-\t\tconn = ERR_PTR(ret);\n--\nnet/rds/connection.c-379-\t\t\tfree_cp = conn-\u003ec_path;\nnet/rds/connection.c:380:\t\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-381-\t\t\tconn = ERR_PTR(-ENETDOWN);\n--\nnet/rds/connection.c-387-\t\t\tfree_cp = conn-\u003ec_path;\nnet/rds/connection.c:388:\t\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-389-\t\t\tif (READ_ONCE(passive-\u003ec_destroy_in_prog)) {\n--\nnet/rds/connection.c-421-\t\t\tfree_cp = conn-\u003ec_path;\nnet/rds/connection.c:422:\t\t\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-423-\t\t\tconn = found;\n--\nnet/rds/connection.c=664=static void rds_conn_destroy_fini(struct kref *kref)\n--\nnet/rds/connection.c-675-\tkfree(conn-\u003ec_path);\nnet/rds/connection.c:676:\tkmem_cache_free(rds_conn_slab, conn);\nnet/rds/connection.c-677-\n--\nnet/rds/connection.c=1173=int rds_conn_init(void)\n--\nnet/rds/connection.c-1180-\nnet/rds/connection.c:1181:\trds_conn_slab = KMEM_CACHE(rds_connection, 0);\nnet/rds/connection.c:1182:\tif (!rds_conn_slab) {\nnet/rds/connection.c-1183-\t\trds_loop_net_exit();\n--\nnet/rds/connection.c=1202=void rds_conn_exit(void)\n--\nnet/rds/connection.c-1208-\nnet/rds/connection.c:1209:\tkmem_cache_destroy(rds_conn_slab);\nnet/rds/connection.c-1210-\n"
}

21/3 2026/09/21 06:15 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 26432
							output: 44
							thoughts: 87
						

Thoughts:
Here's my thought process, as it relates to the code snippet:

**Connection Initialization and IB Node Detachment**

Okay, so I see `conn = kmem_cache_zalloc(rds_conn_slab, gfp);`. This confirms that we're allocating memory for a new connection using the kernel memory cache with zeroing. That `zalloc` is important, because it guarantees that `c_destroy_in_prog` will be initialized to `false`. That's good, predictable behavior. I don't have to worry about garbage values or uninitialized memory corruption there.

Now, my mind wanders a bit. I know we are dealing with connection setup. I wonder about the interaction with the InfiniBand node detachment process, specifically the `i_ib_node_detached` flag within the `struct rds_ib_connection`. This is potentially a race condition or a tricky synchronization point. Where is *that* structure being allocated and initialized in relation to this connection setup? Need to get the full context to ensure this is sound.

Time to reach for the trusty `grepper`... I need to find the allocation site for `struct rds_ib_connection`. That's where the answer to how the system is initializing the IB detach state will live.



22/3 2026/09/21 06:15 tool grepper
0m Args:
{
  "Expression": "kzalloc.*rds_ib_connection|kmem_cache_zalloc.*ib_conn",
  "PathPrefix": "net/rds/"
}

Results:
{
  "Output": "net/rds/ib_cm.c=1233=int rds_ib_conn_alloc(struct rds_connection *conn, gfp_t gfp)\n--\nnet/rds/ib_cm.c-1239-\t/* XXX too lazy? */\nnet/rds/ib_cm.c:1240:\tic = kzalloc_obj(struct rds_ib_connection, gfp);\nnet/rds/ib_cm.c-1241-\tif (!ic)\n"
}

23/3 2026/09/21 06:15 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 26968
							output: 44
							thoughts: 69
						

Thoughts:
Here's my summary, as if I were thinking these things:

**Kernel Memory Allocation Investigation**

Okay, so `kzalloc_obj(struct rds_ib_connection, gfp)` is pretty standard, it zero-initializes the allocated memory, which is good. That saves me a step. But I'm curious about the TCP side of things, specifically the `t_tcp_node_detached` member within the `struct rds_tcp_connection`. I need to figure out how *that* memory is being handled. I need to know where it's being allocated.

Right, so to understand its life cycle, the next logical step is to use `grepper` and see where `struct rds_tcp_connection` itself is allocated within the codebase. That should give me the context I need.



24/3 2026/09/21 06:15 tool grepper
0m Args:
{
  "Expression": "kzalloc.*rds_tcp_connection|kmem_cache_zalloc.*tcp_conn",
  "PathPrefix": "net/rds/"
}

Results:
{
  "Output": "net/rds/tcp.c=442=static int rds_tcp_conn_alloc(struct rds_connection *conn, gfp_t gfp)\n--\nnet/rds/tcp.c-448-\tfor (i = 0; i \u003c RDS_MPATH_WORKERS; i++) {\nnet/rds/tcp.c:449:\t\ttc = kmem_cache_zalloc(rds_tcp_conn_slab, gfp);\nnet/rds/tcp.c-450-\t\tif (!tc) {\n"
}

25/3 2026/09/21 06:15 llm kmsan-evaluator
0m Model:
gemini-3.8-flash

Args:
null

Results:
null

Tokens:
							input: 27397
							output: 281
							thoughts: 876
						

Thoughts:
**KMSAN Assessment: RDS Connection Lifetime Management Patch**

Okay, let's break this down. My initial thought was to double-check the memory initialization. `kmem_cache_zalloc` does zero the memory, so the allocation itself should be fine. I had to look into the struct rds_transport, which is globally/statically defined, so BSS zero-initialization applies and atomic operations are used.

Then, I went through the checklist. There are no complex allocations with potential uninitialized reads since everything is zeroed. Also, there are no buffer-related concerns.

The core of this patch is about connection lifetime management, specifically reference counting. It's designed to address use-after-free conditions and race conditions that occur during connection teardown. Previously, `rds_conn_destroy()` synchronously freed connections; that's the problem. Threads were likely still using the connection. The `kref` implementation is the fix. Bugs like these are textbook UAFs, race conditions, and synchronization issues – classic cases KASAN and LOCKDEP are built to detect. No uninitialized memory usage is involved.

Now, let's apply the rule. We are *specifically* told to avoid KMSAN if other standard bug detectors can catch the bugs. So, I need to check if there are any risks related to uninitialized memory.

Based on the nature of the patch, it seems a clear `NeedsKMSAN=false` situation.

So, here's my reasoning, structured for clarity:

1.  **Patch Nature:** This patch is fundamentally about reference counting using `kref` for `struct rds_connection`. It's ensuring proper lifetime management across the RDS core and transport drivers (IB, TCP, Loop). The key is synchronizing teardown and safely unlinking nodes. The patch prevents connections from being freed until *all* references are gone, eliminating the UAF.
2.  **Memory Safety/Sanitizer Applicability:** The risks addressed and potentially introduced are *concurrency* related, namely UAFs, race conditions, and locking/deadlock issues. These are already effectively addressed by KASAN and LOCKDEP.
3.  **KMSAN Applicability:** All involved structures (`struct rds_connection`, etc.) are zeroed via `kmem_cache_zalloc` or `kzalloc_obj`. Fields like `c_destroy_in_prog`, `c_refcount`, etc. are guaranteed to be initialized. There is *no* chance of uninitialized memory usage or info-leaks, and no buffer sizing logic is modified. Therefore, KMSAN is not justified here.

Therefore, it's clear: `NeedsKMSAN=false`. Let's set the results and move on.



26/3 2026/09/21 06:15 tool set-results
0m Args:
{
  "KMSANReasoning": "The patch introduces reference counting (kref) for 'struct rds_connection' to manage connection lifetimes safely and prevent use-after-free and concurrency races during connection teardown, module unload, and asynchronous operations (across RDS core, TCP, Loop, and IB transports).\n\nKASAN vs KMSAN applicability:\n- The primary failure modes and risks associated with these changes are use-after-free, double-free, reference leaks, and locking issues/deadlocks. These are covered by KASAN, KASAN-quarantine, and LOCKDEP in standard debug builds.\n- All touched structures (including newly added fields such as 'c_refcount', 'c_destroy_in_prog', 'i_ib_node_detached', and 't_tcp_node_detached') are allocated using zeroing allocators (kmem_cache_zalloc, kzalloc_obj) or are static/atomic variables initialized to zero.\n- No uninitialized stack/heap memory or structure padding is read or exposed across security boundaries (e.g. copied to userspace or passed uninitialized to kernel subsystems).\n\nConsequently, there are no uninitialized memory hazards that would necessitate a dedicated KMSAN fuzzing session.",
  "NeedsKMSAN": false
}

Results:
{
  "KMSANReasoning": "The patch introduces reference counting (kref) for 'struct rds_connection' to manage connection lifetimes safely and prevent use-after-free and concurrency races during connection teardown, module unload, and asynchronous operations (across RDS core, TCP, Loop, and IB transports).\n\nKASAN vs KMSAN applicability:\n- The primary failure modes and risks associated with these changes are use-after-free, double-free, reference leaks, and locking issues/deadlocks. These are covered by KASAN, KASAN-quarantine, and LOCKDEP in standard debug builds.\n- All touched structures (including newly added fields such as 'c_refcount', 'c_destroy_in_prog', 'i_ib_node_detached', and 't_tcp_node_detached') are allocated using zeroing allocators (kmem_cache_zalloc, kzalloc_obj) or are static/atomic variables initialized to zero.\n- No uninitialized stack/heap memory or structure padding is read or exposed across security boundaries (e.g. copied to userspace or passed uninitialized to kernel subsystems).\n\nConsequently, there are no uninitialized memory hazards that would necessitate a dedicated KMSAN fuzzing session.",
  "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)