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