testpython.git

1) #!/usr/bin/env python3
2) # From https://pybay.com/site_media/slides/raymond2017-keynote/threading.html
3) import threading, queue

4) 
5) 

6) _counter:int = 0

7) 

8) counter_queue = queue.Queue()

9) 

10) def counter_manager() -> None:

11)     'I have EXCLUSIVE rights to update the _counter variable'
12)     global _counter
13) 
14)     while True:
15)         increment = counter_queue.get()
16)         _counter += increment
17)         print_queue.put([
18)             'The count is %d' % _counter,
19)             '---------------'])
20)         counter_queue.task_done()
21) 
22) 

23) print_queue = queue.Queue()

24) 

25) def print_manager() -> None:

26)     'I have EXCLUSIVE rights to call the "print" keyword'
27)     while True:
28)         job = print_queue.get()
29)         for line in job:
30)             print(line)
31)         print_queue.task_done()
32) 
33) 

34) def worker() -> None:

35)     'My job is to increment the counter and print the current count'
36)     counter_queue.put(1)
37) 

38) 

39) def run() -> None:

40)     # Set up the counter manager
41)     t = threading.Thread(target=counter_manager)
42)     t.daemon = True
43)     t.start()
44)     del t
45) 
46)     # Set up the print manager
47)     t = threading.Thread(target=print_manager)
48)     t.daemon = True
49)     t.start()
50)     del t
51) 
52)     # Start doing work
53)     print_queue.put(['Starting up'])
54)     worker_threads = []

55)     for _ in range(10):

56)         t = threading.Thread(target=worker)
57)         worker_threads.append(t)
58)         t.start()
59)     for t in worker_threads:
60)         t.join()
61) 
62)     counter_queue.join()
63)     print_queue.put(['Finishing up'])
64)     print_queue.join()
65) 

66)