协程与多任务调度

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

任务调度器是系统最核心的功能，相当于Linux中的init程序，用来管理所有的系统任务。其它任务通过注册到任务调度器来实现其功能:

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是如此好用，以至于很多语言都逐渐加入对其的支持，如PHP5.5开始加入yield,javascript 6(ECMAScript 6)也加入了对其的支持，虽然其使用起来有一些区别，但是原理是相通的，深入理解协程和yield，对于理解任务调度，系统原理意义重大。

文件下载

coroutine.py

参考资料

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