协程

1
2
3
4
5

>>> my_coro = simple_coroutine()
>>> my_coro.send(1729)
Traceback (most recent call last):
    File "<stdin>", line 1, in <module>
    TypeError: can not send non-None value to a just-started generator

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from functools import wraps

def coroutine(func):
    """装饰器：向前执行到第一个`yield`表达式，预激`func`"""
    @wraps(func)
    def primer(*args,**kwargs): ➊
        gen = func(*args,**kwargs) ➋
        next(gen) ➌
        return gen ➍
    return primer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

>>> def simple_coro2(a):
       print('-> Started: a =', a)
       b = yield a
       print('-> Received: b =', b)
       c = yield a + b
       print('-> Received: c =', c)

>>> my_coro2 = simple_coro2(14)
>>> from inspect import getgeneratorstate
>>> getgeneratorstate(my_coro2)
'GEN_CREATED'
>>> next(my_coro2)  # 预激(prime)协程
  Started: a = 14  # 协程返回 a 的值 14
14
>>> getgeneratorstate(my_coro2)
'GEN_SUSPENDED'
>>> my_coro2.send(28) # b 被赋值 28，协程返回 a+b 的值
   Received: b = 28   # 协程暂定在 c = yeild a + b 等号的左边，等待调用者发送值
42

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

class DemoException(Exception):
"""为这次演示定义的异常类型。"""

def demo_exc_handling():
    print('-> coroutine started')
    try:
        while True:
            try:
                x = yield
            except DemoException: ➊
                print('*** DemoException handled. Continuing...')
            else: ➋
                print('-> coroutine received: {!r}'.format(x))
        # 最后一行代码不会执行，因为只有未处理的异常才会中止无限循环，
        # 而一旦出现未处理的异常，协程会立即终止
        raise RuntimeError('This line should never run.') ➌
    finally:
        print('-> coroutine ending')

>>> exc_coro = demo_exc_handling()
>>> next(exc_coro)
coroutine started
>>> exc_coro.send(11)
coroutine received: 11
>>> exc_coro.close()  # 协程关闭
>>> from inspect import getgeneratorstate
>>> getgeneratorstate(exc_coro)
'GEN_CLOSED'

>>> exc_coro = demo_exc_handling()
>>> next(exc_coro)
coroutine started
>>> exc_coro.throw(DemoException)  # 异常管理 - 异常在协程内得到处理
DemoException handled. Continuing
>>> getgeneratorstate(exc_coro)
'GEN_SUSPENDED'

>>> exc_coro = demo_exc_handling()
>>> next(exc_coro)
coroutine started
>>> exc_coro.throw(ZeroDivisionError) # 异常管理 - 异常在协程内未处理
Traceback (most recent call last):
ZeroDivisionError
>>> getgeneratorstate(exc_coro)
'GEN_CLOSED'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from collections import namedtuple
Result = namedtuple('Result', 'count average')
def averager():
    total = 0.0
    count = 0
    average = None
    while True:
        term = yield
        if term is None:
            break       #  为了返回值，协程必须正常终止
        total += term
        count += 1
        average = total/count  # Python 3.3 之前，如果生成器返回值，解释器会报句法错误
    return Result(count, average)  

>> coro_avg = averager()
>>> next(coro_avg)
>>> coro_avg.send(10)
>>> coro_avg.send(30)
>>> coro_avg.send(6.5)
>>> try:
...    coro_avg.send(None) #  发送 None 会终止循环，导致协程结束，返回结果
... except StopIteration as exc:
...    result = exc.value # 生成器对象会抛出StopIteration 异常
...                       # 异常对象的 value 属性保存着返回的值
>>> result
Result(count=3, average=15.5)

1
2
3
4

def gen():
    ....
    yield from  x
    ....

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from collections import Iterable

def flatten(items, ignore_types=(str, bytes)):
    for x in items:
        if isinstance(x, Iterable) and not isinstance(x, ignore_types):
            yield from flatten(x)
        else:
            yield x    # 只需实现 __next__ 方法的简单迭代器

items = [1, 2, [3, 4, [5, 6], 7], 8]
# Produces 1 2 3 4 5 6 7 8
for x in flatten(items):
    print(x)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 子生成器
def averager():
  # 同上
  ....
  return Result(count, average)  # return 值会成为 yeild from 表达式的值

# 委派生成器
def grouper(results, key):
  while True:  # 每次迭代时会新建一个 averager 实例；每个实例都是作为协程使用的生成器对象
      results[key] = yield from averager() # grouper 接受值，通过管道传递给 averager 实例

# 客户端代码，即调用方
def main(data):
  results = {}
  for key, values in data.items():
      group = grouper(results, key) #  group 是调用 grouper 函数得到的委派生成器对象
      next(group)    # 预激委派生成器 group，此时进入 while True 循环，调用子生成
                      # 器 averager 后，在 yield from 表达式处暂
      for value in values:
          group.send(value)
      group.send(None) # 重要！
  report(results)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# RESULT = yield from EXPR 伪代码
_i = iter(EXPR)  # <1>
try:
    _y = next(_i)  # <2>
except StopIteration as _e:
    _r = _e.value  # <3>
else:
    while 1:  # <4>
        try:
            _s = yield _y  # <5>
        except GeneratorExit as _e:  # <6>
            try:
                _m = _i.close
            except AttributeError:
                pass
            else:
                _m()
            raise _e
        except BaseException as _e:  # <7>
            _x = sys.exc_info()
            try:
                _m = _i.throw
            except AttributeError:
                raise _e
            else:  # <8>
                try:
                    _y = _m(*_x)
                except StopIteration as _e:
                    _r = _e.value
                    break
        else:  # <9>
            try:  # <10>
                if _s is None:  # <11>
                    _y = next(_i)
                else:
                    _y = _i.send(_s)
            except StopIteration as _e:  # <12>
                _r = _e.value
                break

RESULT = _r  # <13>
# END YIELD_FROM_EXPANSION

1
2
3
4
5
6
7

def taxi_process(ident, trips, start_time=0): ➊
    """每次改变状态时创建事件，把控制权让给仿真器"""
    time = yield Event(start_time, ident, 'leave garage') ➋
    for i in range(trips): ➌
        time = yield Event(time, ident, 'pick up passenger') ➍
        time = yield Event(time, ident, 'drop off passenger') ➎
    yield Event(time, ident, 'going home')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class Simulator:
    def __init__(self, procs_map):
        self.events = queue.PriorityQueue()
        self.procs = dict(procs_map)

    def run(self, end_time): ➊
        """排定并显示事件，直到时间结束"""
        # 排定各辆出租车的第一个事件
        for _, proc in sorted(self.procs.items()): ➋
            first_event = next(proc) ➌
            self.events.put(first_event)
            # 这个仿真系统的主循环
        sim_time = 0 ➎
        while sim_time < end_time: ➏
            if self.events.empty(): ➐
                print('*** end of events ***')
                break
            current_event = self.events.get() ➑
            sim_time, proc_id, previous_action = current_event ➒
            print('taxi:', proc_id, proc_id * ' ', current_event) ➓
            active_proc = self.procs[proc_id]
            next_time = sim_time + compute_duration(previous_action)
            try:
                next_event = active_proc.send(next_time)
            except StopIteration:
                del self.procs[proc_id]
            else:
                self.events.put(next_event)
        else:
            msg = '*** end of simulation time: {} events pending ***'
            print(msg.format(self.events.qsize()))

> taxis = {i: taxi_process(i, (i + 1) * 2, i * DEPARTURE_INTERVAL)
           for i in range(num_taxis)}
> sim = Simulator(taxis)