asyncore关闭旧套接字

Question

我有一个带有asyncore的python上的tcp服务器：

class AsyncClientHandler(asyncore.dispatcher_with_send):
    def __init__(self,sock):
        asyncore.dispatcher_with_send.__init__(self,sock)

        self.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        self.message=""
        self.protocol=Protocol(DBSession, logger)

    def handle_read(self):
        data = self.recv(8192)
        if data:
            self.message+=data
            while TERMINATOR in self.message:   
                index=self.message.index(TERMINATOR)
                msg=self.message[:index]
                self.message=self.message[index+len(TERMINATOR):]

                answer=self.protocol.process_msg(msg, DBSession, tarif_dict)
                if answer:
                    msg = HEADER+answer+TERMINATOR
                    self.send(msg)

    def handle_close(self):
        self.close()

class AsyncServer(asyncore.dispatcher):
    def __init__(self, host, port): 
        asyncore.dispatcher.__init__(self)

        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)

        self.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        self.set_reuse_addr()
        self.bind((host, port))
        self.listen(5)

    def handle_accept(self):
        pair = self.accept()
        if pair is None:
            pass
        else:
            sock, addr = pair
            logging.info("Incoming connection from %s",repr(addr))
            AsyncClientHandler(sock)

一些客户端不会关闭连接，因此在某些情况下，服务器会因为大量套接字而崩溃。

如何在一段时间后关闭非活动套接字？settimeout不起作用。

Answer 1

要实现这一点，您可以使用TCP的Keepalive (就像您已经做过的那样)，并设置其延迟、ping……但此配置仅适用于长期连接，并且仅在Unixes上可用。让一些人阅读here。

您还可以对套接字进行一些调度，在经过一段时间后关闭套接字，或者在套接字处于活动状态时延迟它们。我用你的代码做了一个例子：

import sched, time

class SocketSched(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.daemon = True
        self.to_run = []
        self.scheds = {}
        self.start()

    def add(self, what):
        self.to_run.append(what.values()[0])
        self.scheds.update(what)

    def run(self):
        while True:
            if self.to_run:
                run = self.to_run.pop()
                if not run.empty(): run.run()
                else: self.to_run.append(run)

这里我们在不同的线程中定义了新的调度器类--这一点很重要，像asyncore.loop()一样，sched模块会持续阻塞。这需要稍微修改一下你的代码：

class AsyncClientHandler(asyncore.dispatcher_with_send):
    def __init__(self,sock, sch_class):
        ...
        self.delay = 10
        self.sch_class = sch_class
        self.sch = sched.scheduler(time.time, time.sleep)
        self.sch_class.add({self.fileno(): self.sch})
        self.event = self.sch_class.scheds[self.fileno()].enter(self.delay, 1, self.handle_close, ())

    ...

    def delay_close(self):
        self.sch_class.scheds[self.fileno()].cancel(self.event)
        self.event = self.sch_class.scheds[self.fileno()].enter(self.delay, 1, self.handle_close, ())

    ...

    def handle_close(self):
        try:
            self.sch_class.scheds[self.fileno()].cancel(self.event)
        except:
            pass
        ...

self.delay是以秒为单位的超时。在此时间过后，任何操作都不会延迟它，套接字将关闭。handle_close()中的代码行确保它不会因为调度器中的任务而被调用两次。

现在你必须在每个方法的开头添加self.delay_close()来确保socket是活动的，例如。handle_read()。

服务器类(获取SocketSched实例并将其传递给新通道)：

class AsyncServer(asyncore.dispatcher):
    def __init__(self, host, port, sch_class): 
        ...
        self.sch_class = sch_class

    ...

    def handle_accept(self):
    ...
            AsyncClientHandler(sock, self.sch_class)

准备好的。使用以下命令：

server = AsyncServer('', 1337, SocketSched())
asyncore.loop()

此解决方案可以工作，但在某些关闭事件上可能容易出错。无论如何，当给定的超时发生时，套接字将读取、延迟和关闭。不幸的是，运行这样的调度循环会占用一些CPU。