pytest/fork_test.py at master · RiderLai/pytest · GitHub

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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
# import os
#
#
# def child_process():
#     print("We are in Child_Process")
#     print("My PID: %d" % os.getpid())
#     print("Child_Process is exiting")
#
#
# def parent_process():
#     print("-------Parent_process---------")
#     wpid = os.fork()
#     if wpid == 0:
#         print("wpid is 0 means We are in Child_process")
#         print("Child :%d" % wpid)
#         child_process()
#     else:
#         print("Execute Parent_process")
#         print("Parent_process %d" % wpid)
#         parent_process()
#
#
# parent_process()


# import os
# pid = os.fork()
# print(pid)


# import os
# import time
#
# NUM_PROCESSES = 7
#
#
# def timeConsumingFunction():
#     x = 1
#     for n in range(10000000):
#         x += 1
#
#
# children = []
# start_time = time.time()
# for process in range(NUM_PROCESSES):
#     pid = os.fork()
#     if pid:
#         children.append(pid)
#         print(pid)
#     else:
#         timeConsumingFunction()
#         os._exit(0)
# for i, child in enumerate(children):
#     os.waitpid(child, 0)
# print(time.time() - start_time)


# import os
# import asyncio
# from tornado.process import fork_processes
#
# if __name__ == '__main__':
#     print(id(asyncio.get_event_loop()))
#     fork_processes(2)
#     loop1 = asyncio.new_event_loop()
#     print(id(loop1))
#     loop2 = asyncio.new_event_loop()
#     print(id(loop1))
#     print(id(loop2))

import asyncio
from multiprocessing import Process


def main():
    loop = asyncio.get_event_loop()
    print(id(loop))


if __name__ == '__main__':
    for i in range(1):
        t = Process(target=main)
        t.start()
    main()