标签: python-asyncio

问题：asyncio.as_completed 产生结果后，如何获取原始任务的引用？

基本上与这个 C# 问题相同，除了在 Python：按完成排序任务后获取对原始任务的引用？

示例问题：

# Takes a list of WebClient objects,
# calls each one simultaneously,
# and yields the results immediately as they arrive
# to a synchronous caller.

def yieldThingsAsTheyArrive(webClients):

    tasks = []
    for webClient in webClients:
        # This is what we want to get a reference to later:
        task = webClient.fetch_thing()  # start long-running request asynchronously
        tasks.append(task)

    loop = asyncio.get_event_loop()
    for future in asyncio.as_completed(tasks):
        thing = loop.run_until_complete(future)  # since our caller is synchronous, …

我有 6 个大文件，每个文件都包含一个我使用 pickle 函数保存在硬盘中的字典对象。按顺序加载所有这些大约需要 600 秒。我想同时开始加载所有这些文件以加快进程。假设它们都具有相同的大小，我希望在 100 秒内加载它们。我使用 multiprocessing 和 apply_async 分别加载它们中的每一个，但它像顺序一样运行。这是我使用的代码，它不起作用。该代码适用于其中 3 个文件，但其中 6 个文件的代码相同。我将第三个文件放在另一个硬盘中以确保 IO 不受限制。

def loadMaps():    
    start = timeit.default_timer()
    procs = []
    pool = Pool(3)
    pool.apply_async(load1(),)
    pool.apply_async(load2(),)
    pool.apply_async(load3(),)
    pool.close()
    pool.join()
    stop = timeit.default_timer()
    print('loadFiles takes in %.1f seconds' % (stop - start))

我对 Python asyncio 和协议的工作方式缺乏了解。

似乎相关的文档： 状态机显示相关转换
class asyncio.Protocol
Protocol.eof_received()AbstractEventLoop.run_until_complete(future)

echo tcp 客户端的示例代码：

import asyncio

class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, loop):
        self.message = message.encode()

    def connection_made(self, transport):
        self.transport = transport
        self.write_data()

    def data_received(self, data):
        print('Data received: {!r}',len(data))
        self.write_data()

    def eof_received(self):
        print("eof")
        return True

    def write_data(self):
        print("write")
        self.transport.write(self.message)

    def connection_lost(self, exc):
        print('The server closed the connection')
        print('Stop the event loop')


loop = asyncio.get_event_loop()
message = 'Hello World!'

coro = loop.create_connection(lambda: EchoClientProtocol(message, loop),
                              '127.0.0.1', 5676)
loop.run_until_complete(coro)
print("done")

在回显服务器上连接时的输出：

write
Data received: {!r} …

我有一个模块可以阻止对某些 TCP 服务器的网络请求并接收响应。我必须将它集成到 asyncio 应用程序中。我的模块看起来像这样：

import socket

# class providing transport facilities
# can be implemented in any manner
# for example it can manage asyncio connection
class CustomTransport:
    def __init__(self, host, port):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.host = host
        self.port = port

    def write(self, data):
        left = len(data)
        while left:
            written = self.sock.send(data.encode('utf-8'))
            left = left - written
            data = data[written:]

    def read(self, sz):
        return self.sock.recv(sz)

    def open(self):
        self.sock.connect((self.host, self.port))

    def close(self):
        self.sock.shutdown(2)


# generated. shouldn't be modified
# …

我观察到该asyncio.run_coroutine_threadsafe函数不接受一般的可等待对象，我不明白这种限制的原因。观察

import asyncio


async def native_coro():
    return


@asyncio.coroutine
def generator_based_coro():
    return


class Awaitable:
    def __await__(self):
        return asyncio.Future()


loop = asyncio.get_event_loop()

asyncio.run_coroutine_threadsafe(native_coro(), loop)
asyncio.run_coroutine_threadsafe(generator_based_coro(), loop)
asyncio.run_coroutine_threadsafe(Awaitable(), loop)

使用 Python 3.6.6 运行它会产生

Traceback (most recent call last):
  File "awaitable.py", line 24, in <module>
    asyncio.run_coroutine_threadsafe(Awaitable(), loop)
  File "~/.local/python3.6/lib/python3.6/asyncio/tasks.py", line 714, in run_coroutine_threadsafe
    raise TypeError('A coroutine object is required')
TypeError: A coroutine object is required

第 24 行在哪里asyncio.run_coroutine_threadsafe(Awaitable(), loop)。

我知道我可以将我的可等待对象包装在一个定义如下的协程中

awaitable = Awaitable()

async def wrapper():
    return await awaitable

asyncio.run_coroutine_threadsafe(wrapper(), …

我正在尝试学习如何在 Python 3.7 中使用 AsyncIO，但我仍然对其原理有些困惑。

我的目标是编写一个简单的聊天程序，但是我需要使用环形网络拓扑——一个节点只知道它的两个邻居。当消息被发送时，它会被节点传递，直到它再次到达发送者。这意味着每个节点基本上同时是客户端和服务器。

我还需要能够检测到死节点，以便我的环不会损坏。

我认为每个节点为每个邻居建立单独的连接可能是一个很好的解决方案 -successor和predecessor.

class Node:
    ...
    def run():
        ...
        s = loop.create_connection(lambda: Client(...), addr1, port1)
        p = loop.create_server(lambda: Server(...), addr2, port2)
        successor = loop.run_until_complete(s)
        predecessor = loop.run_until_complete(p)
        loop.run_forever()
        ...
    ...

Server并且Client是实现asyncio.Protocol.

我想这样做的原因是，如果有消息通过圈子发送，它总是从 发送predecessor到successor。在I 的connection_lost方法中，predecessor我可以检测到它已断开连接并发送它predecessor的消息（通过整个环）以连接到我。

我希望能够将我从我predecessor进一步收到的消息发送到我的successor. 我还希望能够将包含我的地址的消息发送给我successor，以防万一我predecessor死了（此消息将从predecessor's发送Server.connection_lost()并一直传递到我死者predecessor的predecessor）。

我的问题是：我可以将接收到的数据从predecessor …

在哪里可以找到将 gRPC 与 asyncio 结合使用的示例，特别是如何使用 gRPC 和 asyncio 创建客户端

下面是一个简单的回显服务器。但是如果客户端在 10 秒内没有发送任何内容，我想关闭连接。

import asyncio


async def process(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    print("awaiting for data")
    line = await reader.readline()
    print(f"received {line}")
    writer.write(line)
    print(f"sent {line}")
    await writer.drain()
    print(f"Drained")


async def timeout(task: asyncio.Task, duration):
    print("timeout started")
    await asyncio.sleep(duration)
    print("client unresponsive, cancelling")
    task.cancel()
    print("task cancelled")


async def new_session(reader, writer):
    print("new session started")
    task = asyncio.create_task(process(reader, writer))
    timer = asyncio.create_task(timeout(task, 10))
    await task
    print("task complete")
    timer.cancel()
    print("timer cancelled")
    writer.close()
    print("writer closed")


async def a_main():
    server = await asyncio.start_server(new_session, port=8088)
    await server.serve_forever()


if __name__ == '__main__': …

我见过asyncio.gather 与 asyncio.wait，但我不确定这是否解决了这个特定问题。我想要做的是将asyncio.gather()协程包装在中asyncio.wait_for()，并带有一个timeout参数。我还需要满足以下条件：

• return_exceptions=True(from asyncio.gather()) - 而不是将异常传播到等待的任务gather()，我想在结果中包含异常实例
• 顺序：保留asyncio.gather()结果顺序与输入顺序相同的属性。（否则，将输出映射回输入。）。 asyncio.wait_for()没有达到这个标准，我不确定实现它的理想方式。

超时是针对整个 可等待asyncio.gather()列表的——如果它们在超时中被捕获或返回异常，那么这两种情况中的任何一种都应该在结果列表中放置一个异常实例。

考虑这个设置：

>>> import asyncio
>>> import random
>>> from time import perf_counter
>>> from typing import Iterable
>>> from pprint import pprint
>>> 
>>> async def coro(i, threshold=0.4):
...     await asyncio.sleep(i)
...     if i > threshold:
...         # For illustration's sake - some coroutines may raise,
...         # and we want …

如果我asyncio在顶级类中创建一系列任务，所有这些任务本质上都应该永远运行，如下所示：

self.event_loop.create_task(...)
self.event_loop.create_task(...)
self.event_loop.create_task(...)
...

self.event_loop.run_forever()

# Once we fall out of the event loop, collect all remaining tasks,
# cancel them, and terminate the asyncio event loop
tasks = asyncio.Task.all_tasks()
group = asyncio.gather(*tasks, return_exceptions=True)
group.cancel()
self.event_loop.run_until_complete(group)
self.event_loop.close()

上面的代码没有处理以下情况，我发现我越来越需要这种情况，而且我还没有在谷歌搜索或asyncio文档中看到一个例子：

如果其中一项任务因异常而失败，则不会处理该异常 - 所有其他任务都会继续进行，但该任务只是静默停止（异常输出除外）。

那么，我该如何：

• 设置要处理的异常，以便失败不再沉默
• 最重要的是，重新启动失败的任务，有效地self.event_loop.create_task(...)再次运行，只是为了那个任务？这似乎需要找到在事件循环中收到异常的任务，将其删除，然后添加一个新任务 - 我不清楚如何做到这一点。
• 允许没有问题的任务不间断地继续。希望避免处理收到异常的任务的任何副作用。