1# test concurrent mutating access to a shared dict object
2#
3# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
4
5import _thread
6
7# the shared dict
8di = {"a": "A", "b": "B", "c": "C", "d": "D"}
9
10# main thread function
11def th(n, lo, hi):
12    for repeat in range(n):
13        for i in range(lo, hi):
14            di[i] = repeat + i
15            assert di[i] == repeat + i
16
17            del di[i]
18            assert i not in di
19
20            di[i] = repeat + i
21            assert di[i] == repeat + i
22
23            assert di.pop(i) == repeat + i
24
25    with lock:
26        global n_finished
27        n_finished += 1
28
29
30lock = _thread.allocate_lock()
31n_thread = 4
32n_finished = 0
33
34# spawn threads
35for i in range(n_thread):
36    _thread.start_new_thread(th, (30, i * 300, (i + 1) * 300))
37
38# busy wait for threads to finish
39while n_finished < n_thread:
40    pass
41
42# check dict has correct contents
43print(sorted(di.items()))
44