multiprocessing hang问题

linewer

1. #!/usr/bin/python
   
2. 
   
3. import os, sys, time, atexit, errno
   
4. import multiprocessing as mp
   
5. from signal import signal
   
6. from signal import SIGTERM
   
7. from signal import SIGINT
   
8. 
   
9. 
   
10. 
    
11. exit_flag = mp.Event()
    
12. stdin =  "/dev/null"
    
13. stdout = "/dev/null"
    
14. #stderr = "/dev/null"
    
15. stderr = "/tmp/test.log"
    
16. pidfile ="/tmp/test.pid"
    
17. 
    
18. def daemonize():
    
19.     global stdin, stdout, stderr, pidfile
    
20. 
    
21.     try:
    
22.         pid = os.fork()
    
23.         if pid > 0:
    
24.             sys.exit(0)
    
25.     except OSError, e:
    
26.         sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
    
27.         sys.exit(1)
    
28. 
    
29.     os.chdir("/")
    
30.     os.setsid()
    
31.     os.umask(0)
    
32. 
    
33.     try:
    
34.         pid = os.fork()
    
35.         if pid > 0:
    
36.             sys.exit(0)
    
37.     except OSError, e:
    
38.         sys.stderr.write("fork #2 failed: %d (%s)\n" % (e.errno, e.strerror))
    
39.         sys.exit(1)
    
40. 
    
41.     sys.stdout.flush()
    
42.     sys.stderr.flush()
    
43.     si = file(stdin, 'r')
    
44.     so = file(stdout, 'a+')
    
45.     se = file(stderr, 'a+', 0)
    
46.     os.dup2(si.fileno(), sys.stdin.fileno())
    
47.     os.dup2(so.fileno(), sys.stdout.fileno())
    
48.     os.dup2(se.fileno(), sys.stderr.fileno())
    
49. 
    
50.     atexit.register(delpid)
    
51.     pid = str(os.getpid())
    
52.     file(pidfile, 'w+').write("%s\n" % pid)
    
53. 
    
54. def delpid():
    
55.     os.remove(pidfile)
    
56. 
    
57. def handle_signal():
    
58.     signal(SIGINT, handler)
    
59.     signal(SIGTERM, handler)
    
60. 
    
61. def handler(signum, frame):
    
62.     global exit_flag
    
63.     sys.stderr.write("get a signal: %s" % str(signum))
    
64.     exit_flag.set()
    
65. 
    
66. def start(run):
    
67.     global pidfile
    
68. 
    
69.     try:
    
70.         pf = file(pidfile, 'r')
    
71.         pid = int(pf.read().strip())
    
72.         pf.close()
    
73.     except IOError:
    
74.         pid = None
    
75. 
    
76.     if pid:
    
77.         message = "pidfile %s already exists. Daemon already running!\n"
    
78.         sys.stderr.write(message % pidfile)
    
79.         sys.exit(1)
    
80. 
    
81.     daemonize()
    
82.     handle_signal()
    
83.     run()
    
84. 
    
85. def stop():
    
86.     global pidfile
    
87.     try:
    
88.         pf = file(pidfile, 'r')
    
89.         pid = int(pf.read().strip())
    
90.         pf.close()
    
91.     except IOError:
    
92.         pid = None
    
93. 
    
94.     if not pid:
    
95.         message = "pidfile %s does not exist. Daemon not running!\n"
    
96.         sys.stderr.write(message % pidfile)
    
97.         return
    
98.     try:
    
99.         while True:
    
100.             os.kill(pid, SIGTERM)
     
101.             time.sleep(1)
     
102.     except OSError, err:
     
103.         if err.errno == errno.ESRCH:
     
104.             if os.path.exists(pidfile):
     
105.                 os.remove(pidfile)
     
106.         else:
     
107.             sys.exit(1)
     
108. 
     
109. def restart(run):
     
110.     stop()
     
111.     start(run)
     
112. 
     
113. 
     
114. def worker():
     
115.     name = mp.current_process().name
     
116.     message = "my name:%s,mypid is:%s\n"
     
117.     sys.stderr.write(message % (name,os.getpid()))
     
118.     while True:
     
119.        if exit_flag.is_set():
     
120.           break
     
121. 
     
122. def test_start():
     
123.     proc_pool = {}
     
124.     proc_info = {}
     
125. 
     
126.     for task in range(3):
     
127.         print task
     
128.         proc = mp.Process(target=worker)
     
129.         proc.start()
     
130.         proc_pool[proc.pid] = proc
     
131.         ##debug info
     
132.         sys.stderr.write('===pid %s: task %s===\n' % (proc.pid, proc.name))
     
133.         proc_info[proc.pid] = task
     
134.         #proc.join(timeout=1)
     
135. 
     
136. 
     
137.     while True:
     
138.         is_alive = False
     
139.         for pid, proc in proc_pool.items():
     
140.             #???
     
141.             proc.join(timeout=1)
     
142.             if proc.is_alive():
     
143.                 is_alive = True
     
144.             else:
     
145.                 proc_pool.pop(pid)
     
146.                 sys.stderr.write('===pid %s exit===\n' % (pid))
     
147.                 if not exit_flag.is_set():
     
148.                    #start another process for this task!(1)abort (2)???normal quit???
     
149.                    proc1 = mp.Process(target=worker)
     
150.                    proc1.start()
     
151.                    proc_pool[proc1.pid]=proc1
     
152.                    sys.stderr.write('===-pid %s: task %s-===\n' % (proc1.pid, proc_info[pid]))
     
153.                    proc_info[proc1.pid]=proc_info[pid]
     
154.                    proc_info.pop(pid)
     
155.                    #proc1.join(timeout=1)
     
156. 
     
157.                 #for k,v in proc_info.items():
     
158.                 #      sys.stderr.write('new===pid %s: task %s===\n' % (k, v))
     
159.                 #sys.stderr.write('~~~~~~~~~%s\n' % exit_flag.is_set())
     
160.                
     
161.         if not is_alive:
     
162.             break
     
163. 
     
164. def main():
     
165.     if len(sys.argv) == 2:
     
166.         if "start" == sys.argv[1]:
     
167.                 start(test_start)
     
168.         elif "stop" == sys.argv[1]:
     
169.             stop()
     
170.         elif "restart" == sys.argv[1]:
     
171.             restart(run)            
     
172.         else:
     
173.             print "Unknown Command"
     
174.             sys.exit(2)
     
175.         sys.exit(0)
     
176.     else:
     
177.         print "usage: %s start|stop|restart" % sys.argv[0]
     
178.         sys.exit(2)
     
179. 
     
180. 
     
181. 
     
182. if __name__ == "__main__":
     
183.     main()

复制代码

如上简化代码示例要实现的功能为: 某子进程异常退出时能自动重新生成一个，正常运行没有问题,  但是在模拟测试时发现,  在kill -9 某个子进程或者某两个时, 偶尔会出现父进程和其余子进程挂起的问题,不能新生成一个新进程(gdb看到是在sem_wait)，MS exit_flag死锁了,这个应该怎么修正，哪位指导下，多谢！
#0  0x00000038e280d720 in sem_wait () from /lib64/libpthread.so.0
#1  0x00007f24dc80aef1 in ?? () from /usr/lib64/python2.6/lib-dynload/_multiprocessing.so
#2  0x000000357dad59e4 in PyEval_EvalFrameEx () from /usr/lib64/libpython2.6.so.1.0
#3  0x000000357dad6b8f in PyEval_EvalFrameEx () from /usr/lib64/libpython2.6.so.1.0
#4  0x000000357dad7657 in PyEval_EvalCodeEx () from /usr/lib64/libpython2.6.so.1.0

tc1989tc

1. class Event(object):
   
2. 
   
3.     def __init__(self):
   
4.         self._cond = Condition(Lock())
   
5.         self._flag = Semaphore(0)
   
6. 
   
7.     def is_set(self):
   
8.         self._cond.acquire()
   
9.         try:
   
10.             if self._flag.acquire(False):
    
11.                 self._flag.release()
    
12.                 return True
    
13.             return False
    
14.         finally:
    
15.             self._cond.release()
    
16. 
    
17.     def set(self):
    
18.         self._cond.acquire()
    
19.         try:
    
20.             self._flag.acquire(False)
    
21.             self._flag.release()
    
22.             self._cond.notify_all()
    
23.         finally:
    
24.             self._cond.release()
    
25. 
    
26.     def clear(self):
    
27.         self._cond.acquire()
    
28.         try:
    
29.             self._flag.acquire(False)
    
30.         finally:
    
31.             self._cond.release()
    
32. 
    
33.     def wait(self, timeout=None):
    
34.         self._cond.acquire()
    
35.         try:
    
36.             if self._flag.acquire(False):
    
37.                 self._flag.release()
    
38.             else:
    
39.                 self._cond.wait(timeout)
    
40. 
    
41.             if self._flag.acquire(False):
    
42.                 self._flag.release()
    
43.                 return True
    
44.             return False
    
45.         finally:
    
46.             self._cond.release()

复制代码

简单说说，以上是mp.Event代码的set 和is_set方法，可以看出是都存在信号量的获取的
所以你在中断和非中断都在调用Event的set和is_set  肯定存在死锁的可能啊
尽量不要在中断处理中调用信号量等处理

linewer

多谢LS,我后来自己看了下,在父子进程中，有些东西是继承的(复制的),比如信号及信号处理函数,这样才导致死锁现象. 程序的本意就是父进程收到终止信号后,设置Event变量,子进程判断后退出, 因为处理不好(其实是在修別人的bug),导致子进程同样处理此信号, 改成启动子进程之前忽略信号,之后恢复，代码如下,如果还有问题，欢迎斧正.

1. #!/usr/bin/python
   
2. 
   
3. import os, sys, time, atexit, errno
   
4. import multiprocessing as mp
   
5. from signal import signal
   
6. from signal import getsignal
   
7. from signal import SIGTERM
   
8. from signal import SIGINT
   
9. from signal import SIG_IGN
   
10. 
    
11. 
    
12. 
    
13. exit_flag = mp.Event()
    
14. stdin =  "/dev/null"
    
15. stdout = "/dev/null"
    
16. #stderr = "/dev/null"
    
17. stderr = "/tmp/test.log"
    
18. pidfile ="/tmp/test.pid"
    
19. 
    
20. def daemonize():
    
21.     global stdin, stdout, stderr, pidfile
    
22. 
    
23.     try:
    
24.         pid = os.fork()
    
25.         if pid > 0:
    
26.             sys.exit(0)
    
27.     except OSError, e:
    
28.         sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
    
29.         sys.exit(1)
    
30. 
    
31.     os.chdir("/")
    
32.     os.setsid()
    
33.     os.umask(0)
    
34. 
    
35.     try:
    
36.         pid = os.fork()
    
37.         if pid > 0:
    
38.             sys.exit(0)
    
39.     except OSError, e:
    
40.         sys.stderr.write("fork #2 failed: %d (%s)\n" % (e.errno, e.strerror))
    
41.         sys.exit(1)
    
42. 
    
43.     sys.stdout.flush()
    
44.     sys.stderr.flush()
    
45.     si = file(stdin, 'r')
    
46.     so = file(stdout, 'a+')
    
47.     se = file(stderr, 'a+', 0)
    
48.     os.dup2(si.fileno(), sys.stdin.fileno())
    
49.     os.dup2(so.fileno(), sys.stdout.fileno())
    
50.     os.dup2(se.fileno(), sys.stderr.fileno())
    
51. 
    
52.     atexit.register(delpid)
    
53.     pid = str(os.getpid())
    
54.     file(pidfile, 'w+').write("%s\n" % pid)
    
55. 
    
56. def delpid():
    
57.     os.remove(pidfile)
    
58. 
    
59. def handle_signal():
    
60.     signal(SIGINT, handler)
    
61.     signal(SIGTERM, handler)
    
62. 
    
63. def handler(signum, frame):
    
64.     global exit_flag
    
65.     sys.stderr.write("get a signal: %s" % str(signum))
    
66.     exit_flag.set()
    
67. 
    
68. def start(run):
    
69.     global pidfile
    
70. 
    
71.     try:
    
72.         pf = file(pidfile, 'r')
    
73.         pid = int(pf.read().strip())
    
74.         pf.close()
    
75.     except IOError:
    
76.         pid = None
    
77. 
    
78.     if pid:
    
79.         message = "pidfile %s already exists. Daemon already running!\n"
    
80.         sys.stderr.write(message % pidfile)
    
81.         sys.exit(1)
    
82. 
    
83.     daemonize()
    
84.     handle_signal()
    
85.     run()
    
86. 
    
87. def stop():
    
88.     global pidfile
    
89.     try:
    
90.         pf = file(pidfile, 'r')
    
91.         pid = int(pf.read().strip())
    
92.         pf.close()
    
93.     except IOError:
    
94.         pid = None
    
95. 
    
96.     if not pid:
    
97.         message = "pidfile %s does not exist. Daemon not running!\n"
    
98.         sys.stderr.write(message % pidfile)
    
99.         return
    
100.     try:
     
101.         while True:
     
102.             os.kill(pid, SIGTERM)
     
103.             time.sleep(1)
     
104.     except OSError, err:
     
105.         if err.errno == errno.ESRCH:
     
106.             if os.path.exists(pidfile):
     
107.                 os.remove(pidfile)
     
108.         else:
     
109.             sys.exit(1)
     
110. 
     
111. def restart(run):
     
112.     stop()
     
113.     start(run)
     
114. 
     
115. 
     
116. def worker():
     
117.     name = mp.current_process().name
     
118.     message = "my name:%s,mypid is:%s\n"
     
119.     sys.stderr.write(message % (name,os.getpid()))
     
120.     while True:
     
121.        if exit_flag.is_set():
     
122.           break
     
123. 
     
124. def test_start():
     
125.     proc_pool = {}
     
126.     proc_info = {}
     
127.     default_handler = getsignal(SIGTERM)
     
128.     #Set signal handling of SIGTERM to ignore mode.
     
129.     signal(SIGTERM, SIG_IGN)
     
130. 
     
131.     for task in range(3):
     
132.         print task
     
133.         proc = mp.Process(target=worker)
     
134.         proc.start()
     
135.         proc_pool[proc.pid] = proc
     
136.         ##debug info
     
137.         sys.stderr.write('===pid %s: task %s===\n' % (proc.pid, proc.name))
     
138.         proc_info[proc.pid] = task
     
139.         #proc.join(timeout=1)
     
140. 
     
141.     signal(SIGTERM, default_handler)
     
142. 
     
143.     while True:
     
144.         is_alive = False
     
145.         for pid, proc in proc_pool.items():
     
146.             #???
     
147.             proc.join(timeout=1)
     
148.             if proc.is_alive():
     
149.                 is_alive = True
     
150.             else:
     
151.                 proc_pool.pop(pid)
     
152.                 sys.stderr.write('===pid %s exit===\n' % (pid))
     
153.                 if not exit_flag.is_set():
     
154.                    #start another process for this task!(1)abort (2)???normal quit???
     
155.                    signal(SIGTERM, SIG_IGN)
     
156.                    proc1 = mp.Process(target=worker)
     
157.                    proc1.start()
     
158.                    signal(SIGTERM, default_handler)
     
159.                    proc_pool[proc1.pid]=proc1
     
160.                    sys.stderr.write('===-pid %s: task %s-===\n' % (proc1.pid, proc_info[pid]))
     
161.                    proc_info[proc1.pid]=proc_info[pid]
     
162.                    proc_info.pop(pid)
     
163.                    #proc1.join(timeout=1)
     
164.                    is_alive = True
     
165. 
     
166.                 #for k,v in proc_info.items():
     
167.                 #      sys.stderr.write('new===pid %s: task %s===\n' % (k, v))
     
168.                 #sys.stderr.write('~~~~~~~~~%s\n' % exit_flag.is_set())
     
169.                
     
170.         if not is_alive:
     
171.             break
     
172. 
     
173. def main():
     
174.     if len(sys.argv) == 2:
     
175.         if "start" == sys.argv[1]:
     
176.                 start(test_start)
     
177.         elif "stop" == sys.argv[1]:
     
178.             stop()
     
179.         elif "restart" == sys.argv[1]:
     
180.             restart(run)            
     
181.         else:
     
182.             print "Unknown Command"
     
183.             sys.exit(2)
     
184.         sys.exit(0)
     
185.     else:
     
186.         print "usage: %s start|stop|restart" % sys.argv[0]
     
187.         sys.exit(2)
     
188. 
     
189. 
     
190. 
     
191. if __name__ == "__main__":
     
192.     main()

复制代码

tc1989tc

依然有问题啊  建议不要在中断去exit_flags.set

linewer

tc1989tc 发表于 2015-09-18 17:44
依然有问题啊  建议不要在中断去exit_flags.set

如果要在收到信号后，去设置exit_flags.set呢？设置个变量,放在父进程主循环里做个判断,然后设置exit_flags.set么? 或者还有别的好的方法么?

tc1989tc

回复 5# linewer


    把你的需求说出来吧  然后在看怎么解决

linewer

tc1989tc 发表于 2015-09-18 19:31
回复 5# linewer

就是父进程通过multiprocess启动几个进程(这些进程做一些独立的工作,比如请求web抓取页面),在父进程收到终止信号后，设置exit_flag.set, 子进程内的工作函数用到这个exit_flag做一些判断(平滑关闭链接),然后退出, 父进程会监控子进程的运行情况，退出就新启一个.   整个程序放到cron中，定期执行/终止

tc1989tc

sorry  现在你没在子进程中断中调用exit_flag.set啦   
你后面贴的代码 可以解决你的问题啦

linewer

tc1989tc 发表于 2015-09-18 21:05
sorry  现在你没在子进程中断中调用exit_flag.set啦   
你后面贴的代码 可以解决你的问题啦

嗯, 生成子进程之前我先忽略此信号,之后恢复.  感觉是修复了.   

晚上测试，发现一种情况,

1.     while True:
   
2.         is_alive = False
   
3.         for pid, proc in proc_pool.items():
   
4.             #???
   
5.             proc.join(timeout=1)
   
6.             if proc.is_alive():
   
7.                 is_alive = True
   
8.             else:
   
9.                 proc_pool.pop(pid)
   
10.                 sys.stderr.write('===pid %s exit===\n' % (pid))
    
11.                 if not exit_flag.is_set():
    
12.                    #start another process for this task!(1)abort (2)???normal quit???
    
13.                    signal(SIGTERM, SIG_IGN)
    
14.                    proc1 = mp.Process(target=worker)
    
15.                    proc1.start()
    
16.                    signal(SIGTERM, default_handler)
    
17.                    proc_pool[proc1.pid]=proc1
    
18.                    sys.stderr.write('===-pid %s: task %s-===\n' % (proc1.pid, proc_info[pid]))
    
19.                    proc_info[proc1.pid]=proc_info[pid]
    
20.                    proc_info.pop(pid)
    
21.                    #proc1.join(timeout=1)
    
22.                    is_alive = True    #这个如果注释掉，在子进程全异常退出时,会生成新子进程，但是父进程按逻辑应该退出，gdb看到是waitpid, 这个估计是等新子进程，这应该是逻辑错误导致的???!!!
    
23. 
    
24.                 #for k,v in proc_info.items():
    
25.                 #      sys.stderr.write('new===pid %s: task %s===\n' % (k, v))
    
26.                 #sys.stderr.write('~~~~~~~~~%s\n' % exit_flag.is_set())
    
27.                
    
28.         if not is_alive:
    
29.             break

复制代码

感觉同步碰上中断处理确实是个麻烦事，而且是在修别人的写的代码

tc1989tc

看文档如下描述：
When a process exits, it attempts to terminate all of its daemonic child processes.

大概原因如下：
你的子程序忽略的term信号处理，所有导致父进程退出时，去终止子进程时候
会存在阻塞等到子进程退出。
解决方法：
1.父进程在创建完子进程后在重新注册你的信号处理
2.子进程恢复系统默认的信号处理 SIG_TERM