使用变量超时定期发送数据的Python异步。有更好的办法吗？

Question

我想编写一个服务器，客户端可以连接到该服务器并接收定期更新，而无需轮询。我在asyncore中遇到的问题是，如果在调用dispatcher.writable（）时不返回true，则必须等到asyncore.loop超时后（默认值为30s）。

我试图解决这一问题的两种方法是：1）将超时时间减少到一个较低的值，或2）查询连接，以便它们下次更新时生成足够的超时值。但是，如果您在“man 2 Select tut”中提到“Select Law”，它会指出，“您应该始终尝试在没有超时的情况下使用Select（）。”

有更好的办法吗？也许扭曲了？我想尽量避免额外的线程。我将在这里包括变量timeout示例：

#!/usr/bin/python

import time
import socket
import asyncore


# in seconds
UPDATE_PERIOD = 4.0

class Channel(asyncore.dispatcher):

    def __init__(self, sock, sck_map):
        asyncore.dispatcher.__init__(self, sock=sock, map=sck_map)
        self.last_update = 0.0  # should update immediately
        self.send_buf = ''
        self.recv_buf = ''

    def writable(self):
        return len(self.send_buf) > 0

    def handle_write(self):
        nbytes = self.send(self.send_buf)
        self.send_buf = self.send_buf[nbytes:]

    def handle_read(self):
        print 'read'
        print 'recv:', self.recv(4096)

    def handle_close(self):
        print 'close'
        self.close()

    # added for variable timeout
    def update(self):
        if time.time() >= self.next_update():
            self.send_buf += 'hello %f\n'%(time.time())
            self.last_update = time.time()

    def next_update(self):
        return self.last_update + UPDATE_PERIOD


class Server(asyncore.dispatcher):

    def __init__(self, port, sck_map):
        asyncore.dispatcher.__init__(self, map=sck_map)
        self.port = port
        self.sck_map = sck_map
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.bind( ("", port))
        self.listen(16)
        print "listening on port", self.port

    def handle_accept(self):
        (conn, addr) = self.accept()
        Channel(sock=conn, sck_map=self.sck_map)

    # added for variable timeout
    def update(self):
        pass

    def next_update(self):
        return None


sck_map = {}

server = Server(9090, sck_map)
while True:
    next_update = time.time() + 30.0
    for c in sck_map.values():
        c.update()  # <-- fill write buffers
        n = c.next_update()
        #print 'n:',n
        if n is not None:
            next_update = min(next_update, n)
    _timeout = max(0.1, next_update - time.time())

    asyncore.loop(timeout=_timeout, count=1, map=sck_map)

Answer 1

也许你可以用sched.scheduler来完成，就像这样（n.b.未测试）：

import sched, asyncore, time

# Create a scheduler with a delay function that calls asyncore.loop
scheduler = sched.scheduler(time.time, lambda t: _poll_loop(t, time.time()) )

# Add the update timeouts with scheduler.enter
# ...

def _poll_loop(timeout, start_time):  
  asyncore.loop(timeout, count=1)
  finish_time = time.time()
  timeleft = finish_time - start_time
  if timeleft > timeout:  # there was a message and the timeout delay is not finished
    _poll_loop(timeleft, finish_time) # so wait some more polling the socket

def main_loop():
  while True:
    if scheduler.empty():
      asyncore.loop(30.0, count=1) # just default timeout, use what suits you
      # add other work that might create scheduled events here
    else:
      scheduler.run()