协程和并发课程【部分八~九】

2017-05-01

第八部分：栈引起的问题

Trampolining

当程序涉及协程时，我们不能调用使用了 yield （可以挂起）的子函数。yield 语句只能在最上层挂起协程。例如在以下例子中，foo() 函数并不会因为调用了可以被挂起的 bar() 函数就能够被挂起：

def bar():
    yield

def foo():
    bar()

去除这个限制的方法之一就是无栈 Python。具体的实现技巧叫 Trampolining，中文译名跳床。这中文译名怪怪的，我们还是叫 Tramplining 吧。例如，下边就是利用了该技巧的简单例子：

# trampoline.py
#
# A simple of example of trampoling between coroutines

# A subroutine
def add(x, y):
    yield x + y


# A function that calls a subroutine
def main():
    r = yield add(2, 2)
    print r
    yield


def run():
    m = main()
    # An example of a "trampoline"
    sub = m.send(None)
    result = sub.send(None)
    m.send(result)


run()

作者对此还用流程图进行了解释：

改进操作系统

# ------------------------------------------------------------
# pyos8.py  -  The Python Operating System
#
# Step 7 : Support for coroutine trampolines (subroutines)
# ------------------------------------------------------------

import types


# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
class Task(object):
    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
        self.stack = []  # Call stack

    # Run a task until it hits the next yield statement
    def run(self):
        while True:
            try:
                result = self.target.send(self.sendval)
                if isinstance(result, SystemCall):
                    return result
                if isinstance(result, types.GeneratorType):
                    self.stack.append(self.target)
                    self.sendval = None
                    self.target = result
                else:
                    if not self.stack:
                        return
                    self.sendval = result
                    self.target = self.stack.pop()
            except StopIteration:
                if not self.stack: raise
                self.sendval = None
                self.target = self.stack.pop()


# ------------------------------------------------------------
#                      === Scheduler ===
# ------------------------------------------------------------
from Queue import Queue
import select


class Scheduler(object):
    def __init__(self):
        self.ready = Queue()
        self.taskmap = {}

        # Tasks waiting for other tasks to exit
        self.exit_waiting = {}

        # I/O waiting
        self.read_waiting = {}
        self.write_waiting = {}

    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]
        # Notify other tasks waiting for exit
        for task in self.exit_waiting.pop(task.tid, []):
            self.schedule(task)

    def waitforexit(self, task, waittid):
        if waittid in self.taskmap:
            self.exit_waiting.setdefault(waittid, []).append(task)
            return True
        else:
            return False

    # I/O waiting
    def waitforread(self, task, fd):
        self.read_waiting[fd] = task

    def waitforwrite(self, task, fd):
        self.write_waiting[fd] = task

    def iopoll(self, timeout):
        if self.read_waiting or self.write_waiting:
            r, w, e = select.select(self.read_waiting,
                                    self.write_waiting, [], timeout)
            for fd in r:
                self.schedule(self.read_waiting.pop(fd))
            for fd in w:
                self.schedule(self.write_waiting.pop(fd))

    def iotask(self):
        while True:
            if self.ready.empty():
                self.iopoll(None)
            else:
                self.iopoll(0)
            yield

    def schedule(self, task):
        self.ready.put(task)

    def mainloop(self):
        self.new(self.iotask())
        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(object):
    def handle(self):
        pass


# Return a task's ID number
class GetTid(SystemCall):
    def handle(self):
        self.task.sendval = self.task.tid
        self.sched.schedule(self.task)


# Create a new task
class NewTask(SystemCall):
    def __init__(self, target):
        self.target = target

    def handle(self):
        tid = self.sched.new(self.target)
        self.task.sendval = tid
        self.sched.schedule(self.task)


# Kill a task
class KillTask(SystemCall):
    def __init__(self, tid):
        self.tid = tid

    def handle(self):
        task = self.sched.taskmap.get(self.tid, None)
        if task:
            task.target.close()
            self.task.sendval = True
        else:
            self.task.sendval = False
        self.sched.schedule(self.task)


# Wait for a task to exit
class WaitTask(SystemCall):
    def __init__(self, tid):
        self.tid = tid

    def handle(self):
        result = self.sched.waitforexit(self.task, self.tid)
        self.task.sendval = result
        # If waiting for a non-existent task,
        # return immediately without waiting
        if not result:
            self.sched.schedule(self.task)


# Wait for reading
class ReadWait(SystemCall):
    def __init__(self, f):
        self.f = f

    def handle(self):
        fd = self.f.fileno()
        self.sched.waitforread(self.task, fd)


# Wait for writing
class WriteWait(SystemCall):
    def __init__(self, f):
        self.f = f

    def handle(self):
        fd = self.f.fileno()
        self.sched.waitforwrite(self.task, fd)


# ------------------------------------------------------------
#                      === Library Functions ===
# ------------------------------------------------------------

def Accept(sock):
    yield ReadWait(sock)
    yield sock.accept()


def Send(sock, buffer):
    while buffer:
        yield WriteWait(sock)
        len = sock.send(buffer)
        buffer = buffer[len:]


def Recv(sock, maxbytes):
    yield ReadWait(sock)
    yield sock.recv(maxbytes)

# ------------------------------------------------------------
#                      === Example ===
# ------------------------------------------------------------

# Look at the echoserver.py script for an example

作者解释如下：

改进回显服务器

现在所有的 I/O 操作都是都是子协程了：

# echoserver.py
#
# A concurrent echo server using coroutines

from pyos8 import *
from socket import *


def handle_client(client, addr):
    print "Connection from", addr
    while True:
        data = yield Recv(client, 65536)
        if not data:
            break
        yield Send(client, data)
    print "Client closed"
    client.close()


def server(port):
    print "Server starting"
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(("", port))
    sock.listen(5)
    while True:
        client, addr = yield Accept(sock)
        yield NewTask(handle_client(client, addr))


sched = Scheduler()
sched.new(server(45000))
sched.mainloop()

非阻塞 I/O

除了上边的做法，我们还可以用非阻塞 I/O 重新封装原生 socket（这也是一些协程库的做法，如 Greenlet）：

# sockwrap.py
#
# Coroutine wrapper around a socket object

from pyos8 import *


class Socket(object):
    def __init__(self, sock):
        self.sock = sock

    def accept(self):
        yield ReadWait(self.sock)
        client, addr = self.sock.accept()
        yield Socket(client), addr

    def send(self, buffer):
        while buffer:
            yield WriteWait(self.sock)
            len = self.sock.send(buffer)
            buffer = buffer[len:]

    def recv(self, maxbytes):
        yield ReadWait(self.sock)
        yield self.sock.recv(maxbytes)

    def close(self):
        yield self.sock.close()

这样，回显服务器就可以这样子写了：

# echoserver2.py
#
# A concurrent echo server using coroutines

from pyos8 import *
from socket import *
from sockwrap import Socket


def handle_client(client, addr):
    print "Connection from", addr
    while True:
        data = yield client.recv(65536)
        if not data:
            break
        yield client.send(data)
    print "Client closed"
    yield client.close()


def server(port):
    print "Server starting"
    rawsock = socket(AF_INET, SOCK_STREAM)
    rawsock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    rawsock.bind(("", port))
    rawsock.listen(1024)

    sock = Socket(rawsock)
    while True:
        client, addr = yield sock.accept()
        yield NewTask(handle_client(client, addr))


sched = Scheduler()
sched.new(server(45000))
sched.mainloop()

基于 Twisted 回显服务器

下例取自 Twisted 官方例子：

#!/usr/bin/python
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
# See LICENSE for details.

#

from twisted.internet.protocol import Protocol, Factory
from twisted.internet import reactor


### Protocol Implementation

# This is just about the simplest possible protocol
class Echo(Protocol):
    def connectionMade(self):
        print "Got a connection"

    def dataReceived(self, data):
        """As soon as any data is received, write it back."""
        self.transport.write(data)


def main():
    f = Factory()
    f.protocol = Echo
    reactor.listenTCP(45000, f)
    reactor.run()


if __name__ == '__main__':
    main()

第 9 部分：终言

未来主题

在我们的任务调度器上，我们还可以继续探索：

任务间通信
对阻塞操作处理（如数据库访问）
协程多任务和线程
错误处理

生成器

生成器其实远比大家认识的要强大：

自定义迭代模式
处理流水线和数据流
事件处理
多任务协作

yield 使用场景

yield 常用使用场景：

迭代（数据生产者）
接收数据（消费者）
陷阱（多任务协作）

终于结束了….