diff --git a/Doc/library/bisect.rst b/Doc/library/bisect.rst index 78da563397b625..d02ffe469adb1a 100644 --- a/Doc/library/bisect.rst +++ b/Doc/library/bisect.rst @@ -24,6 +24,16 @@ method to determine whether a value has been found. Instead, the functions only call the :meth:`~object.__lt__` method and will return an insertion point between values in an array. +.. note:: + + The functions in this module are not thread-safe. If multiple threads + concurrently use :mod:`bisect` functions on the same sequence, this + may result in undefined behaviour. Likewise, if the provided sequence + is mutated by a different thread while a :mod:`bisect` function + is operating on it, the result is undefined. For example, using + :py:func:`~bisect.insort_left` on the same list from multiple threads + may result in the list becoming unsorted. + .. _bisect functions: The following functions are provided: diff --git a/Lib/test/test_free_threading/test_bisect.py b/Lib/test/test_free_threading/test_bisect.py new file mode 100644 index 00000000000000..bd7732da035257 --- /dev/null +++ b/Lib/test/test_free_threading/test_bisect.py @@ -0,0 +1,56 @@ +import unittest +from test.support import import_helper, threading_helper +import random + +py_bisect = import_helper.import_fresh_module('bisect', blocked=['_bisect']) +c_bisect = import_helper.import_fresh_module('bisect', fresh=['_bisect']) + + +NTHREADS = 4 +OBJECT_COUNT = 500 + + +class TestBase: + def do_racing_insort(self, insert_method): + def insert(data): + for _ in range(OBJECT_COUNT): + x = random.randint(-OBJECT_COUNT, OBJECT_COUNT) + insert_method(data, x) + + data = list(range(OBJECT_COUNT)) + threading_helper.run_concurrently( + worker_func=insert, args=(data,), nthreads=NTHREADS + ) + if False: + # These functions are not thread-safe and so the list can become + # unsorted. However, we don't want Python to crash if these + # functions are used concurrently on the same sequence. This + # should also not produce any TSAN warnings. + self.assertTrue(self.is_sorted_ascending(data)) + + def test_racing_insert_right(self): + self.do_racing_insort(self.mod.insort_right) + + def test_racing_insert_left(self): + self.do_racing_insort(self.mod.insort_left) + + @staticmethod + def is_sorted_ascending(lst): + """ + Check if the list is sorted in ascending order (non-decreasing). + """ + return all(lst[i - 1] <= lst[i] for i in range(1, len(lst))) + + +@threading_helper.requires_working_threading() +class TestPyBisect(unittest.TestCase, TestBase): + mod = py_bisect + + +@threading_helper.requires_working_threading() +class TestCBisect(unittest.TestCase, TestBase): + mod = c_bisect + + +if __name__ == "__main__": + unittest.main()