mirror of
https://github.com/python/cpython.git
synced 2026-07-26 03:43:11 -04:00
Miss Islington (bot)
e9d9fd9c80
gh-154431: Fix data race in `sys.audithook` (GH-154462)
(cherry picked from commit 596cd5c5d7)
Co-authored-by: sobolevn <mail@sobolevn.me>
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
import sys
|
|
import unittest
|
|
from test.support import threading_helper
|
|
|
|
|
|
class SysModuleTest(unittest.TestCase):
|
|
@unittest.skipUnless(hasattr(sys, "setdlopenflags"),
|
|
"test needs sys.setdlopenflags()")
|
|
def test_dlopenflags_concurrent(self):
|
|
# gh-151644: getdlopenflags() and setdlopenflags() must be safe to
|
|
# call concurrently in free-threaded builds.
|
|
original = sys.getdlopenflags()
|
|
self.addCleanup(sys.setdlopenflags, original)
|
|
|
|
# Use a small set of known-valid flag values to avoid integer overflow.
|
|
flag_values = [1, 2, 256, 257]
|
|
|
|
def worker(worker_id):
|
|
for i in range(20_000):
|
|
if worker_id % 2 == 0:
|
|
sys.getdlopenflags()
|
|
else:
|
|
sys.setdlopenflags(flag_values[worker_id % len(flag_values)])
|
|
|
|
workers = [lambda i=i: worker(i) for i in range(6)]
|
|
threading_helper.run_concurrently(workers)
|
|
|
|
def test_int_max_str_digits_thread(self):
|
|
# gh-151218: Check that it's safe to call get_int_max_str_digits() and
|
|
# set_int_max_str_digits() in parallel. Previously, this test triggered
|
|
# warnings in TSan on a free threaded build.
|
|
|
|
old_limit = sys.get_int_max_str_digits()
|
|
self.addCleanup(sys.set_int_max_str_digits, old_limit)
|
|
|
|
def worker(worker_id):
|
|
if not worker_id:
|
|
for i in range (20_000):
|
|
sys.get_int_max_str_digits()
|
|
else:
|
|
for i in range (20_000):
|
|
sys.set_int_max_str_digits(4300 + (i & 7))
|
|
|
|
workers = [lambda: worker(i) for i in range(5)]
|
|
threading_helper.run_concurrently(workers)
|
|
|
|
def test_sys_audit_hooks(self):
|
|
def _hook(*args):
|
|
return None
|
|
|
|
def adder():
|
|
for _ in range(100):
|
|
sys.addaudithook(_hook)
|
|
|
|
def auditor():
|
|
for _ in range(2000):
|
|
sys.audit("fusil.tsan.test")
|
|
|
|
threading_helper.run_concurrently([adder, auditor])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|