1# test that we can run the garbage collector within threads
2#
3# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
4
5import gc
6import _thread
7
8
9def thread_entry(n):
10    # allocate a bytearray and fill it
11    data = bytearray(i for i in range(256))
12
13    # do some work and call gc.collect() a few times
14    for i in range(n):
15        for i in range(len(data)):
16            data[i] = data[i]
17        gc.collect()
18
19    # print whether the data remains intact and indicate we are finished
20    with lock:
21        print(list(data) == list(range(256)))
22        global n_finished
23        n_finished += 1
24
25
26lock = _thread.allocate_lock()
27n_thread = 4
28n_finished = 0
29
30# spawn threads
31for i in range(n_thread):
32    _thread.start_new_thread(thread_entry, (10,))
33
34# busy wait for threads to finish
35while n_finished < n_thread:
36    pass
37