python asyncore 在客户端连接后使用 100% CPU

Question

我正在使用 asyncore 来实现发布订阅。我知道使用 twisted 或 ZMQ 可能是更好的解决方案，但在这种情况下，它需要是纯 python。等待连接时，CPU 使用率约为 1%，一旦客户端连接，CPU 使用率就会跃升至 100%。即使在客户端断开连接后它也不会回落。

我的服务器类:

class Host(asyncore.dispatcher):

    log = logging.getLogger('Host')

    def __init__(self, port=7655):
        asyncore.dispatcher.__init__(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.set_reuse_addr()
        self.bind(('0.0.0.0', port,))
        self.listen(5)
        self.clients = []

    def handle_accept(self):
        socket, addr = self.accept()
        self.log.info("Aceepted client at {0}:{1}".format(addr[0], addr[1]))
        self.clients.append(RemoteClient(self, socket, addr))

    def broadcast(self, message):
        self.log.info("Broadcasting message: {0}".format(message))
        for client in self.clients:
            client.message(message)

还有我的处理程序:

class RemoteClient(asyncore.dispatcher):

    log = logging.getLogger('Host')

    def __init__(self, host, socket, address):
        asyncore.dispatcher.__init__(self, socket)
        self.host = host
        self.outbox = collections.deque()

    def message(self, message):
        self.outbox.append(message)

    def handle_write(self):
        if not self.outbox:
            return
        message = self.outbox.popleft()
        if len(message) > MAX_MESSAGE_LENGTH:
            raise ValueError('Message too long')
        self.send(message)

    def handle_close(self):
        self.host.clients.remove(self)
        self.log.info("Client removed from list")
        self.close()

    def handle_error(self):
        self.log.error("Socket error")

我曾尝试寻找解决方案，但似乎无法弄清楚发生了什么。任何帮助表示赞赏!

Answer 1

说明

你的问题是你没有覆盖方法 asyncore.dispatcher.writeable ，默认实现:

def writable(self):
    return True

这导致 asyncore.poll在 while True 循环中运行(因此使用 100% CPU):

def poll(timeout=0.0, map=None):
    if map is None:
        map = socket_map
    if map:
        r = []; w = []; e = []
        for fd, obj in map.items():
            is_r = obj.readable()
            is_w = obj.writable() # This is always true
            if is_r:
                r.append(fd)
            if is_w:
                w.append(fd)
            if is_r or is_w:
                e.append(fd) # This always happens
        if [] == r == w == e:
            time.sleep(timeout) # This never happens
            return

        try:
            # Here, having w (as function parameter) set to non-empty value
            # causes select to immediately return without waiting with w set
            # just to your client
            r, w, e = select.select(r, w, e, timeout) 

---

解决方案

我认为真正干净的解决方案是使用某种 threading.Event 机制重新实现 asyncore，该机制将强制等待可写对象，但到目前为止这对我有用:

# Add to RemoteClient
def writable(self):
    ''' It has point to call handle_write only when there's something in outbox
        Having this method always returning true will cause 100% CPU usage
    '''
    return bool(self.outbox)

# When you start loop
asyncore.loop(timeout=5.0) 

您还可以在 official documentation 中的示例中看到 writable() 被覆盖.

我只是更喜欢能够在 30 秒后结束等待。