tests/thread: Add tests that mutate shared objects.

Tests concurrent mutating access to: list, dict, set, bytearray.

tests/thread/mutate_bytearray.py

+# test concurrent mutating access to a shared bytearray object
+#
+# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
+
+import _thread
+
+# the shared bytearray
+ba = bytearray()
+
+# main thread function
+def th(n, lo, hi):
+    for repeat in range(n):
+        for i in range(lo, hi):
+            l = len(ba)
+            ba.append(i)
+            assert len(ba) >= l + 1
+
+            l = len(ba)
+            ba.extend(bytearray([i]))
+            assert len(ba) >= l + 1
+
+    with lock:
+        global n_finished
+        n_finished += 1
+
+lock = _thread.allocate_lock()
+n_thread = 4
+n_finished = 0
+
+# spawn threads
+for i in range(n_thread):
+    _thread.start_new_thread(th, (40, i * 256 // n_thread, (i + 1) * 256 // n_thread))
+
+# busy wait for threads to finish
+while n_finished < n_thread:
+    pass
+
+# check bytearray has correct contents
+print(len(ba))
+count = [0 for _ in range(256)]
+for b in ba:
+    count[b] += 1
+print(count)
+

tests/thread/mutate_dict.py

+# test concurrent mutating access to a shared dict object
+#
+# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
+
+import _thread
+
+# the shared dict
+di = {'a':'A', 'b':'B', 'c':'C', 'd':'D'}
+
+# main thread function
+def th(n, lo, hi):
+    for repeat in range(n):
+        for i in range(lo, hi):
+            di[i] = repeat + i
+            assert di[i] == repeat + i
+
+            del di[i]
+            assert i not in di
+
+            di[i] = repeat + i
+            assert di[i] == repeat + i
+
+            assert di.pop(i) == repeat + i
+
+    with lock:
+        global n_finished
+        n_finished += 1
+
+lock = _thread.allocate_lock()
+n_thread = 4
+n_finished = 0
+
+# spawn threads
+for i in range(n_thread):
+    _thread.start_new_thread(th, (30, i * 300, (i + 1) * 300))
+
+# busy wait for threads to finish
+while n_finished < n_thread:
+    pass
+
+# check dict has correct contents
+print(sorted(di.items()))

tests/thread/mutate_list.py

+# test concurrent mutating access to a shared list object
+#
+# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
+
+import _thread
+
+# the shared list
+li = list()
+
+# main thread function
+def th(n, lo, hi):
+    for repeat in range(n):
+        for i in range(lo, hi):
+            li.append(i)
+            assert li.count(i) == repeat + 1
+
+            li.extend([i, i])
+            assert li.count(i) == repeat + 3
+
+            li.remove(i)
+            assert li.count(i) == repeat + 2
+
+            li.remove(i)
+            assert li.count(i) == repeat + 1
+
+    with lock:
+        global n_finished
+        n_finished += 1
+
+lock = _thread.allocate_lock()
+n_thread = 4
+n_finished = 0
+
+# spawn threads
+for i in range(n_thread):
+    _thread.start_new_thread(th, (4, i * 60, (i + 1) * 60))
+
+# busy wait for threads to finish
+while n_finished < n_thread:
+    pass
+
+# check list has correct contents
+li.sort()
+print(li)

tests/thread/mutate_set.py

+# test concurrent mutating access to a shared set object
+#
+# MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
+
+import _thread
+
+# the shared set
+se = set([-1, -2, -3, -4])
+
+# main thread function
+def th(n, lo, hi):
+    for repeat in range(n):
+        for i in range(lo, hi):
+            se.add(i)
+            assert i in se
+
+            se.remove(i)
+            assert i not in se
+
+    with lock:
+        global n_finished
+        n_finished += 1
+
+lock = _thread.allocate_lock()
+n_thread = 4
+n_finished = 0
+
+# spawn threads
+for i in range(n_thread):
+    _thread.start_new_thread(th, (50, i * 500, (i + 1) * 500))
+
+# busy wait for threads to finish
+while n_finished < n_thread:
+    pass
+
+# check set has correct contents
+print(sorted(se))