1# test concurrent mutating access to a shared user instance
2#
3# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
4
5import _thread
6
7# the shared user class and instance
8class User:
9    def __init__(self):
10        self.a = "A"
11        self.b = "B"
12        self.c = "C"
13
14
15user = User()
16
17# main thread function
18def th(n, lo, hi):
19    for repeat in range(n):
20        for i in range(lo, hi):
21            setattr(user, "attr_%u" % i, repeat + i)
22            assert getattr(user, "attr_%u" % i) == repeat + i
23
24    with lock:
25        global n_finished
26        n_finished += 1
27
28
29lock = _thread.allocate_lock()
30n_repeat = 30
31n_range = 50  # 300 for stressful test (uses more heap)
32n_thread = 4
33n_finished = 0
34
35# spawn threads
36for i in range(n_thread):
37    _thread.start_new_thread(th, (n_repeat, i * n_range, (i + 1) * n_range))
38
39# busy wait for threads to finish
40while n_finished < n_thread:
41    pass
42
43# check user instance has correct contents
44print(user.a, user.b, user.c)
45for i in range(n_thread * n_range):
46    assert getattr(user, "attr_%u" % i) == n_repeat - 1 + i
47