gh-114271: Fix race in Thread.join()

Include/cpython/pystate.h

@@ -161,32 +161,6 @@ struct _ts {
  */
  uintptr_t critical_section;
 
- /* Called when a thread state is deleted normally, but not when it
- * is destroyed after fork().
- * Pain: to prevent rare but fatal shutdown errors (issue 18808),
- * Thread.join() must wait for the join'ed thread's tstate to be unlinked
- * from the tstate chain. That happens at the end of a thread's life,
- * in pystate.c.
- * The obvious way doesn't quite work: create a lock which the tstate
- * unlinking code releases, and have Thread.join() wait to acquire that
- * lock. The problem is that we _are_ at the end of the thread's life:
- * if the thread holds the last reference to the lock, decref'ing the
- * lock will delete the lock, and that may trigger arbitrary Python code
- * if there's a weakref, with a callback, to the lock. But by this time
- * _PyRuntime.gilstate.tstate_current is already NULL, so only the simplest
- * of C code can be allowed to run (in particular it must not be possible to
- * release the GIL).
- * So instead of holding the lock directly, the tstate holds a weakref to
- * the lock: that's the value of on_delete_data below. Decref'ing a
- * weakref is harmless.
- * on_delete points to _threadmodule.c's static release_sentinel() function.
- * After the tstate is unlinked, release_sentinel is called with the
- * weakref-to-lock (on_delete_data) argument, and release_sentinel releases
- * the indirectly held lock.
- */
- void (*on_delete)(void *);
- void *on_delete_data;
-
  int coroutine_origin_tracking_depth;
 
  PyObject *async_gen_firstiter;

Include/internal/pycore_lock.h

@@ -153,16 +153,6 @@ PyAPI_FUNC(void) PyEvent_Wait(PyEvent *evt);
 // and 0 if the timeout expired or thread was interrupted.
 PyAPI_FUNC(int) PyEvent_WaitTimed(PyEvent *evt, PyTime_t timeout_ns);
 
-// A one-time event notification with reference counting.
-typedef struct _PyEventRc {
- PyEvent event;
- Py_ssize_t refcount;
-} _PyEventRc;
-
-_PyEventRc *_PyEventRc_New(void);
-void _PyEventRc_Incref(_PyEventRc *erc);
-void _PyEventRc_Decref(_PyEventRc *erc);
-
 // _PyRawMutex implements a word-sized mutex that that does not depend on the
 // parking lot API, and therefore can be used in the parking lot
 // implementation.

Include/internal/pycore_pythread.h

@@ -78,7 +78,7 @@ struct _pythread_runtime_state {
  } stubs;
 #endif
 
- // Linked list of ThreadHandleObjects
+ // Linked list of ThreadHandles
  struct llist_node handles;
 };
 

Lib/test/test_audit.py

@@ -209,7 +209,7 @@ def test_threading(self):
  expected = [
  ("_thread.start_new_thread", "(<test_func>, (), None)"),
  ("test.test_func", "()"),
- ("_thread.start_joinable_thread", "(<test_func>,)"),
+ ("_thread.start_joinable_thread", "(<test_func>, 1, None)"),
  ("test.test_func", "()"),
  ]
 

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -201,13 +201,13 @@ def test_python_finalization_error(self):
  # QueueFeederThread.
  orig_start_new_thread = threading._start_joinable_thread
  nthread = 0
- def mock_start_new_thread(func, *args):
+ def mock_start_new_thread(func, *args, **kwargs):
  nonlocal nthread
  if nthread >= 1:
  raise RuntimeError("can't create new thread at "
  "interpreter shutdown")
  nthread += 1
- return orig_start_new_thread(func, *args)
+ return orig_start_new_thread(func, *args, **kwargs)
 
  with support.swap_attr(threading, '_start_joinable_thread',
  mock_start_new_thread):

Lib/test/test_thread.py

@@ -289,6 +289,54 @@ def joiner():
  with self.assertRaisesRegex(RuntimeError, "Cannot join current thread"):
  raise error
 
+ def test_join_with_timeout(self):
+ lock = thread.allocate_lock()
+ lock.acquire()
+
+ def thr():
+ lock.acquire()
+
+ with threading_helper.wait_threads_exit():
+ handle = thread.start_joinable_thread(thr)
+ handle.join(0.1)
+ self.assertFalse(handle.is_done())
+ lock.release()
+ handle.join()
+ self.assertTrue(handle.is_done())
+
+ def test_join_unstarted(self):
+ handle = thread._ThreadHandle()
+ with self.assertRaisesRegex(RuntimeError, "thread not started"):
+ handle.join()
+
+ def test_set_done_unstarted(self):
+ handle = thread._ThreadHandle()
+ with self.assertRaisesRegex(RuntimeError, "thread not started"):
+ handle._set_done()
+
+ def test_start_duplicate_handle(self):
+ lock = thread.allocate_lock()
+ lock.acquire()
+
+ def func():
+ lock.acquire()
+
+ handle = thread._ThreadHandle()
+ with threading_helper.wait_threads_exit():
+ thread.start_joinable_thread(func, handle=handle)
+ with self.assertRaisesRegex(RuntimeError, "thread already started"):
+ thread.start_joinable_thread(func, handle=handle)
+ lock.release()
+ handle.join()
+
+ def test_start_with_none_handle(self):
+ def func():
+ pass
+
+ with threading_helper.wait_threads_exit():
+ handle = thread.start_joinable_thread(func, handle=None)
+ handle.join()
+
 
 class Barrier:
  def __init__(self, num_threads):

Lib/test/test_threading.py

@@ -408,7 +408,7 @@ def run(self):
 
  def test_limbo_cleanup(self):
  # Issue 7481: Failure to start thread should cleanup the limbo map.
- def fail_new_thread(*args):
+ def fail_new_thread(*args, **kwargs):
  raise threading.ThreadError()
  _start_joinable_thread = threading._start_joinable_thread
  threading._start_joinable_thread = fail_new_thread
@@ -912,41 +912,6 @@ def f():
  rc, out, err = assert_python_ok("-c", code)
  self.assertEqual(err, b"")
 
- def test_tstate_lock(self):
- # Test an implementation detail of Thread objects.
- started = _thread.allocate_lock()
- finish = _thread.allocate_lock()
- started.acquire()
- finish.acquire()
- def f():
- started.release()
- finish.acquire()
- time.sleep(0.01)
- # The tstate lock is None until the thread is started
- t = threading.Thread(target=f)
- self.assertIs(t._tstate_lock, None)
- t.start()
- started.acquire()
- self.assertTrue(t.is_alive())
- # The tstate lock can't be acquired when the thread is running
- # (or suspended).
- tstate_lock = t._tstate_lock
- self.assertFalse(tstate_lock.acquire(timeout=0), False)
- finish.release()
- # When the thread ends, the state_lock can be successfully
- # acquired.
- self.assertTrue(tstate_lock.acquire(timeout=support.SHORT_TIMEOUT), False)
- # But is_alive() is still True: we hold _tstate_lock now, which
- # prevents is_alive() from knowing the thread's end-of-life C code
- # is done.
- self.assertTrue(t.is_alive())
- # Let is_alive() find out the C code is done.
- tstate_lock.release()
- self.assertFalse(t.is_alive())
- # And verify the thread disposed of _tstate_lock.
- self.assertIsNone(t._tstate_lock)
- t.join()
-
  def test_repr_stopped(self):
  # Verify that "stopped" shows up in repr(Thread) appropriately.
  started = _thread.allocate_lock()
@@ -1112,30 +1077,6 @@ def checker():
  self.assertEqual(threading.getprofile(), old_profile)
  self.assertEqual(sys.getprofile(), old_profile)
 
- @cpython_only
- def test_shutdown_locks(self):
- for daemon in (False, True):
- with self.subTest(daemon=daemon):
- event = threading.Event()
- thread = threading.Thread(target=event.wait, daemon=daemon)
-
- # Thread.start() must add lock to _shutdown_locks,
- # but only for non-daemon thread
- thread.start()
- tstate_lock = thread._tstate_lock
- if not daemon:
- self.assertIn(tstate_lock, threading._shutdown_locks)
- else:
- self.assertNotIn(tstate_lock, threading._shutdown_locks)
-
- # unblock the thread and join it
- event.set()
- thread.join()
-
- # Thread._stop() must remove tstate_lock from _shutdown_locks.
- # Daemon threads must never add it to _shutdown_locks.
- self.assertNotIn(tstate_lock, threading._shutdown_locks)
-
  def test_locals_at_exit(self):
  # bpo-19466: thread locals must not be deleted before destructors
  # are called