-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathtest.py
59 lines (48 loc) · 1.06 KB
/
test.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
51
52
53
54
55
56
57
58
59
import time
import threading
#_lock = threading.Lock()
def _io_bound():
while True:
time.sleep(1)
print("io: ", threading.current_thread().ident)
# _lock.acquire()
# for i in range(10000000):
# pass
# _lock.release()
def _cpu_bound():
k = 0
while k < 50:
for i in range(1000000):
pass
#print(threading.current_thread().ident)
#k += 1
#print(k)
# if k == 5:
# break
#k += 1
#print(k)
import os
import sys
import time
print(sys.executable, " ", os.getpid())
t0 = time.time()
# import gil_load
# gil_load.init()
# gil_load.start(output=sys.stdout)
for i in range(1):
t = threading.Thread(target=_io_bound)
t.daemon = True
t.start()
wait_threads = []
for i in range(3):
t = threading.Thread(target=_cpu_bound)
t.daemon = True
t.start()
wait_threads.append(t)
print("all threads created!")
# try:
for t in wait_threads:
t.join()
# except:
# sys.exit(0)
print("Elapsed=%0.6f" % (time.time() - t0))