import unittest
-from collections import deque
+from collections import OrderedDict, deque
from copy import copy
from test.support import threading_helper
)
+class TestOrderedDict(unittest.TestCase):
+ def test_iterator_update_clear_race(self):
+ # gh-151627: OrderedDict iterator construction must not race with
+ # concurrent clear()/update() operations that mutate the linked list.
+ od = OrderedDict((i, i) for i in range(100))
+
+ def mutate():
+ for i in range(5000):
+ od.clear()
+ od.update(((i, i), (i + 1, i + 1), (i + 2, i + 2)))
+
+ def iterate():
+ for _ in range(5000):
+ try:
+ for _ in od:
+ pass
+ list(reversed(od))
+ except RuntimeError:
+ pass
+
+ threading_helper.run_concurrently(
+ [mutate, *[iterate for _ in range(8)]],
+ )
+
+
if __name__ == "__main__":
unittest.main()
}
di->kind = kind;
+ Py_BEGIN_CRITICAL_SECTION(od);
node = reversed ? _odict_LAST(od) : _odict_FIRST(od);
di->di_current = node ? Py_NewRef(_odictnode_KEY(node)) : NULL;
di->di_size = PyODict_SIZE(od);
di->di_state = od->od_state;
di->di_odict = (PyODictObject*)Py_NewRef(od);
+ Py_END_CRITICAL_SECTION();
_PyObject_GC_TRACK(di);
return (PyObject *)di;