Reproduce race in Thread.join()

mpage · mpage · commit 81946532792f · 2024-01-25T16:30:50.000-08:00
diff --git a/Lib/threading.py b/Lib/threading.py
@@ -6,6 +6,7 @@
 import functools
 import warnings
 
+import time
 from time import monotonic as _time
 from _weakrefset import WeakSet
 from itertools import count as _count
@@ -1112,6 +1113,7 @@ def _stop(self):
         # module's _shutdown() function.
         lock = self._tstate_lock
         if lock is not None:
+            self._log("Checking _tstate_lock is unlocked")
             assert not lock.locked()
         self._is_stopped = True
         self._tstate_lock = None
@@ -1182,6 +1184,11 @@ def _join_os_thread(self):
                 self._handle = None
                 # No need to keep this around
                 self._join_lock = None
+    def _log(self, msg):
+        cur_thr_name = current_thread()._name
+        if cur_thr_name not in ("join-race-B", "join-race-C"):
+            return
+        print(f"[{current_thread()._name} join {self._name}] - {msg}")
 
     def _wait_for_tstate_lock(self, block=True, timeout=-1):
         # Issue #18808: wait for the thread state to be gone.
@@ -1196,9 +1203,19 @@ def _wait_for_tstate_lock(self, block=True, timeout=-1):
             assert self._is_stopped
             return
 
+        cur_thr_name = current_thread()._name
         try:
             if lock.acquire(block, timeout):
+                self._log(f"Acquired _tstate_lock for {self._name}")
+                if cur_thr_name =="join-race-C":
+                    self._log("Sleeping for 2")
+                    time.sleep(2)
+                self._log("Releasing _tstate_lock")
                 lock.release()
+                if cur_thr_name == "join-race-B":
+                    self._log("Sleeping for 0.5")
+                    time.sleep(0.5)
+                self._log("Calling _stop")
                 self._stop()
         except:
             if lock.locked():
diff --git a/repro_join_race.py b/repro_join_race.py
@@ -0,0 +1,47 @@
+import os
+import threading
+import traceback
+
+
+def waiter(event: threading.Event) -> None:
+    event.wait()
+
+
+def joiner(thr: threading.Thread) -> None:
+    try:
+        thr.join()
+    except AssertionError as exc:
+        traceback.print_exc()
+        os._exit(1)
+
+
+def repro() -> None:
+    event = threading.Event()
+    threads = []
+    threads.append(threading.Thread(target=waiter, name="join-race-A", args=(event,)))
+    threads.append(threading.Thread(target=joiner, name="join-race-B", args=(threads[0],)))
+    threads.append(threading.Thread(target=joiner, name="join-race-C", args=(threads[0],)))
+    for thr in threads:
+        thr.start()
+    # Unblock waiter
+    event.set()
+
+    # Wait for joiners to exit. We must allow the joiner threads to wait first,
+    # otherwise we may wait on the _tstate_lock first, acquire it, and set it
+    # to None before either of the joiners have a chance to do so.
+    threads[1].join()
+    threads[2].join()
+
+    # Wait for waiter to exit
+    threads[0].join()
+
+
+def main() -> None:
+    for i in range(1000):
+        print(f"=== Attempt {i} ===")
+        repro()
+        print()
+
+
+if __name__ == "__main__":
+    main()
