标签:
之前两篇文章讨论了进程意外退出时,如何杀死子进程,这节我们研究下在使用进程池multiprocessing.Pool时,如何保证主进程意外退出,进程池中的worker进程同时退出,不产生孤儿进程。如果对python标准库进程池不清楚的园友,可以看下之前写的几篇文章。我们尝试下主进程中使用进程池,看看worker进程是否会退出:
1 import time 2 import os 3 import signal 4 from multiprocessing import Pool 5 6 7 def fun(x): 8 print ‘current sub-process pid is %s‘ % os.getpid() 9 while True: 10 print ‘args is %s ‘ % x 11 time.sleep(1) 12 13 def term(sig_num, addtion): 14 print ‘current pid is %s, group id is %s‘ % (os.getpid(), os.getpgrp()) 15 os.killpg(os.getpgid(os.getpid()), signal.SIGKILL) 16 17 if __name__ == ‘__main__‘: 18 print ‘current pid is %s‘ % os.getpid() 19 mul_pool = Pool() 20 signal.signal(signal.SIGTERM, term) 21 22 for i in range(3): 23 mul_pool.apply_async(func=fun, args=(str(i),))
运行上面的代码,发现在我还没来得及通过kill命令发送SIGTERM时,进程竟然退出了,而且主进程和进程池中的worker进程都退出了。结合线程的特征想了下,可能在新建worker进程时,默认启动方式为daemon。通过查看源码,发现worker进程启动之前,被设置为daemon=True,也就是说主进程不会等待worker进程执行完再退出,这种情况下worker进程作为主进程的子进程,会随着主进程的退出而退出,部分源码如下:
1 w = self.Process(target=worker, 2 args=(self._inqueue, self._outqueue, 3 self._initializer, 4 self._initargs, self._maxtasksperchild) 5 ) 6 self._pool.append(w) 7 w.name = w.name.replace(‘Process‘, ‘PoolWorker‘) 8 w.daemon = True 9 w.start()
接着我手动改了下源码,将daemon设置为False,接着启动进程,发现现象依然如前,程序刚启动紧接着就全部退出(主进程和子进程)。很奇怪,难道daemon表示的含义在进程和线程中有不同?联系之前对进程池分析的两篇文章,发现进程池中的几个线程在启动之前也被设置为daemon=True,继续手动修改下源码,将线程的daemon设置为False,再次启动进程,这次进程持续运行,主进程并未退出,通过kill命令发送SIGTERM信号后,整个进程组退出。编码中,我们当然不能去修改源码了,标准库中的Pool提供了一个join方法,它可以对进程池中的线程以及worker进程进行等待,注意在调用join之前调用close方法保证进程池不在接收新任务。我们在对上面的代码进行一些修改:
1 if __name__ == ‘__main__‘: 2 print ‘current pid is %s‘ % os.getpid() 3 mul_pool = Pool() 4 signal.signal(signal.SIGTERM, term) 5 6 for i in range(3): 7 mul_pool.apply_async(func=fun, args=(str(i),)) 8 9 mul_pool.close() 10 mul_pool.join()
改过之后程序不会自动退出了,但是又出现了新的问题,向进程发送kill命令,进程并没有捕获到信号,仍然继续运行。在stackoverflow找到了类似的问题,对标准库中signal有如下描述:A Python signal handler does not get executed inside the low-level (C) signal handler. Instead, the low-level signal handler sets a flag which tells the virtual machine to execute the corresponding Python signal handler at a later point(for example at the next bytecode instruction). This has consequences:
SIGFPE
or SIGSEGV
that are caused by an invalid operation in C code. Python will return from the signal handler to the C code, which is likely to raise the same signal again, causing Python to apparently hang. From Python 3.3 onwards, you can use the faulthandler
module to report on synchronous errors.标准库对signal handler的解释大致是说,python中的信号处理函数不会被低级别的信号处理器触发调用。取而代之的是,低级别的信号处理程序会设置一个标志,用来告诉虚拟机在稍后(例如下一个字节代码指令)来执行信号处理函数。这样的结果是:
调用mul_pool.join使得主进程(线程)阻塞在join处,意味着它阻塞在C方法pthread_join调用中。pthread_join并不是一个long-running calculation的程序,而是一个系统调用的阻塞,尽管如此,直到它结束,否则信号处理函数无法被执行。帖子中给出的解决方法时更新python版本至3.3,而我使用的版本是python2.7。这里我并未尝试使用python3.3版本,而是将join用loop sleep代替,简单修改下上面的代码:
1 if __name__ == ‘__main__‘: 2 print ‘current pid is %s‘ % os.getpid() 3 mul_pool = Pool() 4 signal.signal(signal.SIGTERM, term) 5 6 for i in range(3): 7 mul_pool.apply_async(func=fun, args=(str(i),)) 8 9 while True: 10 time.sleep(60)
这样整个进程组仍然能够在收到SIGTERM命令之后退出,而不留下孤儿进程。但是仔细想想我们这样做是不是有些武断,如果一些worker进程在运行一些重要的业务逻辑,强制结束可能会使得数据的丢失,或者一些其他难以恢复的后果,那么有没有更合理的处理方式,使worker进程在处理完本轮数据后,再退出呢?答案同样是肯定的,python标准库中提供了一些进程间同步的工具,这里我们使用Event对象来做同步。首先我们需要通过multiprocessing.Manager类来获取一个Event对象,用Event来控制worker进程的退出,首先修改worker进程的回调函数:
1 def fun(x, event): 2 while not event.is_set(): 3 print ‘process %s running args is %s‘ % (os.getpid(), x) 4 time.sleep(3) 5 print ‘process %s, call fun finish‘ % os.getpid()
event对象是用来控制worker进程的,当然代码中的使用只是一个简单的示例,现实情况中worker进程并非一个while这么简单。我们要通过event来控制worker进程的退出,那么可以看到,当event.is_set() == True时,worker会自动退出,那么可以捕获SIGTERM信号,在signal_handler中将event对象进行set:
1 def terminate(pool, event, sig_num, addtion): 2 print ‘terminate process %d‘ % os.getpid() 3 if not event.is_set(): 4 event.set() 5 6 pool.close() 7 pool.join() 8 9 print ‘exit...‘
在主进程中,首先要创建一个Manager对象,有它来产生Event对象,注意在创建Manager对象后,通过后台ps命令可以看到,此时会多了一个进程,实际上创建Manager对象就会创建一个新的进程,用于数据的同步,我们在signal信号处理函数中实现设置event,并且终止进程池,而signal.signal回调函数只能有两个参数,所以依旧使用partial偏函数进行处理:
1 if __name__ == ‘__main__‘: 2 print ‘current pid is %s‘ % os.getpid() 3 mul_pool = Pool() 4 manager = Manager() 5 event = manager.Event() 6 7 handler = functools.partial(terminate, mul_pool, event) 8 signal.signal(signal.SIGTERM, handler) 9 10 for i in range(4): 11 mul_pool.apply_async(func=fun, args=(str(i), event)) 12 13 while True: 14 time.sleep(60)
运行程序,通过kill命令发送SIGTERM信号,观察到的现象是收到signal信号之后,执行了event.set()方法,worker进程退出,进程池关闭,但是ps之后,发现还有两个进程在运行,通过进程id和strace命令发现一个是主进程,一个是Manager进程同步对象。代码中,主进程最后进入了loop sleep状态,所以当我们收到信号之后,虽然通过event将worker进程和进程池结束,但是主进程的仍然在sleep,所以Manager进程同步对象也为退出。这样我们可以简单修改下代码来处理,可以在terminate方法中添加manager参数,在方法中显示调用manager.shutdown()关闭进程同步对象,然后强制退出,也可以在主进程中同样使用event来代替whlie True循环。这里我们采用第一种方式,简单修改下上面的代码:
1 def terminate(pool, event, manager, sig_num, addtion): 2 print ‘terminate process %d‘ % os.getpid() 3 if not event.is_set(): 4 event.set() 5 6 pool.close() 7 pool.join() 8 manager.shutdown() 9 print ‘exit ...‘ 10 os._exit(0) 11 12 if __name__ == ‘__main__‘: 13 print ‘current pid is %s‘ % os.getpid() 14 mul_pool = Pool() 15 manager = Manager() 16 event = manager.Event() 17 18 handler = functools.partial(terminate, mul_pool, event, manager) 19 signal.signal(signal.SIGTERM, handler) 20 21 for i in range(4): 22 mul_pool.apply_async(func=fun, args=(str(i), event)) 23 24 while True: 25 time.sleep(60)
主进程被杀死时,如何保证子进程同时退出,而不变为孤儿进程(三)
标签:
原文地址:http://www.cnblogs.com/Tour/p/5191946.html