在学习了Beazley关于协同程序的相同课程(我必须说是非常棒的)之后,我问了自己一个同样的问题——如何调整代码以与Python 3.5中引入的本地协同程序一起工作?
事实证明
可以
只需对代码进行相对较小的更改即可完成。我假设读者熟悉课程材料
pyos4.py
版本作为基础-第一个
Scheduler
支持“系统调用”的版本。
提示:
可以在中找到完整的可运行示例
附录A
最后。
客观的
目标是编写以下协同程序代码:
def foo():
mytid = yield GetTid() # a "system call"
for i in xrange(3):
print "I'm foo", mytid
yield # a "trap"
…转换为本机协同程序,并仍像以前一样使用:
async def foo():
mytid = await GetTid() # a "system call"
for i in range(3):
print("I'm foo", mytid)
await ??? # a "trap" (will explain the missing bit later)
我们想在没有
asyncio
,因为我们已经有了自己的事件循环来驱动整个过程
计划程序
班
可等待的对象
本机协同程序无法立即工作,以下代码会导致错误:
async def foo():
mytid = await GetTid()
print("I'm foo", mytid)
sched = Scheduler()
sched.new(foo())
sched.mainloop()
Traceback (most recent call last):
...
mytid = await GetTid()
TypeError: object GetTid can't be used in 'await' expression
PEP 492
解释了可以等待什么类型的对象。其中一个选项是
“具有
__await__
方法返回迭代器“
.
就像
yield from
,如果你熟悉它,
await
充当等待的对象和驱动协同程序(通常是事件循环)的最外层代码之间的隧道。这最好用一个例子来说明:
class Awaitable:
def __await__(self):
value = yield 1
print("Awaitable received:", value)
value = yield 2
print("Awaitable received:", value)
value = yield 3
print("Awaitable received:", value)
return 42
async def foo():
print("foo start")
result = await Awaitable()
print("foo received result:", result)
print("foo end")
驾驶
foo()
协同程序交互生成以下内容:
>>> f_coro = foo() # calling foo() returns a coroutine object
>>> f_coro
<coroutine object foo at 0x7fa7f74046d0>
>>> f_coro.send(None)
foo start
1
>>> f_coro.send("one")
Awaitable received: one
2
>>> f_coro.send("two")
Awaitable received: two
3
>>> f_coro.send("three")
Awaitable received: three
foo received result: 42
foo end
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
StopIteration
发送到的内容
f_coro
被引导到
Awaitable
例子同样,不管怎样
Awaitable.__await__()
products被冒泡到发送值的最顶层代码。
整个过程对
f_科罗
协程,它不直接参与,也不看到值被上下传递。然而,当
等待
的迭代器已用尽
回来
值成为
等候
表达式(本例中为42)
f_科罗
最终恢复。
请注意
等候
协程中的表达式也可以链接。一个协程可以等待另一个等待另一协程的协程…直到整个链以
yield
在路上的某个地方。
将值发送到协程本身
这些知识如何帮助我们?好吧,在课程材料中,协程可以产生
SystemCall
例子调度器理解这些,并让系统调用处理请求的操作。
为了让验尸官提出
系统调用
直到调度程序
系统调用
实例可以简单地
收益率本身
,并且它将如前一节所述被引导到调度器。
因此,第一个需要的更改是将此逻辑添加到基础
系统调用
类别:
class SystemCall:
...
def __await__(self):
yield self
使用
系统调用
实例可等待,下面的程序现在实际运行:
async def foo():
mytid = await GetTid()
print("I'm foo", mytid)
>>> sched = Scheduler()
>>> sched.new(foo())
>>> sched.mainloop()
输出:
I'm foo None
Task 1 terminated
太好了,它不再崩溃了!
然而,协同程序没有收到任务ID
None
相反这是因为系统调用的
handle()
方法并由
Task.run()
方法:
# in Task.run()
self.target.send(self.sendval)
…最终在
SystemCall.__await__()
方法如果我们想将值带入协程,系统调用必须
回来
使其成为
等候
协程中的表达式。
class SystemCall:
...
def __await__(self):
return (yield self)
使用修改的运行相同的代码
系统调用
产生期望的输出:
I'm foo 1
Task 1 terminated
同时运行协同程序
我们仍然需要一种暂停协同程序的方法,即使用系统“陷阱”代码。在课程材料中,这是用素材完成的
产量
在协同程序中,但尝试使用普通
等候
实际上是一个语法错误:
async def foo():
mytid = await GetTid()
for i in range(3):
print("I'm foo", mytid)
await # SyntaxError here
幸运的是,解决方法很简单。由于我们已经有了工作的系统调用,我们可以添加一个虚拟的no-op系统调用,它的唯一任务是挂起协同程序并立即重新调度它:
class YieldControl(SystemCall):
def handle(self):
self.task.sendval = None # setting sendval is optional
self.sched.schedule(self.task)
设置
sendval
这个任务是可选的,因为这个系统调用不期望产生任何有意义的值,但我们选择将其明确化。
我们现在已经具备了运行多任务操作系统的一切条件!
async def foo():
mytid = await GetTid()
for i in range(3):
print("I'm foo", mytid)
await YieldControl()
async def bar():
mytid = await GetTid()
for i in range(5):
print("I'm bar", mytid)
await YieldControl()
sched = Scheduler()
sched.new(foo())
sched.new(bar())
sched.mainloop()
输出:
I'm foo 1
I'm bar 2
I'm foo 1
I'm bar 2
I'm foo 1
I'm bar 2
Task 1 terminated
I'm bar 2
I'm bar 2
Task 2 terminated
脚注
这个
计划程序
代码完全不变。
它只是作品
这显示了原始设计的美妙之处,其中调度器和在其中运行的任务没有相互耦合,我们能够在没有
计划程序
知道这一点。即使是
Task
包装协程的类不必更改。
不需要蹦床。
在
pyos8.py
系统的版本,一个概念
蹦床
实现。它允许协同程序在shceduler的帮助下将其工作的一部分委托给另一个协同程序(调度器代表父协同程序调用子协同程序,并将前者的结果发送给父协同程序)。
不需要这种机制,因为
等候
(和它的老同伴,
收益率
)正如开头所解释的那样,已经使这种链接成为可能。
附录A-一个完全可运行的示例(需要Python 3.5+)
示例_完整.py
from queue import Queue
# ------------------------------------------------------------
# === Tasks ===
# ------------------------------------------------------------
class Task:
taskid = 0
def __init__(self,target):
Task.taskid += 1
self.tid = Task.taskid # Task ID
self.target = target # Target coroutine
self.sendval = None # Value to send
# Run a task until it hits the next yield statement
def run(self):
return self.target.send(self.sendval)
# ------------------------------------------------------------
# === Scheduler ===
# ------------------------------------------------------------
class Scheduler:
def __init__(self):
self.ready = Queue()
self.taskmap = {}
def new(self,target):
newtask = Task(target)
self.taskmap[newtask.tid] = newtask
self.schedule(newtask)
return newtask.tid
def exit(self,task):
print("Task %d terminated" % task.tid)
del self.taskmap[task.tid]
def schedule(self,task):
self.ready.put(task)
def mainloop(self):
while self.taskmap:
task = self.ready.get()
try:
result = task.run()
if isinstance(result,SystemCall):
result.task = task
result.sched = self
result.handle()
continue
except StopIteration:
self.exit(task)
continue
self.schedule(task)
# ------------------------------------------------------------
# === System Calls ===
# ------------------------------------------------------------
class SystemCall:
def handle(self):
pass
def __await__(self):
return (yield self)
# Return a task's ID number
class GetTid(SystemCall):
def handle(self):
self.task.sendval = self.task.tid
self.sched.schedule(self.task)
class YieldControl(SystemCall):
def handle(self):
self.task.sendval = None # setting sendval is optional
self.sched.schedule(self.task)
# ------------------------------------------------------------
# === Example ===
# ------------------------------------------------------------
if __name__ == '__main__':
async def foo():
mytid = await GetTid()
for i in range(3):
print("I'm foo", mytid)
await YieldControl()
async def bar():
mytid = await GetTid()
for i in range(5):
print("I'm bar", mytid)
await YieldControl()
sched = Scheduler()
sched.new(foo())
sched.new(bar())
sched.mainloop()
