多个进程共享一个监听套接字:当新进程进入时,为什么旧进程会停止?
这些代码是我的代理程序的Server部分,其功能是创建socket并fork四个进程一一接受。
在我的程序中,我使用 gevent 模型来调度我的所有函数,在我将其更改为多进程之前,我的程序是没问题的。但是现在当我使用第二个进程时,第一个进程停止运行,我没有找到问题所在,也许是“接受”函数或者我的事件停止调度。
已经困扰我两天了,希望有人能帮助我。
顺便说一句,我的英语很差,我尽力解释一下,希望你能理解。
class Client(object):
def __init__(self, ent, ev):
...
def receive( self ):
...
if "Content-Length" in dic:
self.ent_s_send = core.event(core.EV_WRITE,
self.conn.fileno(),
self.ser_send,
[self.conn,self.body]
)
self.recv_ent = core.event(core.EV_READ,
self.sock.fileno(),
self.recv_content
)
self.recv_ent.add()
...
def recv_content(self, ent, ev):
...
self.n = self.sock.recv_into(self.msg,
min(self.total-self.num, 20000),
socket.MSG_DONTWAIT)
**time.sleep(0.1)**
#if i add it here to let the event slow down the problem solved, how it could be?
self.num += self.n
self.msg_buffer.fromstring(self.msg.tostring()[:self.n])
...
if self.total > self.num: #if not the last msg continue recving and sending...
self.ent_s_send.add()
self.recv_ent.add()
...
def ser_send(self, ent, ev):
...
num = self.conn.send(self.msg_buffer,socket.MSG_DONTWAIT)
...
self.msg_buffer = self.msg_buffer[num:]
...
...
class Server(object):
def __init__( self ):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.bind(('localhost', 18001))
self.sock.listen(50)
self.mutex = multiprocessing.Lock()
def loop( self, ):
for i in range(0,4):
pid = os.fork()
if pid == 0 or pid == -1:
break
if pid == -1:
print "Fork failed!!!"
sys.exit()
elif pid == 0: **# create four child ps to accept the socket!**
print "Child PID = %d" % os.getpid()
core.init()
self.event = core.event(core.EV_READ,
self.sock.fileno(),
self.onlink)
self.event.add()
core.dispatch()
else:
os.wait()
def onlink( self, ent, ev):
self.mutex.acquire()
print 'Accept PID = %s' % os.getpid()
try:
self.conn, self.addr = self.sock.accept()
**#I think 'accept' is the the problem, but I cannot see how.**
except socket.error, why:
if why.args[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED]:
return
else:
raise
print self.sock,self.conn,self.addr
self.mutex.release()
print 'Release PID = %s' % os.getpid()
cc = Chat( self.conn, self.sock )
self.event.add()
if __name__ == '__main__':
s1 = Server()
s1.loop()
These code is the Server part of my proxy program, and its function is creating the socket and fork four process to accept one by one.
In my program I use gevent model to dispatch all my function and before I alter it to multiple process, my program is all right. but now when I use the second process, the first one stop running, I don't find where is wrong, maybe the 'accept' function or my event is stop dispatch.
It have already bothered me for two days I hope someone can help me.
BTW, my English is poor, I try my best to explain it, hoping you can understand.
class Client(object):
def __init__(self, ent, ev):
...
def receive( self ):
...
if "Content-Length" in dic:
self.ent_s_send = core.event(core.EV_WRITE,
self.conn.fileno(),
self.ser_send,
[self.conn,self.body]
)
self.recv_ent = core.event(core.EV_READ,
self.sock.fileno(),
self.recv_content
)
self.recv_ent.add()
...
def recv_content(self, ent, ev):
...
self.n = self.sock.recv_into(self.msg,
min(self.total-self.num, 20000),
socket.MSG_DONTWAIT)
**time.sleep(0.1)**
#if i add it here to let the event slow down the problem solved, how it could be?
self.num += self.n
self.msg_buffer.fromstring(self.msg.tostring()[:self.n])
...
if self.total > self.num: #if not the last msg continue recving and sending...
self.ent_s_send.add()
self.recv_ent.add()
...
def ser_send(self, ent, ev):
...
num = self.conn.send(self.msg_buffer,socket.MSG_DONTWAIT)
...
self.msg_buffer = self.msg_buffer[num:]
...
...
class Server(object):
def __init__( self ):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.bind(('localhost', 18001))
self.sock.listen(50)
self.mutex = multiprocessing.Lock()
def loop( self, ):
for i in range(0,4):
pid = os.fork()
if pid == 0 or pid == -1:
break
if pid == -1:
print "Fork failed!!!"
sys.exit()
elif pid == 0: **# create four child ps to accept the socket!**
print "Child PID = %d" % os.getpid()
core.init()
self.event = core.event(core.EV_READ,
self.sock.fileno(),
self.onlink)
self.event.add()
core.dispatch()
else:
os.wait()
def onlink( self, ent, ev):
self.mutex.acquire()
print 'Accept PID = %s' % os.getpid()
try:
self.conn, self.addr = self.sock.accept()
**#I think 'accept' is the the problem, but I cannot see how.**
except socket.error, why:
if why.args[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED]:
return
else:
raise
print self.sock,self.conn,self.addr
self.mutex.release()
print 'Release PID = %s' % os.getpid()
cc = Chat( self.conn, self.sock )
self.event.add()
if __name__ == '__main__':
s1 = Server()
s1.loop()
如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。
绑定邮箱获取回复消息
由于您还没有绑定你的真实邮箱,如果其他用户或者作者回复了您的评论,将不能在第一时间通知您!
发布评论
评论(1)
accept()
是一个阻塞调用
。它将无限期地等待客户端连接。在这样的阻塞操作上持有互斥体是一个坏主意
TM,因为你完全锁定了所有其他并发进程。另外,正如 @Maxim 在评论中指出的那样,您实际上不需要锁定
accept()
。只需让操作系统仲裁传入连接的出队并将它们分派到您的进程即可。accept()
is ablocking call
. It'll wait indefinitely for a client to connect. Holding a mutex over a blocking operation like that is aBad Idea
TM since you totally lock all other concurrent processes out.Also, as @Maxim noted in the comments, you don't really need to lock around
accept()
. Just let the OS arbitrate dequeuing of incoming connections and dispatch them to your processes.