2

これらのコードは、私のプロキシ プログラムのサーバー部分であり、その関数はソケットを作成し、1 つずつ受け入れる 4 つのプロセスをフォークします。

私のプログラムでは、すべての関数をディスパッチするために gevent モデルを使用していますが、それを複数のプロセスに変更する前に、私のプログラムは問題ありません。しかし、2番目のプロセスを使用すると、最初のプロセスが実行を停止し、どこが間違っているのかわかりません。おそらく「受け入れ」機能または私のイベントがディスパッチを停止することです。

すでに2日間悩まされています。誰かが私を助けてくれることを願っています。

ところで、私の英語は下手です。理解できるように最善を尽くして説明します。 

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

1 に答える 1

1

accept()ですblocking call。クライアントが接続するまで無期限に待機します。そのようなブロッキング操作でミューテックスを保持することは、他のすべての同時プロセスを完全にロックアウトするため、 TMです。Bad Idea

また、@Maxim がコメントで指摘したように、実際にロックする必要はありませんaccept()。着信接続のデキューを OS に調停させ、それらをプロセスにディスパッチするだけです。

于 2011-08-17T15:11:08.493 に答える