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()

復(fù)制代碼

如上簡(jiǎn)化代碼示例要實(shí)現(xiàn)的功能為: 某子進(jìn)程異常退出時(shí)能自動(dòng)重新生成一個(gè)，正常運(yùn)行沒有問題,  但是在模擬測(cè)試時(shí)發(fā)現(xiàn),  在kill -9 某個(gè)子進(jìn)程或者某兩個(gè)時(shí), 偶爾會(huì)出現(xiàn)父進(jìn)程和其余子進(jìn)程掛起的問題,不能新生成一個(gè)新進(jìn)程(gdb看到是在sem_wait)，MS exit_flag死鎖了,這個(gè)應(yīng)該怎么修正，哪位指導(dǎo)下，多謝！
#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()

復(fù)制代碼

簡(jiǎn)單說說，以上是mp.Event代碼的set 和is_set方法，可以看出是都存在信號(hào)量的獲取的
所以你在中斷和非中斷都在調(diào)用Event的set和is_set  肯定存在死鎖的可能啊
盡量不要在中斷處理中調(diào)用信號(hào)量等處理

linewer

多謝LS,我后來自己看了下,在父子進(jìn)程中，有些東西是繼承的(復(fù)制的),比如信號(hào)及信號(hào)處理函數(shù),這樣才導(dǎo)致死鎖現(xiàn)象. 程序的本意就是父進(jìn)程收到終止信號(hào)后,設(shè)置Event變量,子進(jìn)程判斷后退出, 因?yàn)樘幚聿缓?其實(shí)是在修別人的bug),導(dǎo)致子進(jìn)程同樣處理此信號(hào), 改成啟動(dòng)子進(jìn)程之前忽略信號(hào),之后恢復(fù)，代碼如下,如果還有問題，歡迎斧正.

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()

復(fù)制代碼

tc1989tc

依然有問題啊  建議不要在中斷去exit_flags.set

linewer

tc1989tc 發(fā)表于 2015-09-18 17:44
依然有問題啊  建議不要在中斷去exit_flags.set

如果要在收到信號(hào)后，去設(shè)置exit_flags.set呢？設(shè)置個(gè)變量,放在父進(jìn)程主循環(huán)里做個(gè)判斷,然后設(shè)置exit_flags.set么? 或者還有別的好的方法么?

tc1989tc

回復(fù) 5# linewer


    把你的需求說出來吧  然后在看怎么解決

linewer

tc1989tc 發(fā)表于 2015-09-18 19:31
回復(fù) 5# linewer

就是父進(jìn)程通過multiprocess啟動(dòng)幾個(gè)進(jìn)程(這些進(jìn)程做一些獨(dú)立的工作,比如請(qǐng)求web抓取頁面),在父進(jìn)程收到終止信號(hào)后，設(shè)置exit_flag.set, 子進(jìn)程內(nèi)的工作函數(shù)用到這個(gè)exit_flag做一些判斷(平滑關(guān)閉鏈接),然后退出, 父進(jìn)程會(huì)監(jiān)控子進(jìn)程的運(yùn)行情況，退出就新啟一個(gè).   整個(gè)程序放到cron中，定期執(zhí)行/終止

tc1989tc

sorry  現(xiàn)在你沒在子進(jìn)程中斷中調(diào)用exit_flag.set啦   
你后面貼的代碼 可以解決你的問題啦

linewer

tc1989tc 發(fā)表于 2015-09-18 21:05
sorry  現(xiàn)在你沒在子進(jìn)程中斷中調(diào)用exit_flag.set啦   
你后面貼的代碼 可以解決你的問題啦

嗯, 生成子進(jìn)程之前我先忽略此信號(hào),之后恢復(fù).  感覺是修復(fù)了.   

晚上測(cè)試，發(fā)現(xiàn)一種情況,

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    #這個(gè)如果注釋掉，在子進(jìn)程全異常退出時(shí),會(huì)生成新子進(jìn)程，但是父進(jìn)程按邏輯應(yīng)該退出，gdb看到是waitpid, 這個(gè)估計(jì)是等新子進(jìn)程，這應(yīng)該是邏輯錯(cuò)誤導(dǎo)致的???!!!
    
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

復(fù)制代碼

感覺同步碰上中斷處理確實(shí)是個(gè)麻煩事，而且是在修別人的寫的代碼

tc1989tc

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

大概原因如下：
你的子程序忽略的term信號(hào)處理，所有導(dǎo)致父進(jìn)程退出時(shí)，去終止子進(jìn)程時(shí)候
會(huì)存在阻塞等到子進(jìn)程退出。
解決方法：
1.父進(jìn)程在創(chuàng)建完子進(jìn)程后在重新注冊(cè)你的信號(hào)處理
2.子進(jìn)程恢復(fù)系統(tǒng)默認(rèn)的信號(hào)處理 SIG_TERM