标签: coroutine

目前的Boost 1.55实现提供了两种单向协同程序.一个是pull-type,它是一个不带参数的协程并将值返回给主上下文; 另一种是push-type,它是一个从主要上下文接受参数但没有返回值的协程.

如何组合这两个来创建一个双向协程,它既接受参数又返回一个值？从表面上看似乎应该是可能的,但我无法弄清楚如何使用我所拥有的构建块来实现它boost::coroutine.在旧的Boosts中曾经有过双向协程,但它现在已被弃用且没有文档,因此我不应该依赖它.

即,我想要一些类似于此的东西:

void accumulate( pull_func &in, push_func &out )
{
  int x = 0;
  while ( in ) 
  {
    x += in.get() ; // transfers control from main context 
    out(x); // yields control to main context
  }
}

void caller( int n ) 
{
   bidirectional_coro( accumulate );
   for ( int i = 0 ; i < n ; ++i )
   {
      int y = accumulate(i); 
      printf( "%d ", y ); // "0 1 …

基于生成器的协同程序具有send()允许调用者和被调用者之间的双向通信并且从调用者恢复生成的生成器协同程序的方法.这是将生成器转换为协同程序的功能.

虽然新的本机async/await协同程序为异步I/O提供了出色的支持,但我看不出如何send()使用它们.明确禁止使用yieldin async函数,因此本机协同程序只能使用return语句返回一次.虽然await表达式为协程带来了新的值,但这些值来自被叫者,而不是来电者,等待的呼叫每次都从头开始评估,而不是从它停止的地方开始.

有没有办法从它停止的地方恢复返回的协同程序,并可能发送一个新值？如何使用本机协同程序模拟David Beazley 关于协同程序和并发的好奇课程中的技巧？

我想到的一般代码模式就像

def myCoroutine():
  ...
  while True:
    ...
    ping = yield(pong)
    ...

并在呼叫者

while True:
  ...
  buzz = myCoroutineGen.send(bizz)
  ...

编辑

我接受了Kevin的回答,但我注意到PEP 说

协同程序在内部基于生成器,因此它们共享实现.与生成器对象类似,协同程序具有throw(),send()和close()方法.

...

用于协同程序的throw(),send()方法用于推送值并将错误引发到类似Future的对象中.

显然,原生协同程序确实有一个send()？如果没有yield表达式来接收协程内的值,它如何工作？

我有一个asyncio.Protocol从服务器接收数据的子类.我正在存储这些数据(每行,因为数据是文本)asyncio.Queue.

import asyncio

q = asyncio.Queue()

class StreamProtocol(asyncio.Protocol):
    def __init__(self, loop):
        self.loop = loop
        self.transport = None

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

    def data_received(self, data):
        for message in data.decode().splitlines():
            yield q.put(message.rstrip())

    def connection_lost(self, exc):
        self.loop.stop()

loop = asyncio.get_event_loop()
coro = loop.create_connection(lambda: StreamProtocol(loop),
                              '127.0.0.1', '42')
loop.run_until_complete(coro)
loop.run_forever()
loop.close()

我想让另一个协程负责消耗队列中的数据并进行处理.

• 这应该是一个asyncio.Task吗？
• 如果队列变空是因为几秒钟没有收到数据怎么办？如何确保我的消费者不会停止(run_until_complete)？
• 有没有比使用全局变量更简洁的方法？

我正在使用Python 3.5,根据PEP 492应该可以访问async with语法,但是当我尝试使用它时,我得到了一个SyntaxError.我究竟做错了什么？

In [14]: sys.version
Out[14]: '3.5.2 (default, Oct 11 2016, 04:59:56) \n[GCC 4.2.1 Compatible Apple LLVM 8.0.0 (clang-800.0.38)]'

In [15]: async with aiohttp.ClientSession() as session:
  File "<ipython-input-15-9799c5ce74cf>", line 1
    async with aiohttp.ClientSession() as session:
             ^
SyntaxError: invalid syntax

在这里的文档:https://docs.python.org/3/library/asyncio-task.html,我发现很多yield from都可以替换await.

我想知道它们在Python 3.5中是否一直是等效的.有没有人有这个想法？

由于我们正在使用协程（使用 1.3.5），我们有很多崩溃：JobCancellationException - StandaloneCoroutine was cancelled。

我阅读了很多关于这些问题的线程，我在生产中尝试了很多解决方案，但总是发生崩溃。

在我们所有的视图模型中，我们都使用了视图模型范围，所以没关系。

但是在我们的数据层中，我们需要启动一个跟踪事件，即触发即忘任务。在第一步中，我们使用了一个GlobalScope.launch. 我认为 CancelletationException 是由于这个全局范围，所以我删除了它并使用 aSupervisorJob和 a在数据层中创建了一个扩展CoroutineExceptionHandler：

private val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
private val coroutineExceptionHandler by lazy { CoroutineExceptionHandler { _, throwable -> logw("Error occurred inside Coroutine.", throwable) } }

fun launchOnApp(block: suspend CoroutineScope.() -> Unit) {
    appScope.launch(coroutineExceptionHandler) { block() }
}

但我总是看到这段代码崩溃。我需要使用cancelAndJoin方法吗？我可以将哪种策略与干净的建筑和这种工作一起使用？

提前致谢

我使用C#迭代器作为协同程序的替代品,它一直很好用.我想切换到async/await,因为我认为语法更清晰,它给了我类型安全. 在这篇(过时的)博客文章中,Jon Skeet展示了实现它的可能方式.

我选择采用略有不同的方式(通过实现我自己SynchronizationContext和使用Task.Yield).这很好.

然后我意识到会有问题; 目前协程不必完成运行.它可以在任何产生的点上优雅地停止.我们可能有这样的代码:

private IEnumerator Sleep(int milliseconds)
{
    Stopwatch timer = Stopwatch.StartNew();
    do
    {
        yield return null;
    }
    while (timer.ElapsedMilliseconds < milliseconds);
}

private IEnumerator CoroutineMain()
{
    try
    {
        // Do something that runs over several frames
        yield return Coroutine.Sleep(5000);
    }
    finally
    {
        Log("Coroutine finished, either after 5 seconds, or because it was stopped");
    }
}

协程通过跟踪堆栈中的所有枚举器来工作.C#编译器生成一个Dispose函数,可以调用该函数以确保正确调用'finally'块CoroutineMain,即使枚举未完成.这样我们就可以优雅地停止协程,并通过调用堆栈Dispose上的所有IEnumerator对象来确保最终调用块.这基本上是手动展开.

当我用async/await编写我的实现时,我意识到我们会失去这个功能,除非我弄错了.然后我查找了其他协同解决方案,看起来Jon Skeet的版本看起来也没有任何处理方式.

我能想到处理这个问题的唯一方法就是拥有我们自己的自定义'Yield'函数,它会检查协程是否被停止,然后引发一个表明这个的异常.这将传播,执行finally块,然后被捕获到根附近的某处.我不觉得这很漂亮,因为第三方代码可能会捕获异常.

我误解了什么,这是否可以更容易地做到？或者我需要以异常方式执行此操作吗？

编辑:已请求更多信息/代码,所以这里有一些.我可以保证这只会在一个线程上运行,所以这里没有涉及线程.我们当前的协程实现看起来有点像这样(这是简化的,但它适用于这个简单的情况):

public sealed …

我正在开发一个使用python3.4的asyncio进行网络连接的应用程序.当此应用程序干净地关闭时,节点需要与集线器"断开连接".此断开连接是一个需要网络连接的活动进程,因此循环需要在关闭之前等待此操作完成.

我的问题是使用协同程序作为信号处理程序将导致应用程序不关闭.请考虑以下示例:

import asyncio
import functools
import os
import signal

@asyncio.coroutine
def ask_exit(signame):
    print("got signal %s: exit" % signame)
    yield from asyncio.sleep(10.0)
    loop.stop()

loop = asyncio.get_event_loop()
for signame in ('SIGINT', 'SIGTERM'):
    loop.add_signal_handler(getattr(signal, signame),
                                        functools.partial(ask_exit, signame))

print("Event loop running forever, press CTRL+c to interrupt.")
print("pid %s: send SIGINT or SIGTERM to exit." % os.getpid())
loop.run_forever()

如果您运行此示例然后按Ctrl + C,则不会发生任何事情.问题是,如何使用siganls和coroutines来实现这种行为？

环境:C和micropython虚拟机中的协作RTOS是其中的任务之一.

为了使虚拟机不会阻止其它RTOS的任务,我插入RTOS_sleep()的vm.c:DISPATCH(),这样,每次执行字节码后,虚拟机放弃控制到下一个RTOS任务.

我创建了一个uPy接口,使用生产者 - 消费者设计模式从物理数据总线异步获取数据 - 可以是CAN,SPI,以太网.

在uPy中的用法:

can_q = CANbus.queue()
message = can_q.get()

C中的实现can_q.get()不会阻止RTOS:它轮询C队列,如果没有收到消息,它会调用RTOS_sleep()给另一个任务填充队列的机会.事物是同步的,因为C队列仅由另一个RTOS任务更新,而RTOS任务仅在RTOS_sleep()被调用时切换,即协作

C实现基本上是:

// gives chance for c-queue to be filled by other RTOS task
while(c_queue_empty() == true) RTOS_sleep(); 
return c_queue_get_message();

尽管Python语句can_q.get()不会阻止RTOS,但它会阻止uPy脚本.我想重写它,所以我可以使用async defie coroutine并让它不阻止uPy脚本.

不确定语法,但这样的事情:

can_q = CANbus.queue()
message = await can_q.get()

题

如何编写C函数以便我可以await使用它？

我更喜欢CPython和micropython的答案,但我会接受仅限CPython的答案.

关于 Python 协程（我主要是指async/await）是无堆栈的还是有堆栈的，我看到了相互矛盾的观点。

一些消息来源称它们是堆栈的：

• http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2020/p2074r0.pdf

  “Python 协程是堆栈式的。”

• Python 中的协程与 Lua 中的协程相比如何？

  是的，Python 协程是堆栈式的、一流的且不对称的。

虽然其他人似乎暗示它们是无堆栈的，例如https://gamelisp.rs/reference/coroutines.html

GameLisp 的协程遵循 Rust、Python、C# 和 C++ 设置的模型。我们的协程是“无堆栈的”

总的来说，我的理解始终是，任何有意义的 async/await 实现都意味着无堆栈协程，而堆栈协程基本上是纤程（用户空间线程，通常或多或少地协作切换），例如 goroutine、Boost.Coroutine，显然是 Lua 中的协程等。

我的理解正确吗？或者 Python 协程在某种程度上与 C++ 中的协程有根本的不同，并且是堆栈式的？或者上述来源的作者是否有不同的意思？