Python 使用生成器代替线程的方法

# two simple generator functions
def countdown(n):
  while n > 0:
    print('t-minus', n)
    yield
    n -= 1
  print('blastoff!')

def countup(n):
  x = 0
  while x < n:
    print('counting up', x)
    yield
    x += 1

from collections import deque

class taskscheduler:
  def __init__(self):
    self._task_queue = deque()

  def new_task(self, task):
    '''
    admit a newly started task to the scheduler
    '''
    self._task_queue.append(task)

  def run(self):
    '''
    run until there are no more tasks
    '''
    while self._task_queue:
      task = self._task_queue.popleft()
      try:
        # run until the next yield statement
        next(task)
        self._task_queue.append(task)
      except stopiteration:
        # generator is no longer executing
        pass

# example use
sched = taskscheduler()
sched.new_task(countdown(10))
sched.new_task(countdown(5))
sched.new_task(countup(15))
sched.run()

from collections import deque

class actorscheduler:
  def __init__(self):
    self._actors = {}     # mapping of names to actors
    self._msg_queue = deque()  # message queue

  def new_actor(self, name, actor):
    '''
    admit a newly started actor to the scheduler and give it a name
    '''
    self._msg_queue.append((actor,none))
    self._actors[name] = actor

  def send(self, name, msg):
    '''
    send a message to a named actor
    '''
    actor = self._actors.get(name)
    if actor:
      self._msg_queue.append((actor,msg))

  def run(self):
    '''
    run as long as there are pending messages.
    '''
    while self._msg_queue:
      actor, msg = self._msg_queue.popleft()
      try:
         actor.send(msg)
      except stopiteration:
         pass

# example use
if __name__ == '__main__':
  def printer():
    while true:
      msg = yield
      print('got:', msg)

  def counter(sched):
    while true:
      # receive the current count
      n = yield
      if n == 0:
        break
      # send to the printer task
      sched.send('printer', n)
      # send the next count to the counter task (recursive)
      sched.send('counter', n-1)

  sched = actorscheduler()
  # create the initial actors
  sched.new_actor('printer', printer())
  sched.new_actor('counter', counter(sched))

  # send an initial message to the counter to initiate
  sched.send('counter', 10000)
  sched.run()

from collections import deque
from select import select

# this class represents a generic yield event in the scheduler
class yieldevent:
  def handle_yield(self, sched, task):
    pass

  def handle_resume(self, sched, task):
    pass

# task scheduler
class scheduler:
  def __init__(self):
    self._numtasks = 0    # total num of tasks
    self._ready = deque()  # tasks ready to run
    self._read_waiting = {} # tasks waiting to read
    self._write_waiting = {} # tasks waiting to write

  # poll for i/o events and restart waiting tasks
  def _iopoll(self):
    rset,wset,eset = select(self._read_waiting,
                self._write_waiting,[])
    for r in rset:
      evt, task = self._read_waiting.pop(r)
      evt.handle_resume(self, task)
    for w in wset:
      evt, task = self._write_waiting.pop(w)
      evt.handle_resume(self, task)

  def new(self,task):
    '''
    add a newly started task to the scheduler
    '''
    self._ready.append((task, none))
    self._numtasks += 1

  def add_ready(self, task, msg=none):
    '''
    append an already started task to the ready queue.
    msg is what to send into the task when it resumes.
    '''
    self._ready.append((task, msg))

  # add a task to the reading set
  def _read_wait(self, fileno, evt, task):
    self._read_waiting[fileno] = (evt, task)

  # add a task to the write set
  def _write_wait(self, fileno, evt, task):
    self._write_waiting[fileno] = (evt, task)

  def run(self):
    '''
    run the task scheduler until there are no tasks
    '''
    while self._numtasks:
       if not self._ready:
         self._iopoll()
       task, msg = self._ready.popleft()
       try:
         # run the coroutine to the next yield
         r = task.send(msg)
         if isinstance(r, yieldevent):
           r.handle_yield(self, task)
         else:
           raise runtimeerror('unrecognized yield event')
       except stopiteration:
         self._numtasks -= 1

# example implementation of coroutine-based socket i/o
class readsocket(yieldevent):
  def __init__(self, sock, nbytes):
    self.sock = sock
    self.nbytes = nbytes
  def handle_yield(self, sched, task):
    sched._read_wait(self.sock.fileno(), self, task)
  def handle_resume(self, sched, task):
    data = self.sock.recv(self.nbytes)
    sched.add_ready(task, data)

class writesocket(yieldevent):
  def __init__(self, sock, data):
    self.sock = sock
    self.data = data

  def handle_yield(self, sched, task):
    sched._write_wait(self.sock.fileno(), self, task)

  def handle_resume(self, sched, task):
    nsent = self.sock.send(self.data)
    sched.add_ready(task, nsent)

class acceptsocket(yieldevent):
  def __init__(self, sock):
    self.sock = sock

  def handle_yield(self, sched, task):
    sched._read_wait(self.sock.fileno(), self, task)

  def handle_resume(self, sched, task):
    r = self.sock.accept()
    sched.add_ready(task, r)

# wrapper around a socket object for use with yield
class socket(object):
  def __init__(self, sock):
    self._sock = sock

  def recv(self, maxbytes):
    return readsocket(self._sock, maxbytes)

  def send(self, data):
    return writesocket(self._sock, data)

  def accept(self):
    return acceptsocket(self._sock)

  def __getattr__(self, name):
    return getattr(self._sock, name)

if __name__ == '__main__':
  from socket import socket, af_inet, sock_stream
  import time

  # example of a function involving generators. this should
  # be called using line = yield from readline(sock)
  def readline(sock):
    chars = []
    while true:
      c = yield sock.recv(1)
      if not c:
        break
      chars.append(c)
      if c == b'\n':
        break
    return b''.join(chars)

  # echo server using generators
  class echoserver:
    def __init__(self,addr,sched):
      self.sched = sched
      sched.new(self.server_loop(addr))

    def server_loop(self,addr):
      s = socket(socket(af_inet,sock_stream))

      s.bind(addr)
      s.listen(5)
      while true:
        c,a = yield s.accept()
        print('got connection from ', a)
        self.sched.new(self.client_handler(socket(c)))

    def client_handler(self,client):
      while true:
        line = yield from readline(client)
        if not line:
          break
        line = b'got:' + line
        while line:
          nsent = yield client.send(line)
          line = line[nsent:]
      client.close()
      print('client closed')

  sched = scheduler()
  echoserver(('',16000),sched)
  sched.run()

def some_generator():
  ...
  result = yield data
  ...

f = some_generator()

# initial result. is none to start since nothing has been computed
result = none
while true:
  try:
    data = f.send(result)
    result = ... do some calculation ...
  except stopiteration:
    break