with threading_helper.start_threads([t1, t2]):
pass
+ def test_getstate_race_with_shared_keys(self):
+ box = [None]
+ enter = Barrier(2)
+ leave = Barrier(2)
+
+ def reader():
+ for _ in range(1000):
+ enter.wait()
+ box[0].__getstate__()
+ leave.wait()
+
+ def writer():
+ for i in range(1000):
+ box[0] = type(f"C{i}", (), {})()
+ enter.wait()
+ setattr(box[0], str(i), 1)
+ leave.wait()
+
+ threading_helper.run_concurrently([reader, writer])
+
if __name__ == "__main__":
unittest.main()
PyDictValues *values = _PyObject_InlineValues(obj);
if (FT_ATOMIC_LOAD_UINT8(values->valid)) {
PyDictKeysObject *keys = CACHED_KEYS(tp);
- for (Py_ssize_t i = 0; i < keys->dk_nentries; i++) {
+ for (Py_ssize_t i = 0; i < LOAD_KEYS_NENTRIES(keys); i++) {
if (FT_ATOMIC_LOAD_PTR_RELAXED(values->values[i]) != NULL) {
return 0;
}