From a3649bc6c198e253fe8587d57d248fcc6464d0a7 Mon Sep 17 00:00:00 2001
From: Sankur Kundu <sankur.kundu.tw@gmail.com>
Date: Wed, 21 Oct 2026 16:59:22 +0000
Subject: [PATCH 1/2] [ring] Only touch the socket task queues while holding
 the lock

SocketThread::worker() checked recvs_/sends_ for emptiness and took their
front() after releasing queue_mutex_, while send() and recv() append to the
same lists from other threads. That is a data race on std::list, which
ThreadSanitizer reports on every run of a send-then-recv loop over a socket
pair.

Look up the front tasks while holding the lock and keep pointers to them.
Appending to a std::list does not move existing elements and only the worker
removes them, so the pointers stay valid once the lock is released. Also take
the lock when failing the pending tasks after repeated errors, and when
setting stop_ in the destructor so the worker cannot miss the wake-up.
---
 mlx/distributed/ring/ring.cpp | 28 +++++++++++++++++++++++-----
 1 file changed, 23 insertions(+), 5 deletions(-)

diff --git a/mlx/distributed/ring/ring.cpp b/mlx/distributed/ring/ring.cpp
index 9a81010e34..4e8a93e7fa 100644
--- a/mlx/distributed/ring/ring.cpp
+++ b/mlx/distributed/ring/ring.cpp
@@ -80,7 +80,10 @@ class SocketThread {
     fcntl(fd, F_SETFL, flags | O_NONBLOCK);
   }
   ~SocketThread() {
-    stop_ = true;
+    {
+      std::unique_lock lock(queue_mutex_);
+      stop_ = true;
+    }
     condition_.notify_all();
     worker_.join();
     int flags = fcntl(fd_, F_GETFL, 0);
@@ -151,6 +154,13 @@ class SocketThread {
     bool delete_recv = false;
     bool delete_send = false;
     while (true) {
+      // The tasks at the front of the queues. They are only looked up while
+      // holding the lock, since send() and recv() append to the queues from
+      // other threads. Appending to a std::list does not move its elements
+      // and only this thread removes them, so the pointers stay valid after
+      // the lock is released.
+      SocketTask* recv_task = nullptr;
+      SocketTask* send_task = nullptr;
       {
         std::unique_lock lock(queue_mutex_);
 
@@ -175,10 +185,17 @@ class SocketThread {
             return;
           }
         }
+
+        if (!recvs_.empty()) {
+          recv_task = &recvs_.front();
+        }
+        if (!sends_.empty()) {
+          send_task = &sends_.front();
+        }
       }
 
-      if (!recvs_.empty()) {
-        auto& task = recvs_.front();
+      if (recv_task) {
+        auto& task = *recv_task;
         ssize_t r =
             ::recv(fd_, task.buffer, std::min(task.size, MAX_IO_BYTES), 0);
         if (r > 0) {
@@ -195,8 +212,8 @@ class SocketThread {
               true, "Receiving from socket", fd_, "failed with errno", errno);
         }
       }
-      if (!sends_.empty()) {
-        auto& task = sends_.front();
+      if (send_task) {
+        auto& task = *send_task;
         ssize_t r =
             ::send(fd_, task.buffer, std::min(task.size, MAX_IO_BYTES), 0);
         if (r > 0) {
@@ -218,6 +235,7 @@ class SocketThread {
         // Throw exception in invoker's thread.
         auto error = std::make_exception_ptr(
             std::runtime_error("[ring] connection to a peer was lost"));
+        std::unique_lock lock(queue_mutex_);
         for (auto& task : recvs_) {
           task.promise.set_exception(error);
         }

From e534838adfa4d933baa84c76e75b442aaaf67e98 Mon Sep 17 00:00:00 2001
From: Cheng <git@zcbenz.com>
Date: Fri, 23 Oct 2026 10:39:08 +0900
Subject: [PATCH 2/2] nit

---
 mlx/distributed/ring/ring.cpp | 5 -----
 1 file changed, 5 deletions(-)

diff --git a/mlx/distributed/ring/ring.cpp b/mlx/distributed/ring/ring.cpp
index 4e8a93e7fa..0726478ebf 100644
--- a/mlx/distributed/ring/ring.cpp
+++ b/mlx/distributed/ring/ring.cpp
@@ -154,11 +154,6 @@ class SocketThread {
     bool delete_recv = false;
     bool delete_send = false;
     while (true) {
-      // The tasks at the front of the queues. They are only looked up while
-      // holding the lock, since send() and recv() append to the queues from
-      // other threads. Appending to a std::list does not move its elements
-      // and only this thread removes them, so the pointers stay valid after
-      // the lock is released.
       SocketTask* recv_task = nullptr;
       SocketTask* send_task = nullptr;
       {