mutate_dict.py 919 B

123456789101112131415161718192021222324252627282930313233343536373839404142
  1. # test concurrent mutating access to a shared dict object
  2. #
  3. # MIT license; Copyright (c) 2016 Damien P. George on behalf of Pycom Ltd
  4. import _thread
  5. # the shared dict
  6. di = {'a':'A', 'b':'B', 'c':'C', 'd':'D'}
  7. # main thread function
  8. def th(n, lo, hi):
  9. for repeat in range(n):
  10. for i in range(lo, hi):
  11. di[i] = repeat + i
  12. assert di[i] == repeat + i
  13. del di[i]
  14. assert i not in di
  15. di[i] = repeat + i
  16. assert di[i] == repeat + i
  17. assert di.pop(i) == repeat + i
  18. with lock:
  19. global n_finished
  20. n_finished += 1
  21. lock = _thread.allocate_lock()
  22. n_thread = 4
  23. n_finished = 0
  24. # spawn threads
  25. for i in range(n_thread):
  26. _thread.start_new_thread(th, (30, i * 300, (i + 1) * 300))
  27. # busy wait for threads to finish
  28. while n_finished < n_thread:
  29. pass
  30. # check dict has correct contents
  31. print(sorted(di.items()))