--- /dev/null
+import unittest
+
+from threading import Thread, Barrier
+from unittest import TestCase
+
+from test.support import threading_helper
+
+
+@threading_helper.requires_working_threading()
+class TestSet(TestCase):
+ def test_repr_clear(self):
+ """Test repr() of a set while another thread is calling clear()"""
+ NUM_ITERS = 10
+ NUM_REPR_THREADS = 10
+ barrier = Barrier(NUM_REPR_THREADS + 1)
+ s = {1, 2, 3, 4, 5, 6, 7, 8}
+
+ def clear_set():
+ barrier.wait()
+ s.clear()
+
+ def repr_set():
+ barrier.wait()
+ set_reprs.append(repr(s))
+
+ for _ in range(NUM_ITERS):
+ set_reprs = []
+ threads = [Thread(target=clear_set)]
+ for _ in range(NUM_REPR_THREADS):
+ threads.append(Thread(target=repr_set))
+ for t in threads:
+ t.start()
+ for t in threads:
+ t.join()
+
+ for set_repr in set_reprs:
+ self.assertIn(set_repr, ("set()", "{1, 2, 3, 4, 5, 6, 7, 8}"))
+
+
+if __name__ == "__main__":
+ unittest.main()
return PyUnicode_FromFormat("%s()", Py_TYPE(so)->tp_name);
}
- keys = PySequence_List((PyObject *)so);
- if (keys == NULL)
+ // gh-129967: avoid PySequence_List because it might re-lock the object
+ // lock or the GIL and allow something to clear the set from underneath us.
+ keys = PyList_New(so->used);
+ if (keys == NULL) {
goto done;
+ }
+
+ Py_ssize_t pos = 0, idx = 0;
+ setentry *entry;
+ while (set_next(so, &pos, &entry)) {
+ PyList_SET_ITEM(keys, idx++, Py_NewRef(entry->key));
+ }
/* repr(keys)[1:-1] */
listrepr = PyObject_Repr(keys);