如何让协程因超时而停止？
我不明白为什么 asyncio.wait_for() 对我不起作用。我有这样一段代码（计划实现 telnet 客户端）：
def expect(self, pattern, timeout=20):
if type(pattern) == str:
pattern = pattern.encode('ascii', 'ignore')
return self.loop.run_until_complete(asyncio.wait_for(self.asyncxpect(pattern), timeout))
async def asyncxpect(self, pattern): #receives data in a cumulative way until match is found
regexp = re.compile(b'(?P<payload>[\s\S]*)(?P<pattern>%s)' %pattern)
self.buffer = b''
while True:
# add timeout
# add exception handling for unexpectedly closed connections
data = await self.loop.sock_recv(self.sock, 10000)
self.buffer += data
m = re.match(regexp, self.buffer)
if m:
payload = m.group('payload')
match = m.group('pattern')
return payload, match
正如我所认为的，这段代码在某个时刻（在等待语句中）将控制权返回给事件循环。我认为当没有更多数据可以接收时应该发生这种情况。如果事件循环有控制权，它可以超时停止。
但是，如果服务器没有发送任何有用的（匹配的）我的代码就会在这个循环中绊倒，就在等待点。 …
查看 clang-5.0发行说明，它说“ C++ 协程 TS已落地”，并建议查看此示例以开始使用。我显然可以在线运行该示例，因此我决定在我的 Ubuntu Server 16.04 Xenial 机器上尝试一下。
从干净的安装中，我刚刚尝试过
wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key | sudo apt-key add -
sudo apt-add-repository "deb http://apt.llvm.org/xenial/ llvm-toolchain-xenial-5.0 main"
sudo apt-get update
sudo apt-get install -y clang-5.0 libc++-dev
并直奔
clang++-5.0 -Wall -Wextra -std=gnu++2a "-fcoroutines-ts" "-stdlib=libc++" -o test.out test.cpp
（是上述示例test.cpp的复制和粘贴）
然而，我只是得到
coroutines.cpp:2:10: fatal error: 'experimental/coroutine' file not found
#include <experimental/coroutine>
所以我可以想象我缺少一些东西，无论是安装还是在编译标志中。我尝试在网上查找，但找不到任何有用的东西。
我究竟做错了什么？
我正在使用协程来设置重复延迟，如下所示。
在我清醒的时候我有
StartCoroutine(RandomMove());
然后再往下
IEnumerator RandomMove()
{
while (true)
{
// print(Time.time);
yield return new WaitForSeconds(foo);
// print(Time.time);
}
}
其中“foo”是一个随机浮点值，我会在每次迭代时更改它。假设 foo 是 10 秒，并且在延迟中，我需要重置延迟，以便它再次从 10 开始“倒计时”。
我将如何实现这个目标？我应该使用计时器吗？
所以我有这个协程 API，由我扩展，基于我在这里找到的代码：https ://the8bitpimp.wordpress.com/2014/10/21/coroutines-x64-and-visual-studio/
struct mcontext {
U64 regs[8];
U64 stack_pointer;
U64 return_address;
U64 coroutine_return_address;
};
struct costate {
struct mcontext callee;
struct mcontext caller;
U32 state;
};
void coprepare(struct costate **token,
void *stack, U64 stack_size, cofunc_t func); /* C code */
void coenter(struct costate *token, void *arg); /* ASM code */
void coyield(struct costate *token); /* ASM code */
int coresume(struct costate *token); /* ASM code, new */
我坚持实施 coyield()。coyield() 可以用 C 语言编写，但我遇到问题的是程序集。这是我到目前为止所得到的（MASM/VC++ 语法）。
;;; function: void _yield(struct mcontext …我的问题是一个非常广泛的问题。既然可以使用对象，为什么还要使用协程呢？我可以轻松实现一个迭代器，该迭代器仅在 next() 下运行一次，之后您必须调用 o.send(x)。除了可以通过 OOP 实现状态持久性之外，协程还有其他作用吗？它们更轻吗？它只是语法糖吗？我实际上可以对生成器与迭代器提出同样的问题，但我想我已经读到生成器只是语法糖。
如果确实如此，为什么协程如此重要？我确信我错过了一些关于他们的事情，但我不知道是什么。
我试图理解 python asyncio 的call_soon_threadsafe API，但失败了，使用下面的示例代码，如果我的simple协程想要返回某些内容，我应该如何从调用方获取返回值？
import time
import asyncio as aio
import uvloop
from threading import Thread
aio.set_event_loop_policy(uvloop.EventLoopPolicy())
async def simple(a, fut:aio.Future):
await aio.sleep(a)
return fut.set_result(a)
def delegator(loop):
aio.set_event_loop(loop)
loop.run_forever()
loop_exec = aio.new_event_loop()
t = Thread(target=delegator, args=(loop_exec,))
t.start()
if __name__ == '__main__':
start_time = time.time()
fut = loop_exec.create_future() # tried to get back returned value by future
handle = loop_exec.call_soon_threadsafe(aio.ensure_future, simple(3, fut))
res = aio.wait_for(fut, 10)
print('Time consumed: {}s'.format(time.time() - start_time))
print('>>>>>>>>>>', res)
# Output
Time consumed: …考虑这个用例：
class GetPhotosUseCase(
private val photosRepository: IPhotosRepository,
private val favoritesRepository: IFavoritesRepository
) : IGetPhotosUseCase {
override suspend fun getPhotos(): List<Photo> {
val photos = photosRepository.getPhotos()
val favoriteIds = favoritesRepository.getFavoriteIds()
return photos.map {
it.copy(isFavorite = favoriteIds.contains(it.id))
}
}
}
interface IPhotosRepository {
suspend fun getPhotos(): List<Photo>
}
interface IFavoritesRepository {
suspend fun getFavoriteIds(): List<Int>
}
我从两个不同的来源获取数据并将它们组合起来。现在，这是连续运行的。当我想并行运行photosRepository.getPhotos()以favoritesRepository.getFavoriteIds()节省执行时间时，我的天真的做法是：
override suspend fun getPhotos(): List<Photo> {
val photosDeferred = GlobalScope.async { photosRepository.getPhotos() }
val favoriteIdsDeferred = GlobalScope.async { favoritesRepository.getFavoriteIds() }
return …我有一个关于 kotlin 流合并的问题。看看下面的乐趣。
suspend fun method(filter: String): Flow<List<Model>> {
// Search.
val models: List<Model> = repo.getModels(filter) // suspend function
// Get favorites
val favoritesFlow: Flow<List<Int>> = otherRepo.getFavorites()
// Return models as Flow, but mark/unmark every model as favorite when favoritesFlow is updated.
??? val result = models + favoritesFlow ????
return result
}
我需要返回模型列表流，但是当最喜欢的流更改时，我必须将每个模型标记或取消标记为最喜欢的。你知道我该怎么做吗？
我完全理解 suspendCoroutine 与 suspendCancellableCoroutine 在我的示例中是如何工作的。但我想知道为什么println("I finish") （第 13 行 - viewscope 块中的第二行）在我调用 viewScope.cancel() 之后执行。我可以在该行之前使用 isActive 标志修复它，但我不想检查每一行。我在那里缺少什么。我怎样才能取消范围？谢谢
import kotlinx.coroutines.*
import java.lang.Exception
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.resume
import kotlin.coroutines.suspendCoroutine
fun main() {
val parentJob = Job()
val viewScope = CoroutineScope(Dispatchers.IO + parentJob)
viewScope.launch {
println(tryMe())
println("I finished")
}
Thread.sleep(2000)
viewScope.cancel()
Thread.sleep(10000)
}
suspend fun tryMe() = suspendCoroutine<String> {
println("I started working")
Thread.sleep(6000)
println("Im still working :O")
it.resume("I returned object at the end :)")
}
suspend fun tryMe2() = suspendCancellableCoroutine<String> {
println("I …我是 asyncio 的新手，我想利用它来为我的 pyqt 应用程序处理通过 tcp 连接的通信。
我制作了这个简单的演示来学习如何在 asyncio 上下文中处理 QT 循环。我看过与此相关的其他帖子，但它们比我目前想做的要复杂得多。因此，我在一个单独的窗口中启动服务器客户端，以便它进行监听，并尝试通过小部件上的简单按钮单击事件发送消息。正如它所得到的准系统......我的问题是它不起作用。
我希望能够进行一次信息交换以及端口对流保持开放的情况。我认为这些任务在 asyncio 中会很简单，但在这一点上让它与 qt 很好地配合似乎很困难。
现在我得到了
RuntimeWarning: coroutine 'PushButton.sendmessage' was never awaited
rslt = self.__app.exec_()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
我不知道从哪里开始解决这个问题。
测试.py
from PyQt5.QtWidgets import QApplication, QWidget, QPushButton
from PyQt5.QtGui import QIcon
from PyQt5.QtCore import Qt, pyqtSlot
import sys
import asyncio
from asyncqt import QEventLoop
from async_client import tcp_echo_client
class PushButton(QWidget):
loop = None
def __init__(self,app_loop):
super(PushButton,self).__init__()
self.initUI()
self.loop = loop
def initUI(self):
self.setWindowTitle("PushButton") …