协程和并发课程【部分五~七】

此文紧接上文协程和并发课程【部分三~四】。

本文的难度会比较大，主要是第七部分，用协程来写一个操作系统。

第五部分：协程与任务（Task）

注：这里的任务与操作系统的任务概念一致。

对于任务，它有几个核心特性：

  • 独立控制流
  • 内部状态
  • 可以调度（挂起和恢复）
  • 能够与其他任务通信

作者认为协程看起来就像任务（Coroutines look like tasks）。为此，作者针对任务的这些特性一一提出论证：

  1. 协程有自己的控制流

  2. 协程有自己的内部状态（变量），而且内部变量的生存时间跟协程一样，它们构成了执行环境，如

  3. 协程之间可以通信，seng() 方法可以向协程发送数据，而 yield 表达式可以接受输入

  4. 协程可以挂起和恢复：yield 挂起执行，send() 恢复执行，close() 结束执行。

不过有一点，协程不像任务一样可以跟线程和子进程绑定。

第六部分：操作系统

程序执行

对于一个 CPU 而言，程序是一系列的指令（如下图）；而且对于它而言，每刻最多只做一件事，也不知道任务切换的问题。

多任务问题

既然 CPU 不知道多任务的问题，应用程序也不可能知道，那只有可能是操作系统知道了。那操作系统是如何做到多任务的呢？答案就是中断（Interrupt，硬件）和陷阱（Trap，软件）。在这两种情形中，CPU 都会暂时挂起正在做的事情，而去执行操作系统的“代码”，也就是在这个时候，操作系统可能会进行任务切换。

陷阱和系统调用

底层的系统调用实际都是陷阱，它是一条特殊的指令：

每当陷阱发生，应用程序都会被挂起，操作系统接过控制权：

在陷阱发生期间，操作系统进行任务切换：

当任务比较多的时候，还需要进行调度，以确定下一个要运行的任务：

而对于 yield 而言，它就像是一种“陷阱”，当生成器函数执行到 yield 的时候会被挂起，然后控制权转移回调用者。如果我们把 yield 当做陷阱，那我们就可以用 Python 建立一个多任务的操作系统。

第七部分：创建操作系统

我们要创建的操作系统：

  • 支持多任务
  • 使用纯 Python 代码
  • 没有线程
  • 没有子进程
  • 使用生成器/协程

第 1 步：定义任务

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
46
# ------------------------------------------------------------
# pyos1.py  -  The Python Operating System
# 
# Step 1: Tasks
# ------------------------------------------------------------

# This object encapsulates a running task.


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

    # Run a task until it hits the next yield statement
    def run(self):
        return self.target.send(self.sendval)


# ------------------------------------------------------------
#                       == Example ==
# ------------------------------------------------------------
if __name__ == '__main__':
    # A simple generator/coroutine function
    def foo():
        print "Part 1"
        yield
        print "Part 2"
        yield


    t1 = Task(foo())
    print "Running foo()"
    t1.run()
    print "Resuming foo()"
    t1.run()

    # If you call t1.run() one more time, you get StopIteration.
    # Uncomment the next statement to see that.

    # t1.run()

第 2 步：定义调度器

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# ------------------------------------------------------------
# pyos2.py  -  The Python Operating System
#
# Step 2: A Scheduler
# ------------------------------------------------------------


# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
......


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


class Scheduler(object):
    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 schedule(self, task):
        self.ready.put(task)

    def mainloop(self):
        while self.taskmap:
            task = self.ready.get()
            result = task.run()
            self.schedule(task)


# ------------------------------------------------------------
#                      === Example ===
# ------------------------------------------------------------
if __name__ == '__main__':

    # Two tasks
    def foo():
        while True:
            print "I'm foo"
            yield


    def bar():
        while True:
            print "I'm bar"
            yield

            # Run them


    sched = Scheduler()
    sched.new(foo())
    sched.new(bar())
    sched.mainloop()

第 3 步：定义任务出口（Exit）

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
# ------------------------------------------------------------
# pyos3.py  -  The Python Operating System
#
# Step 3: Added handling for task termination
# ------------------------------------------------------------

# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
......


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


class Scheduler(object):
    ......

    def exit(self, task):
        print "Task %d terminated" % task.tid
        del self.taskmap[task.tid]

    ......

    def mainloop(self):
        while self.taskmap:
            task = self.ready.get()
            try:
                result = task.run()
            except StopIteration:
                self.exit(task)
                continue
            self.schedule(task)


# ------------------------------------------------------------
#                      === Example ===
# ------------------------------------------------------------
if __name__ == '__main__':
    def foo():
        for i in xrange(10):
            print "I'm foo"
            yield


    def bar():
        for i in xrange(5):
            print "I'm bar"
            yield


    sched = Scheduler()
    sched.new(foo())
    sched.new(bar())
    sched.mainloop()

# Results:
# I'm foo
# I'm bar
# I'm foo
# I'm bar
# I'm foo
# I'm bar
# I'm foo
# I'm bar
# I'm foo
# I'm bar
# I'm foo
# Task 2 terminated
# I'm foo
# I'm foo
# I'm foo
# I'm foo
# Task 1 terminated

第 4 步：系统调用

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
# ------------------------------------------------------------
# pyos4.py  -  The Python Operating System
#
# Step 4: Introduce the idea of a "System Call"
# ------------------------------------------------------------

# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
......


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


class Scheduler(object):
    ......

    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(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)


# ------------------------------------------------------------
#                      === Example ===
# ------------------------------------------------------------
if __name__ == '__main__':

    def foo():
        mytid = yield GetTid()
        for i in xrange(5):
            print "I'm foo", mytid
            yield


    def bar():
        mytid = yield GetTid()
        for i in xrange(10):
            print "I'm bar", mytid
            yield


    sched = Scheduler()
    sched.new(foo())
    sched.new(bar())
    sched.mainloop()

# Results:
# I'm foo 1
# I'm bar 2
# I'm foo 1
# I'm bar 2
# 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
# I'm bar 2
# I'm bar 2
# I'm bar 2
# Task 2 terminated

第 5 步：任务管理

对于任务，它：

  • 不能看到调度器
  • 不能看见其他任务
  • yield 是唯一的对外接口（在任务内）

一些常见的任务管理函数：

  • 创建新任务
  • 杀死任务
  • 等待任务

先来看下包括创建和杀死任务的样例：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
# ------------------------------------------------------------
# pyos5.py  -  The Python Operating System
#
# Step 5: Added system calls for simple task management
# ------------------------------------------------------------

# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
......


# ------------------------------------------------------------
#                      === Scheduler ===
# ------------------------------------------------------------
......


# ------------------------------------------------------------
#                   === System Calls ===
# ------------------------------------------------------------

......


# 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)


# ------------------------------------------------------------
#                      === Example ===
# ------------------------------------------------------------
if __name__ == '__main__':
    def foo():
        mytid = yield GetTid()
        while True:
            print "I'm foo", mytid
            yield


    def main():
        child = yield NewTask(foo())  # Launch new task
        for i in xrange(5):
            yield
        yield KillTask(child)  # Kill the task
        print "main done"


    sched = Scheduler()
    sched.new(main())
    sched.mainloop()

# Results:
# I'm foo 2
# I'm foo 2
# I'm foo 2
# I'm foo 2
# I'm foo 2
# Task 2 terminated
# main done
# Task 1 terminated

添加等待子进程退出任务管理：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
# ------------------------------------------------------------
# pyos6.py  -  The Python Operating System
#
# Added support for task waiting
# ------------------------------------------------------------

# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
......


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


class Scheduler(object):
    def __init__(self):
        ......

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

    ......

    def exit(self, task):
        ......
        # 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

    ......


# ------------------------------------------------------------
#                   === System Calls ===
# ------------------------------------------------------------
......


# 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)


# ------------------------------------------------------------
#                      === Example ===
# ------------------------------------------------------------
if __name__ == '__main__':
    def foo():
        for i in xrange(5):
            print "I'm foo"
            yield


    def main():
        child = yield NewTask(foo())
        print "Waiting for child"
        yield WaitTask(child)
        print "Child done"


    sched = Scheduler()
    sched.new(main())
    sched.mainloop()

# Results:
# I'm foo
# Waiting for child
# I'm foo
# I'm foo
# I'm foo
# I'm foo
# Task 2 terminated
# Child done
# Task 1 terminated

回显服务器

很神奇吧，我们还能基于刚新建的操作系统来创建一个回显服务器（Echo Server）。Generator and Coroutine Rock!!!

这个回显服务器是阻塞的，但没关系，测试嘛：

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
# echobad.py
#
# A flawed implementation of an concurrent echo server using our task scheduler.
# If you run this, you will notice that it freezes up right away.  This
# is because the accept() operation in the server is blocking.  Since we're
# not using threads, this blocks the operation of everything!

from pyos6 import *
from socket import *


def handle_client(client, addr):
    print "Connection from", addr
    while True:
        data = client.recv(65536)
        if not data:
            break
        client.send(data)
    client.close()
    print "Client closed"
    yield  # Make the function a generator/coroutine


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


def alive():
    while True:
        print "I'm alive!"
        yield


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

作者没有提供客户端，那我们就自己写一个咯：

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

import socket

if "__main__" == __name__:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("localhost", 45000))
    sock.send("Echo test.")

    szBuf = sock.recv(1024)
    print "Received ", szBuf
    sock.close()
    print "Connection closed."

测试结果如下：

第 6 步：I/O 等待

因为上边的服务器是阻塞（socket.accept）的，这样整个 Python 解释器都会被阻塞掉，所以我们要继续改进操作系统为真正的多任务系统，让其支持 I/O 等待。

改进操作系统

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
# ------------------------------------------------------------
# pyos7.py  -  The Python Operating System
#
# Step 6 : I/O Waiting Support added
# ------------------------------------------------------------

# ------------------------------------------------------------
#                       === Tasks ===
# ------------------------------------------------------------
......


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


class Scheduler(object):
    def __init__(self):
        ......

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

    ......

    # 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 mainloop(self):
        self.new(self.iotask())
        ......


# ------------------------------------------------------------
#                   === System Calls ===
# ------------------------------------------------------------
......


# 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)

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

# Run the script echogood.py to see this work

改进回显服务器

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
# echogood2.py
#
# A working concurrent echo server

from socket import *
from pyos7 import *


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


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


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

测试效果如下：

总结

到目前为止，我们已经，

  • 创建了一个多任务操作系统
  • 支持并行运行任务
  • 支持创建、销毁、等待任务
  • 支持任务的 I/O 操作
  • 甚至可以写一个并发服务器

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

谢宏峰，毕业于暨南大学（2009 - 2013）、中国科学院（2013 - 2016），现就职于深信服科技，从事 OpenStack 开发。