test: add regression test

Author: KowalskiThomas

Lib/test/test_free_threading/test_list.py

@@ -1,3 +1,4 @@
+import threading
 import unittest
 
 from threading import Thread, Barrier
@@ -171,5 +172,125 @@ def size_function():
             pass
 
 
+    def test_richcompare_stale_element_list1(self) -> None:
+        # gh-148442: list_richcompare_impl must keep references to the
+        # captured items for the final ordering comparison, not re-read the
+        # list slots after the critical section may have been suspended.
+        #
+        # list1 = [x, 1], list2 = [0, 0] with x > 0.
+        # During list1 > list2, list1[0] is mutated from x to 0.
+        # The result must be True (x > 0), not False (0 > 0).
+
+        class Value:
+            def __init__(self, v: int) -> None:
+                self.v = v
+
+            def __eq__(self, other: object) -> bool:
+                if isinstance(other, Value):
+                    return self.v == other.v
+                return NotImplemented
+
+            def __gt__(self, other: object) -> bool:
+                if isinstance(other, Value):
+                    return self.v > other.v
+                return NotImplemented
+
+            def __hash__(self) -> int:
+                return hash(self.v)
+
+        eq_started = threading.Event()
+        swap_done = threading.Event()
+
+        class SignalingValue(Value):
+            # Value whose __eq__ signals eq_started then waits for swap_done,
+            # giving another thread the window to mutate the list slot.
+            def __eq__(self, other: object) -> bool:
+                eq_started.set()
+                swap_done.wait()
+                return super().__eq__(other)
+
+        x = SignalingValue(5)  # list1[0]; x > 0
+        list1: list[Value] = [x, Value(1)]
+        list2: list[Value] = [Value(0), Value(0)]
+
+        result: list[bool] = []
+
+        def compare() -> None:
+            result.append(list1 > list2)
+
+        def swap() -> None:
+            eq_started.wait()
+            list1[0] = Value(0)  # replace x(5) with 0 -- must not change result
+            swap_done.set()
+
+        t_cmp = Thread(target=compare)
+        t_swp = Thread(target=swap)
+        t_cmp.start()
+        t_swp.start()
+        t_cmp.join()
+        t_swp.join()
+
+        # x(5) > Value(0) is True; old code would compare Value(0) > Value(0) -> False
+        self.assertTrue(result[0])
+
+    def test_richcompare_stale_element_list2(self) -> None:
+        # Same as test_richcompare_stale_element_list1 but list2[0] is mutated
+        # to a value larger than x, which would flip the result under the old code.
+        #
+        # list1 = [x, 1], list2 = [0, 0] with x > 0.
+        # During list1 > list2, list2[0] is mutated from 0 to 100 (> x).
+        # The result must be True (x > 0), not False (x > 100).
+
+        class Value:
+            def __init__(self, v: int) -> None:
+                self.v = v
+
+            def __eq__(self, other: object) -> bool:
+                if isinstance(other, Value):
+                    return self.v == other.v
+                return NotImplemented
+
+            def __gt__(self, other: object) -> bool:
+                if isinstance(other, Value):
+                    return self.v > other.v
+                return NotImplemented
+
+            def __hash__(self) -> int:
+                return hash(self.v)
+
+        eq_started = threading.Event()
+        swap_done = threading.Event()
+
+        class SignalingValue(Value):
+            def __eq__(self, other: object) -> bool:
+                eq_started.set()
+                swap_done.wait()
+                return super().__eq__(other)
+
+        x = SignalingValue(5)  # list1[0]; x > 0
+        list1: list[Value] = [x, Value(1)]
+        list2: list[Value] = [Value(0), Value(0)]
+
+        result: list[bool] = []
+
+        def compare() -> None:
+            result.append(list1 > list2)
+
+        def swap() -> None:
+            eq_started.wait()
+            list2[0] = Value(100)  # replace 0 with 100 (> x) -- must not change result
+            swap_done.set()
+
+        t_cmp = Thread(target=compare)
+        t_swp = Thread(target=swap)
+        t_cmp.start()
+        t_swp.start()
+        t_cmp.join()
+        t_swp.join()
+
+        # x(5) > Value(0) is True; old code would compare Value(5) > Value(100) -> False
+        self.assertTrue(result[0])
+
+
 if __name__ == "__main__":
     unittest.main()