如何终止多处理父进程?
How to Terminate Multiprocessing Parent Process?
如何让我的子进程终止父进程?
我在想……
子进程获取父进程的 PID 并使用 psutil
终止父进程
这是我的代码
from multiprocessing import Process, Value
import keyboard
import sys
def KeyboardScan(v):
Paused = False
while True:
if keyboard.is_pressed('x'):
v.value = 1
print("Child Terminate")
sys.exit()
#Child process's code
if __name__ == "__main__":
val = Value('i', 0)
p = Process(target=KeyboardScan, args=(val,))
p.start()
#Parent process's code
因此,如果按下 x
键,子对象应该终止父对象及其自身。
在此代码中,value
存在的唯一原因是最初我在父进程中有一个 while 循环来检查该值。如果值为 1,则它在父进程上执行 sys.exit()
。
但我尽量避免使用这种方法。
有什么建议吗?
您可以使用 os
模块来 get the ppid
, and then .kill
it。
试试这个:
from multiprocessing import Process, Value
import keyboard
import os
import signal
import sys
import psutil
def KeyboardScan(v):
Paused = False
while True:
if keyboard.is_pressed('x'):
v.value = 1
print("Child Terminate")
#os.kill(os.getppid(), signal.SIGTERM) # <=== if you don't have psutil
# below 2 line depend on psutil
p = psutil.Process(os.getppid())
p.terminate() #or p.kill()
sys.exit()
#Child process's code
if __name__ == "__main__":
val = Value('i', 0)
p = Process(target=KeyboardScan, args=(val,))
p.start()
#Parent process's code
考虑使用多处理原语来确保两个进程的干净关闭。实现此目的的两种方法:
- 对 child 进行编码,使其在检测到 X 时简单地退出。让 parent 偶尔通过
p.is_alive
或 [= 检查 child 进程是否仍然存在12=]。如果不是,则退出
- 将键盘检测移回 parent 进程。表示 child 应该通过共享
multiprocessing.Event
退出
你还没有说你的 parent 进程在做什么,而 child 是 运行。这会影响这两种方法中哪一种更容易使用。我喜欢第二种方法,因为它对我来说很合适。我宁愿让 parent 处理用户输入,只让 child 进程完成不涉及与用户交互的工作。
下面的代码片段显示了第二种方法。
from multiprocessing import Process, Event
import atexit
import time
from random import randint
def child_main(e):
print('Child process booted')
while True:
if e.is_set():
print('Child process got terminate event. Exiting')
return
print('Child process doing some important work')
time.sleep(1.0)
def parent_main():
print("parent process booting")
quit_event = Event()
child = Process(target=child_main, args=(quit_event,))
print('starting child')
child.start()
is_x_pressed = lambda: randint(0,8) == 0
while True:
print('Parent doing its thing')
time.sleep(.5)
if is_x_pressed():
print("Parent detected X. Exiting")
quit_event.set()
return
if __name__== '__main__':
def exit_hook():
print('Parent process atexit handler')
atexit.register(exit_hook)
parent_main()
print('parent exiting __main__')
如何让我的子进程终止父进程?
我在想……
子进程获取父进程的 PID 并使用 psutil
这是我的代码
from multiprocessing import Process, Value
import keyboard
import sys
def KeyboardScan(v):
Paused = False
while True:
if keyboard.is_pressed('x'):
v.value = 1
print("Child Terminate")
sys.exit()
#Child process's code
if __name__ == "__main__":
val = Value('i', 0)
p = Process(target=KeyboardScan, args=(val,))
p.start()
#Parent process's code
因此,如果按下 x
键,子对象应该终止父对象及其自身。
在此代码中,value
存在的唯一原因是最初我在父进程中有一个 while 循环来检查该值。如果值为 1,则它在父进程上执行 sys.exit()
。
但我尽量避免使用这种方法。
有什么建议吗?
您可以使用 os
模块来 get the ppid
, and then .kill
it。
试试这个:
from multiprocessing import Process, Value
import keyboard
import os
import signal
import sys
import psutil
def KeyboardScan(v):
Paused = False
while True:
if keyboard.is_pressed('x'):
v.value = 1
print("Child Terminate")
#os.kill(os.getppid(), signal.SIGTERM) # <=== if you don't have psutil
# below 2 line depend on psutil
p = psutil.Process(os.getppid())
p.terminate() #or p.kill()
sys.exit()
#Child process's code
if __name__ == "__main__":
val = Value('i', 0)
p = Process(target=KeyboardScan, args=(val,))
p.start()
#Parent process's code
考虑使用多处理原语来确保两个进程的干净关闭。实现此目的的两种方法:
- 对 child 进行编码,使其在检测到 X 时简单地退出。让 parent 偶尔通过
p.is_alive
或 [= 检查 child 进程是否仍然存在12=]。如果不是,则退出 - 将键盘检测移回 parent 进程。表示 child 应该通过共享
multiprocessing.Event
退出
你还没有说你的 parent 进程在做什么,而 child 是 运行。这会影响这两种方法中哪一种更容易使用。我喜欢第二种方法,因为它对我来说很合适。我宁愿让 parent 处理用户输入,只让 child 进程完成不涉及与用户交互的工作。
下面的代码片段显示了第二种方法。
from multiprocessing import Process, Event
import atexit
import time
from random import randint
def child_main(e):
print('Child process booted')
while True:
if e.is_set():
print('Child process got terminate event. Exiting')
return
print('Child process doing some important work')
time.sleep(1.0)
def parent_main():
print("parent process booting")
quit_event = Event()
child = Process(target=child_main, args=(quit_event,))
print('starting child')
child.start()
is_x_pressed = lambda: randint(0,8) == 0
while True:
print('Parent doing its thing')
time.sleep(.5)
if is_x_pressed():
print("Parent detected X. Exiting")
quit_event.set()
return
if __name__== '__main__':
def exit_hook():
print('Parent process atexit handler')
atexit.register(exit_hook)
parent_main()
print('parent exiting __main__')