tracing thread safety

Include/internal/pycore_ceval_state.h

@@ -63,6 +63,7 @@ struct _ceval_runtime_state {
  } perf;
  /* Pending calls to be made only on the main thread. */
  struct _pending_calls pending_mainthread;
+ PyMutex sys_trace_profile_mutex;
 };
 
 #ifdef PY_HAVE_PERF_TRAMPOLINE

Include/internal/pycore_pyatomic_ft_wrappers.h

@@ -23,18 +23,25 @@ extern "C" {
 #define FT_ATOMIC_LOAD_SSIZE(value) _Py_atomic_load_ssize(&value)
 #define FT_ATOMIC_LOAD_SSIZE_RELAXED(value) \
  _Py_atomic_load_ssize_relaxed(&value)
+#define FT_ATOMIC_LOAD_PTR_ACQUIRE(value) \
+ _Py_atomic_load_ptr_acquire(&value)
 #define FT_ATOMIC_STORE_PTR_RELAXED(value, new_value) \
  _Py_atomic_store_ptr_relaxed(&value, new_value)
 #define FT_ATOMIC_STORE_PTR_RELEASE(value, new_value) \
  _Py_atomic_store_ptr_release(&value, new_value)
 #define FT_ATOMIC_STORE_SSIZE_RELAXED(value, new_value) \
  _Py_atomic_store_ssize_relaxed(&value, new_value)
+#define FT_ATOMIC_STORE_UINT8_RELAXED(value, new_value) \
+ _Py_atomic_store_uint8_relaxed(&value, new_value)
 #else
 #define FT_ATOMIC_LOAD_SSIZE(value) value
 #define FT_ATOMIC_LOAD_SSIZE_RELAXED(value) value
+#define FT_ATOMIC_LOAD_PTR_ACQUIRE(value) value
 #define FT_ATOMIC_STORE_PTR_RELAXED(value, new_value) value = new_value
 #define FT_ATOMIC_STORE_PTR_RELEASE(value, new_value) value = new_value
 #define FT_ATOMIC_STORE_SSIZE_RELAXED(value, new_value) value = new_value
+#define FT_ATOMIC_STORE_UINT8_RELAXED(value, new_value) value = new_value
+
 #endif
 
 #ifdef __cplusplus

Lib/test/test_free_threading/__init__.py

@@ -0,0 +1,7 @@
+import os
+
+from test import support
+
+
+def load_tests(*args):
+ return support.load_package_tests(os.path.dirname(__file__), *args)

Lib/test/test_free_threading/test_monitoring.py

@@ -0,0 +1,232 @@
+"""Tests monitoring, sys.settrace, and sys.setprofile in a multi-threaded
+environmenet to verify things are thread-safe in a free-threaded build"""
+
+import sys
+import time
+import unittest
+import weakref
+
+from sys import monitoring
+from test.support import is_wasi
+from threading import Thread
+from unittest import TestCase
+
+
+class InstrumentationMultiThreadedMixin:
+ if not hasattr(sys, "gettotalrefcount"):
+ thread_count = 50
+ func_count = 1000
+ fib = 15
+ else:
+ # Run a little faster in debug builds...
+ thread_count = 25
+ func_count = 500
+ fib = 15
+
+ def after_threads(self):
+ """Runs once after all the threads have started"""
+ pass
+
+ def during_threads(self):
+ """Runs repeatedly while the threads are still running"""
+ pass
+
+ def work(self, n, funcs):
+ """Fibonacci function which also calls a bunch of random functions"""
+ for func in funcs:
+ func()
+ if n < 2:
+ return n
+ return self.work(n - 1, funcs) + self.work(n - 2, funcs)
+
+ def start_work(self, n, funcs):
+ # With the GIL builds we need to make sure that the hooks have
+ # a chance to run as it's possible to run w/o releasing the GIL.
+ time.sleep(1)
+ self.work(n, funcs)
+
+ def after_test(self):
+ """Runs once after the test is done"""
+ pass
+
+ def test_instrumention(self):
+ # Setup a bunch of functions which will need instrumentation...
+ funcs = []
+ for i in range(self.func_count):
+ x = {}
+ exec("def f(): pass", x)
+ funcs.append(x["f"])
+
+ threads = []
+ for i in range(self.thread_count):
+ # Each thread gets a copy of the func list to avoid contention
+ t = Thread(target=self.start_work, args=(self.fib, list(funcs)))
+ t.start()
+ threads.append(t)
+
+ self.after_threads()
+
+ while True:
+ any_alive = False
+ for t in threads:
+ if t.is_alive():
+ any_alive = True
+ break
+
+ if not any_alive:
+ break
+
+ self.during_threads()
+
+ self.after_test()
+
+
+class MonitoringTestMixin:
+ def setUp(self):
+ for i in range(6):
+ if monitoring.get_tool(i) is None:
+ self.tool_id = i
+ monitoring.use_tool_id(i, self.__class__.__name__)
+ break
+
+ def tearDown(self):
+ monitoring.free_tool_id(self.tool_id)
+
+
+@unittest.skipIf(is_wasi, "WASI has no threads.")
+class SetPreTraceMultiThreaded(InstrumentationMultiThreadedMixin, TestCase):
+ """Sets tracing one time after the threads have started"""
+
+ def setUp(self):
+ super().setUp()
+ self.called = False
+
+ def after_test(self):
+ self.assertTrue(self.called)
+
+ def trace_func(self, frame, event, arg):
+ self.called = True
+ return self.trace_func
+
+ def after_threads(self):
+ sys.settrace(self.trace_func)
+
+
+@unittest.skipIf(is_wasi, "WASI has no threads.")
+class MonitoringMultiThreaded(
+ MonitoringTestMixin, InstrumentationMultiThreadedMixin, TestCase
+):
+ """Uses sys.monitoring and repeatedly toggles instrumentation on and off"""
+
+ def setUp(self):
+ super().setUp()
+ self.set = False
+ self.called = False
+ monitoring.register_callback(
+ self.tool_id, monitoring.events.LINE, self.callback
+ )
+
+ def tearDown(self):
+ monitoring.set_events(self.tool_id, 0)
+ super().tearDown()
+
+ def callback(self, *args):
+ self.called = True
+
+ def after_test(self):
+ self.assertTrue(self.called)
+
+ def during_threads(self):
+ if self.set:
+ monitoring.set_events(
+ self.tool_id, monitoring.events.CALL | monitoring.events.LINE
+ )
+ else:
+ monitoring.set_events(self.tool_id, 0)
+ self.set = not self.set
+
+
+@unittest.skipIf(is_wasi, "WASI has no threads.")
+class SetTraceMultiThreaded(InstrumentationMultiThreadedMixin, TestCase):
+ """Uses sys.settrace and repeatedly toggles instrumentation on and off"""
+
+ def setUp(self):
+ self.set = False
+ self.called = False
+
+ def after_test(self):
+ self.assertTrue(self.called)
+
+ def tearDown(self):
+ sys.settrace(None)
+
+ def trace_func(self, frame, event, arg):
+ self.called = True
+ return self.trace_func
+
+ def during_threads(self):
+ if self.set:
+ sys.settrace(self.trace_func)
+ else:
+ sys.settrace(None)
+ self.set = not self.set
+
+
+@unittest.skipIf(is_wasi, "WASI has no threads.")
+class SetProfileMultiThreaded(InstrumentationMultiThreadedMixin, TestCase):
+ """Uses sys.setprofile and repeatedly toggles instrumentation on and off"""
+ thread_count = 25
+ func_count = 200
+ fib = 15
+
+ def setUp(self):
+ self.set = False
+ self.called = False
+
+ def after_test(self):
+ self.assertTrue(self.called)
+
+ def tearDown(self):
+ sys.setprofile(None)
+
+ def trace_func(self, frame, event, arg):
+ self.called = True
+ return self.trace_func
+
+ def during_threads(self):
+ if self.set:
+ sys.setprofile(self.trace_func)
+ else:
+ sys.setprofile(None)
+ self.set = not self.set
+
+
+@unittest.skipIf(is_wasi, "WASI has no threads.")
+class MonitoringMisc(MonitoringTestMixin, TestCase):
+ def register_callback(self):
+ def callback(*args):
+ pass
+
+ for i in range(200):
+ monitoring.register_callback(self.tool_id, monitoring.events.LINE, callback)
+
+ self.refs.append(weakref.ref(callback))
+
+ def test_register_callback(self):
+ self.refs = []
+ threads = []
+ for i in range(50):
+ t = Thread(target=self.register_callback)
+ t.start()
+ threads.append(t)
+
+ for thread in threads:
+ thread.join()
+
+ monitoring.register_callback(self.tool_id, monitoring.events.LINE, None)
+ for ref in self.refs:
+ self.assertEqual(ref(), None)
+
+
+if __name__ == "__main__":
+ unittest.main()

Makefile.pre.in

@@ -2374,6 +2374,7 @@ TESTSUBDIRS= idlelib/idle_test \
  test/test_doctest \
  test/test_email \
  test/test_email/data \
+ test/test_free_threading \
  test/test_future_stmt \
  test/test_gdb \
  test/test_import \