Coverage for tests/test_slab.py: 91%

1import sys 

2import threading 

3import unittest 

4 

5from wasmtime._slab import Slab 

6 

7 

8class TestSlab(unittest.TestCase): 

9 def test_concurrent_allocate_and_deallocate(self): 

10 """Handles stay unique and keep their values across threads. 

11 

12 The garbage collector runs wasmtime finalizers, and so `deallocate`, on 

13 whichever thread triggers a collection, concurrently with `allocate` on 

14 the thread that is using wasmtime. 

15 """ 

16 slab: Slab[object] = Slab() 

17 threads = 8 

18 iterations = 20000 

19 barrier = threading.Barrier(threads) 

20 errors = [] 

21 

22 def worker(n): 

23 try: 

24 barrier.wait() 

25 live = {} 

26 for i in range(iterations): 

27 val = (n, i) 

28 live[slab.allocate(val)] = val 

29 if len(live) > 4: 

30 idx = next(iter(live)) 

31 if slab.get(idx) is not live.pop(idx): 

32 raise AssertionError( 

33 'handle {} was handed to two owners'.format(idx)) 

34 slab.deallocate(idx) 

35 except BaseException as e: 

36 errors.append(e) 

37 

38 interval = sys.getswitchinterval() 

39 # Switch threads as often as possible to provoke interleavings. 

40 sys.setswitchinterval(1e-6) 

41 try: 

42 workers = [threading.Thread(target=worker, args=(n,)) 

43 for n in range(threads)] 

44 for t in workers: 

45 t.start() 

46 for t in workers: 

47 t.join() 

48 finally: 

49 sys.setswitchinterval(interval) 

50 

51 self.assertEqual(errors, [])