gh-158600: Fix lost updates in concurrent set.intersection_update()

Calculate the intersection from a copy while holding the target set's
critical section. This prevents concurrent intersection_update() calls from
overwriting one another's results.

Add free-threading tests for one-operand and multi-operand calls, with
__iand__() as a control.

Author: CaQtiml

Lib/test/test_free_threading/test_set.py

@@ -148,6 +148,98 @@ def read_set():
         for t in threads:
             t.join()
 
+    def test_intersection_update_concurrent(self):
+        """Test one-operand intersection updates of one shared set."""
+        NUM_ITERS = 10
+        BLOCK_SIZE = self.SET_SIZE * 100
+
+        sources = [
+            set(range(4 * BLOCK_SIZE)),
+            set(range(3 * BLOCK_SIZE)),
+            set(range(2 * BLOCK_SIZE)),
+            set(range(BLOCK_SIZE)),
+        ]
+        expected = set(range(BLOCK_SIZE))
+
+        for _ in range(NUM_ITERS):
+            target = set(range(5 * BLOCK_SIZE))
+            barrier = Barrier(len(sources), timeout=2)
+
+            def intersect(source):
+                barrier.wait()
+                target.intersection_update(source)
+
+            threads = [Thread(target=intersect, args=(source,))
+                       for source in sources]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join()
+
+            self.assertEqual(target, expected)
+
+    def test_intersection_update_multiple_concurrent(self):
+        """Test multi-operand intersection updates of one shared set."""
+        NUM_ITERS = 10
+        BLOCK_SIZE = self.SET_SIZE * 100
+
+        evens = set(range(0, 4 * BLOCK_SIZE, 2))
+        below_three_blocks = set(range(3 * BLOCK_SIZE))
+        multiples_of_three = set(range(0, 2 * BLOCK_SIZE, 3))
+        below_one_block = set(range(BLOCK_SIZE))
+        expected = set(range(0, BLOCK_SIZE, 6))
+
+        for _ in range(NUM_ITERS):
+            target = set(range(5 * BLOCK_SIZE))
+            barrier = Barrier(2, timeout=2)
+
+            def intersect(first, second):
+                barrier.wait()
+                target.intersection_update(first, second)
+
+            threads = [
+                Thread(target=intersect,
+                       args=(evens, below_three_blocks)),
+                Thread(target=intersect,
+                       args=(multiples_of_three, below_one_block)),
+            ]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join()
+
+            self.assertEqual(target, expected)
+
+    def test_iand_concurrent(self):
+        """Test concurrent &= operations on one shared set."""
+        NUM_ITERS = 10
+        BLOCK_SIZE = self.SET_SIZE * 100
+
+        sources = [
+            set(range(4 * BLOCK_SIZE)),
+            set(range(3 * BLOCK_SIZE)),
+            set(range(2 * BLOCK_SIZE)),
+            set(range(BLOCK_SIZE)),
+        ]
+        expected = set(range(BLOCK_SIZE))
+
+        for _ in range(NUM_ITERS):
+            target = set(range(5 * BLOCK_SIZE))
+            barrier = Barrier(len(sources), timeout=2)
+
+            def intersect(source):
+                barrier.wait()
+                target.__iand__(source)
+
+            threads = [Thread(target=intersect, args=(source,))
+                       for source in sources]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join()
+
+            self.assertEqual(target, expected)
+
 
 @threading_helper.requires_working_threading()
 class SmallSetTest(RaceTestBase, unittest.TestCase):

Objects/setobject.c

@@ -1861,15 +1861,25 @@ set_intersection_update_multi_impl(PySetObject *so, PyObject * const *others,
                                    Py_ssize_t others_length)
 /*[clinic end generated code: output=d768b5584675b48d input=782e422fc370e4fc]*/
 {
-    PyObject *tmp;
+    PyObject *copy;
+    PyObject *result = NULL;
 
-    tmp = set_intersection_multi_impl(so, others, others_length);
-    if (tmp == NULL)
-        return NULL;
     Py_BEGIN_CRITICAL_SECTION(so);
-    set_swap_bodies(so, (PySetObject *)tmp);
+    copy = set_copy_untracked_lock_held(so);
+    if (copy != NULL) {
+        result = set_intersection_multi_impl((PySetObject *)copy,
+                                             others, others_length);
+        Py_DECREF(copy);
+        if (result != NULL) {
+            set_swap_bodies(so, (PySetObject *)result);
+        }
+    }
     Py_END_CRITICAL_SECTION();
-    Py_DECREF(tmp);
+
+    if (result == NULL) {
+        return NULL;
+    }
+    Py_DECREF(result);
     Py_RETURN_NONE;
 }