如何对异步传输对象执行*同步*读/写答案

【问题标题】：How to perform *synchronous* read/write into/from a asyncio transport object如何对异步传输对象执行*同步*读/写
【发布时间】：2015-09-19 12:55:07
【问题描述】：

我在 Windows 上使用 asyncio 并且引用了命名管道的传输对象：

class DataPipeHandler(asyncio.Protocol):
    def connection_made(self, trans):
        self.trans = trans # <<== this is a reference to a transport object of type _ProactorDuplexPipeTransport

loop = asyncio.get_event_loop()
server = loop.start_serving_pipe(lambda: DataPipeHandler(), r'\\.\pipe\test-pipe')

现在我想使用 self.trans 来同步写入然后从命名管道读取数据。我该怎么做？

同步执行此操作对我来说很重要，因为这是我正在使用管道进行的一种 RPC 调用（编写一些内容并快速获得响应）并且我确实希望 阻止偶数循环的所有其他活动 直到此“管道 RPC 调用”返回。如果在此 RPC 调用完成之前我不阻止事件循环的所有其他活动，我将产生不必要的副作用，因为循环将继续处理我不希望它处理的其他事件。

我想要做的（先写入管道然后读取）与从事件循环线程调用urllib2.urlopen(urllib2.Request('http://www.google.com')).read() 的人非常相似——这里所有的事件循环活动都将被阻塞，直到我们从一个远程 http 服务器。

我知道我可以调用 self.trans.write(data) 但这不会同步写入数据（据我了解它不会阻塞）

谢谢。

编辑：在第一条评论之后让我补充一下：

我知道我永远不应该阻塞事件循环，我可以使用同步原语来完成我想要的。但是假设您有一个事件循环正在并行执行 10 个不同的活动，其中一个正在执行某种 RPC（如上所述），并且所有其他 9 个活动应该被阻止，直到这个 RPC 完成。所以我有两个选择：

(1) 按照您对所有这 10 个活动的建议添加同步原语（锁/信号量/条件）以进行同步。

(2) 通过阻止写入然后阻止从管道读取/向管道读取来实现此 RPC。（假设我信任管道的另一端）

我知道这不是使用事件循环的常用方式，但在我的具体情况下，我认为 (2) 更好。（更简单的逻辑）

【问题讨论】：

我认为你做错了什么。你永远不应该阻塞事件循环。如果您需要阻止某些操作执行，请使用同步原语，如 asyncio.Lock、asyncio.Semapthore、asyncio.Condition 等。
假设您有第 11 个活动，不应该被阻止。使用同步原语，您仍然可以阻止您的 9 个活动并保持第 10 和 11 个运行。
我知道使用同步原语在控制哪些活动将被阻止和哪些活动不会被阻止方面提供了更大的灵活性。但就我而言，我有上面描述的固定情况。（进行 RPC 时需要阻塞所有其他事件循环活动）
好的。从传输实例 (sock = transport.get_extra_info('socket')) 中提取套接字并通过常规同步调用执行您想要的操作。
对于管道，它实际上是pipe = transport.get_extra_info('pipe') 调用。

标签： python windows python-asyncio

【解决方案1】：

我认为您必须使用线程同步原语来确保整个循环（当前线程）被阻塞。我认为最好的办法是使用线程队列和连接功能。

【讨论】：

【解决方案2】：

@安德鲁·斯维特洛夫

正如你所说，asyncio Synchronization Primitives 是正常的选择。

但是如何将事件循环协程包装和公开为同步函数 API 是我的问题。我不希望 API 使用者处理 asyncio 概念

更新：找到了很酷的答案：how can I package a coroutine as normal function in event loop?。

【讨论】：