协程与多任务调度

在计算机科学中，多任务（multitasking）是指在同一个时间段内运行多个任务。现代计算机作为一个复杂的系统，运行的任务往往不止一个，所以多任务调度对于计算机来说尤为重要。

现阶段，多任务调度主要分为抢占式多任务和协作式多任务。抢占式多任务由操作系统决定进程的调度方案，而协作式多任务则由当前任务主动放弃执行权，下一个任务继续运行。

由于协作式任务管理更容易受到程序本身的影响，现阶段几乎所有通用操作系统都采用抢占式多任务管理。

现阶段，主要通过多进程或多线程的方式来实现多任务：

#include <stdio.h>
#include <unistd.h>

int main()
{
    pid_t pid;
    pid = fork();

    if(pid < 0){
        printf("Fork Error!\n");
    }else if(pid > 0){
        printf("This is the parent Process! Process Id is %d, Child id is %d\n",getpid(),pid);

        int i = 0;
        while(i < 10){
            printf("This is parent Process output of i %d!\n",i);
            i++;
        }
    }else if(pid == 0){
        printf("This is the child Process! Process Id is %d, parent id is %d\n",getpid(),getppid());

        int j = 0;
        while(j < 10){
            printf("This is child Process output of j %d\n",j);
            j++;
        }
    }

    return 0;
}

在《协程与yield》中，我们说到了协程是一种比进程和线程更加轻量级的解决方案，也通过 yield 实现了协程。

但当时最大的疑问是，我们并没有提供像进程或线程一样的任务调度，因此还没有真正体现出协程的优势。下面我们来实现一个简单的协程和协作式多任务调度。

首先，我们需要对任务（Task）进行包装：

class Task():
    def __init__(self,taskid,coroutine):
        self.__taskId = taskid
        self.__coroutine = coroutine
        self.__sendValue = ''
        self.__beforeFirstYield = True
        self.isFinished = False

    def getTaskId(self):
        return self.__taskId

    def setValue(self,value):
        self.__sendValue = value

    def run(self):
        if(self.__beforeFirstYield):
            self.__beforeFirstYield = False
            return self.__coroutine.next()
        else:
            try:
                retval = self.__coroutine.send(self.__sendValue)
                return retval
            except StopIteration:
                self.isFinished = True
                return ""

这里的“任务”类似系统中的进程，有自己的 ID，也有向任务传递数据的 sendValue。

接下来需要一个任务调度器，专门用来管理这些任务：

from Queue import Queue

class Scheduler():
    def __init__(self):
        self.taskQueue = Queue()
        self.maxTaskId = 0
        self.taskMap = dict()

    def scheduler(self,task):
        self.taskQueue.put(task)

    def newTask(self,coroutine):
        self.maxTaskId += 1
        task = Task(self.maxTaskId,coroutine)
        self.taskMap[self.maxTaskId] = task
        self.scheduler(task)
        return self.maxTaskId

    def KillTask(self,taskid):
        if not taskid in self.taskMap:
            return False

        i = 0
        while i < self.taskQueue.qsize():
            tmp = self.taskQueue.get()

            if tmp == self.taskMap[taskid]:
                del self.taskMap[taskid]
                break
            else:
                self.scheduler(tmp)

            i += 1

        return True

    def run(self):
        while not self.taskQueue.empty():
            task = self.taskQueue.get()
            retval = task.run()

            if task.isFinished:
                tid = task.getTaskId()
                del self.taskMap[tid]
            else:
                self.scheduler(task)

任务调度器是整个系统最核心的功能之一，负责管理所有任务。其它任务通过注册到任务调度器来实现自己的功能：

def task1():
    i = 0
    while i < 10:
        print "This is task 1 i is %s" % i
        i += 1
        yield

def task2():
    i = 0
    while i < 10:
        print "This is task 2 i is %s" % i
        i += 1
        yield

sch = Scheduler()
sch.newTask(task1())
sch.newTask(task2())
sch.run()

其结果输出如下，可以看出任务一和任务二确实是交替执行，实现了任务调度的功能：

This is task 1 i is 0
This is task 2 i is 0
This is task 1 i is 1
This is task 2 i is 1
This is task 1 i is 2
This is task 2 i is 2
This is task 1 i is 3
This is task 2 i is 3
This is task 1 i is 4
This is task 2 i is 4
This is task 1 i is 5
This is task 2 i is 5
This is task 1 i is 6
This is task 2 i is 6
This is task 1 i is 7
This is task 2 i is 7
This is task 1 i is 8
This is task 2 i is 8
This is task 1 i is 9
This is task 2 i is 9

上面我们实现了多个任务的调度，它们能够交替运行。这里的 yield 可以看作一种主动让出执行权的机制，一旦任务执行到 yield，调度器就可以继续运行其它任务。

然而，在上面的例子中，一旦我们把任务提交给调度器，程序本身就没有办法对任务进行更多控制，需要等到任务运行结束。

我们还需要获取任务 ID、结束任务、创建任务等控制能力，这里就需要实现任务和调度器之间的通信，也就是利用 yield 进行传值。

类似 Linux 一样，我们可以给任务提供一些函数接口。任务通过 yield 把需要调用的函数传给调度器，调度器再把结果返回给任务，例如：

def task3():
    pid = yield getpid()
    print "This taskid is %d" % pid

    i = 0
    while i < 10:
        print "This is task 3 i is %d" % i
        yield

要实现上面的调用，可以添加一个系统调用类：

class SysCall():
    def __init__(self,callback):
        self.__callback = callback

    def __call__(self,task,schedular):
        if not isinstance(task,Task):
            raise TypeError(task.__name__+" is not instance of Task")

        self.__callback(task,schedular)

然后对 Scheduler 类的 run 方法作出修改：

def run(self):
    while not self.taskQueue.empty():
        task = self.taskQueue.get()
        retval = task.run()

        if isinstance(retval,SysCall):
            retval(task,self)
            continue

        if task.isFinished:
            tid = task.getTaskId()
            del self.taskMap[tid]
        else:
            self.scheduler(task)

然后添加供任务使用的接口函数：

def getpid():
    def tmp(task,schedular):
        task.setValue(task.getTaskId())
        schedular.scheduler(task)

    return SysCall(tmp)

def Killpid():
    def tmp(task,scheduler):
        task.setValue(scheduler.KillTask(taskid))

    return SysCall(tmp)

def fork():
    pass

这里实现 SysCall 的主要目的是方便调度器对传递过来的函数类型进行控制。为了系统安全考虑，可以限制任务只能提交允许的系统调用，不属于 SysCall 的函数不进行执行。

至此，我们实现了一个相对完整的协程任务调度器，而不只是利用 yield 进行简单的数据传递。

yield 是如此好用，以至于很多语言都逐渐加入了对它或类似生成器机制的支持，如 PHP 5.5 开始加入 yield，JavaScript 在 ECMAScript 6 中也加入了 Generator。

虽然不同语言的实现和使用方式有所区别，但其核心思想是相通的。深入理解协程和 yield，对于理解任务调度和系统原理都有一定帮助。

文件下载

coroutine.py

参考资料

在PHP中使用协程实现多任务调度