-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy paththreading_example.py
50 lines (36 loc) · 1.35 KB
/
threading_example.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import threading
from typing import Callable
def printNumber(i):
print(i, end="")
class ZeroEvenOdd:
def __init__(self, n):
self.n = n
self.lockZero = threading.Lock()
self.lockEven = threading.Lock()
self.lockOdd = threading.Lock()
# printNumber(x) outputs "x", where x is an integer.
def zero(self, printNumber: 'Callable[[int], None]') -> None:
for i in range(self.n):
with self.lockEven and self.lockOdd:
printNumber(0)
def even(self, printNumber: 'Callable[[int], None]') -> None:
for i in range(self.n):
if i % 2 == 0:
with self.lockZero and self.lockOdd:
printNumber(i)
def odd(self, printNumber: 'Callable[[int], None]') -> None:
for i in range(self.n):
if i % 2 == 1:
with self.lockZero and self.lockEven:
printNumber(i)
if __name__ == '__main__':
ss = ZeroEvenOdd(3)
thread1 = threading.Thread(target=ss.zero, args=(printNumber,))
thread2 = threading.Thread(target=ss.even, args=(printNumber,))
thread3 = threading.Thread(target=ss.odd, args=(printNumber,))
thread1.start()
thread2.start()
thread3.start()
thread1.join()
thread2.join()
thread3.join()