python异步循环每15秒发送udp有效负载，并在60秒后停止尝试

Question

我的目标是创建一个连接到UDP ip地址并接收响应的异步循环。但是，如果15秒内未收到响应，请重新发送请求，并在60秒后停止尝试。

main.py

async def main():
    tasks = []
    for i, l in zip(result["tr"], range(len(result["tr"]))):
        task = create_task(udp.connect(urlparse(i).hostname, urlparse(i).port))
        tasks.append(task)

    done, pending = await wait(tasks)
    for i in done:
        print(i.result())

udpClient.py

async def connect(udp, port):
    payload = b'sample_data_input'
    on_conn_lost, datagram_data = loop.create_future(), loop.create_future()

    transport, protocol = await self.default_event_loop.create_datagram_endpoint(
        lambda: ProtocolFactoryUDP(payload, on_conn_lost, datagram_data),
        remote_addr=(str(udp), int(port)))

    try:
        await on_conn_lost
        return datagram_data
    except Exception as e:
        print(e)

ProtocolFactoryUDP-> This is same as the example on python.org

class ProtocolFactoryUDP:
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost
        self.transport = None
        self.datagram_data = datagram_data

    def connection_made(self, transport):
        self.transport = transport
        print('Send:', self.message)
        self.transport.sendto(self.message.encode())

    def datagram_received(self, data, addr):
        print("Received:", data.decode())
        self.datagram_data.set_result((data, addr))
        print("Close the socket")
        self.transport.close()

    def error_received(self, exc):
        print('Error received:', exc)

    def connection_lost(self, exc):
        print("Connection closed")
        self.on_con_lost.set_result(True)

当前，此代码使我可以连接到UDP目标并打印出响应。

但是，这没有达到我的目标，因为如果服务器不响应任何内容/响应时间太长，udpClient最终将永远在那里等待。如何创建异步循环，以便如果在15秒内未收到响应，请重新发送请求，并在尝试60秒后以空有效负载退出（或正常退出）。

用保罗更新的代码编辑1

main.py

for i, l in zip(result["tr"], range(len(result["tr"]))):
    task = create_task(udp.connect(urlparse(i).hostname, urlparse(i).port))
    tasks.append(task)

for f in as_completed(tasks):
    result = await f

udpClient.py

...
for _ in range(4):
    try:
        transport, protocol = await self.default_event_loop.create_datagram_endpoint(
            lambda: ProtocolFactoryUDP(payload, on_conn_lost, datagram_data),
            remote_addr=(str(udp), int(port)))

        await asyncio.wait_for(on_conn_lost, 15.0)  # -> Met with CancelledError here.. 
        return datagram_data
    except TimeoutError:
        pass

Answer 1

您已经对要执行的操作给出了完美的描述，而这通常是最难的部分。为了将其简化为代码，您可以查看asyncio.wait_for函数以及asyncio.as_completed函数。这是我认为您想要的草图。我将UDP详细信息留给您。

wait_for函数采用超时值（在您的情况下为15s）。您想尝试4次，所以您所需要的只是一个简单的for循环。您捕获TimeoutError并重复循环；让其他错误传播回调用者。如果没有超时，您只需返回即可终止任务。

您可以在结果完成时打印结果，或者使用asyncio.gather（return_exceptions = True）等到结果全部完成。

async def main():
    tasks = []
    for i, l in zip(result["tr"], range(len(result["tr"]))):
        coro = my_task(udp.connect(urlparse(i).hostname, urlparse(i).port))
        tasks.append(asyncio.create_task(coro))

    for i in asyncio.as_completed(tasks):
        print(i.result())

async def my_task(host, port):
    for _ in range(4):
        try:
            await asyncio.wait_for(do_something_with_udp(host, port), 15.0)
            return
        except asyncio.TimeoutError:
            pass