Python高级编程之生成器(Generator)与coroutine(四):一个简单的多任务系统

  1 #-*-coding:utf-8 -*-
  2 ‘‘‘
  3 用Python和coroutine实现一个简单的多任务系统
  4 ‘‘‘
  5 # ##Step 1:Define Tasks###################################
  6 import select
  7 class Task(object):
  8     taskid = 0
  9 
 10     def __init__(self,target):
 11         Task.taskid += 1
 12         self.tid = Task.taskid    # Task id
 13         self.target = target      # Target coroutine
 14         self.sendval = None       # Value to send
 15 
 16     def run(self):
 17         return self.target.send(self.sendval)
 18 # ###############################################
 19 
 20 # ##Step 2:The Scheduler#########################
 21 import Queue
 22 class Scheduler(object):
 23     def __init__(self):
 24         self.ready = Queue.Queue()
 25         self.taskmap = {}
 26 
 27         # 正在等待的Tasks,key是taskid
 28         self.exit_waiting = {}
 29 
 30         # 异步IO
 31         # Holding areas for tasks blocking on I/O.These are
 32         # dictionaries mapping file descriptions to tasks
 33         # 键值为文件描述符
 34         self.read_waiting = {}
 35         self.write_waiting = {}
 36 
 37 
 38     def iotask(self):
 39         while True:
 40             if self.ready.empty():
 41                 # 如果ready为空，表示没有正在等待执行的队列
 42                 # timeout 为None,表示不关心任何文件描述符的变化
 43                 self.iopool(None)
 44             else:
 45                 # ready不为空，则设置select函数不管文件描述符是否发生变化都立即返回
 46                 self.iopool(0)
 47             yield
 48 
 49 
 50     def new(self,target):
 51         newtask = Task(target)
 52         self.taskmap[newtask.tid] = newtask
 53         self.schedule(newtask)
 54         return newtask.tid
 55 
 56     def schedule(self,task):
 57         # 把task放到任务队列中去
 58         self.ready.put(task)
 59 
 60     def exit(self,task):
 61         print "Task %d terminated" %task.tid
 62         del self.taskmap[task.tid]
 63         # Notify other tasks waiting for exit
 64         # 把正在等待的任务加入到正在执行的队列中去
 65         for task in self.exit_waiting.pop(task.tid,[]):
 66             self.schedule(task)
 67 
 68     def waitforexit(self,task,waittid):
 69         ‘‘‘
 70         让一个任务等待另外一个任务，把这个任务加入到exit_waiting中去
 71         返回True表示这个task正在等待队列中
 72         ‘‘‘
 73         if waittid in self.taskmap:
 74             self.exit_waiting.setdefault(waittid,[]).append(task)
 75             return True
 76         else:
 77             return False
 78 
 79 
 80     def waitforread(self,task,fd):
 81         ‘‘‘
 82         functions that simply put a task into to
 83         one of the above dictionaries
 84         ‘‘‘
 85         self.read_waiting[fd] = task
 86 
 87     def waitforwrite(self,task,fd):
 88         self.write_waiting[fd] = task
 89 
 90     def iopool(self,timeout):
 91         ‘‘‘
 92         I/O Polling.Use select() to determine which file
 93         descriptors can be used.Unblock any associated task
 94         ‘‘‘
 95         if self.read_waiting or self.write_waiting:
 96             # 获取I/O事件,一旦获取到，就放入到执行队列中取，等待执行
 97             r,w,e = select.select(self.read_waiting,
 98                                   self.write_waiting,[],timeout)
 99             for fd in r:
100                 self.schedule(self.read_waiting.pop(fd))
101 
102             for fd in w:
103                 self.schedule(self.write_waiting.pop(fd))
104 
105     def mainloop(self):
106         self.new(self.iotask())  # Launch I/O polls
107         while self.taskmap:
108             task = self.ready.get()
109             try:
110                 result = task.run()
111                 # 如果task执行的是System call，则对当前环境进行保存
112                 # 然后在执行System Call
113                 if isinstance(result,SystemCall):
114                     # 把当前的环境保存，即保存当前运行的task和sched
115                     result.task = task
116                     result.sched = self
117                     result.handle()
118                     continue
119             except StopIteration:
120                 self.exit(task)
121                 # print("task is over")
122                 continue
123             self.schedule(task)
124 # ##Step 2:The Scheduler#########################
125 
126 
127 # ##SystemCall#########################
128 class SystemCall(object):
129     ‘‘‘
130     所有系统调用的基类，继承自该类的类要重写handle函数
131     ‘‘‘
132     def handle(self):
133         pass
134 
135 
136 class GetTid(SystemCall):
137     ‘‘‘
138     获取任务ID
139     ‘‘‘
140     def handle(self):
141         self.task.sendval = self.task.tid
142         self.sched.schedule(self.task)
143 
144 
145 class NewTask(SystemCall):
146     ‘‘‘
147     新建一个Task
148     ‘‘‘
149     def __init__(self,target):
150         self.target = target
151 
152     def handle(self):
153         # 在这里把target封装成Task
154         # 是在这里把新生成的task加入到执行队列当中去
155         tid = self.sched.new(self.target)
156         self.task.sendval = tid
157         # 把执行这个系统调用的父task重新加入到执行队列中去
158         # 这一点很关键，因为判断一个task是否结束是通过taskmap的
159         # 这个task只是暂时被挂起,要重新放到queue中去
160         self.sched.schedule(self.task)
161 
162 class KillTask(SystemCall):
163     ‘‘‘
164     杀死一个Task
165     ‘‘‘
166     def __init__(self,tid):
167         self.tid = tid
168 
169     def handle(self):
170         task = self.sched.taskmap.get(self.tid,None)
171         # task指的是要被kill掉的那个task
172         # self.task指的是发起KillTask这个系统调用task
173         if task:
174             task.target.close()
175             self.task.sendval = None
176         else:
177             self.task.sendval = False
178         # target.close()只是产生一个StopIteration异常
179         self.sched.schedule(self.task)
180 
181 
182 class WaitTask(SystemCall):
183     ‘‘‘
184     让任务进行等待 系统调用
185     ‘‘‘
186     def __init__(self,tid):
187         self.tid = tid
188 
189     def handle(self):
190         result = self.sched.waitforexit(self.task,self.tid)
191         self.task.sendval = result
192         # 如果等待的是一个不存在的task，则立即返回
193         if not  result:
194             self.sched.schedule(self.task)
195 
196 
197 
198 
199 class ReadWait(SystemCall):
200     ‘‘‘
201     异步读 系统调用
202     ‘‘‘
203     def __init__(self,f):
204         self.f = f
205 
206     def handle(self):
207         fd = self.f.fileno()
208         self.sched.waitforread(self.task,fd)
209 
210 class WriteWait(SystemCall):
211     ‘‘‘
212     异步写 系统调用
213     ‘‘‘
214     def _init__(self,f):
215         self.f = f
216 
217     def handle(self):
218         fd = self.f.fileno()
219         self.sched.waitforwrite(self.task,fd)