-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathserial_thds.py
50 lines (40 loc) · 1.14 KB
/
serial_thds.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 threading import Thread
import time
class ThreadManager:
def __init__(self):
self.thread_queue = []
thm = Thread(target=self.thread_serializer)
thm.daemon = True
thm.start()
def thread_serializer(self):
while True:
time.sleep(0.1)
if self.thread_queue:
th = self.thread_queue[0]
th.start()
th.join()
self.thread_queue.pop(0)
else:
pass
class Tests:
def test1(self):
print 'Test1'
for i in range(10):
time.sleep(1)
def test2(self):
print 'Test2'
for i in range(10):
time.sleep(1)
def test3(self):
print 'Test3'
for i in range(10):
time.sleep(1)
Tests = Tests()
ThreadManager = ThreadManager()
testThread1 = Thread(target=Tests.test1)
ThreadManager.thread_queue.append(testThread1)
testThread2 = Thread(target=Tests.test2)
ThreadManager.thread_queue.append(testThread2)
testThread3 = Thread(target=Tests.test3)
ThreadManager.thread_queue.append(testThread3)