[Python-ideas] Yield-From Example 2: Scheduler

Greg Ewing greg.ewing at canterbury.ac.nz
Thu Feb 19 22:32:23 CET 2009

(I've tried to send this twice before and got smtp 554 errors.
I'm trying again without the attachment.)

Yield-From Example 2: A Scheduler for Generator-Based Threads

Just for fun, let's write a thread scheduler.

Each thread will be a generator. Whenever a thread wants to suspend
itself, it will do a 'yield'. We won't make any use of values sent
or received by yields in this example.

We'll start with a global variable to hold the currently running

   current = None

We'll also want a queue of threads that are waiting to run.

   ready_list = []

The first thing we'll want is a way of getting a thread into
the scheduling system.

   def schedule(g):

The core loop of the scheduler will repeatedly take the thread at the
head of the queue and run it until it yields.

   def run():
     global current
     while ready_list:
       g = ready_list[0]
       current = g
       except StopIteration:

If the thread is still at the head of the ready list after it has
yielded, we move it to the end, so that the ready threads will run
round-robin fashion.

   def expire_timeslice(g):
     if ready_list and ready_list[0] is g:
       del ready_list[0]

When the thread finishes, we use the following function to remove
it from the scheduling system.

   def unschedule(g):
     if g in ready_list:

We've got enough so far to try a simple test.

   def person(name, count):
     for i in xrange(count):
       print name, "running"

   schedule(person("John", 2))
   schedule(person("Michael", 3))
   schedule(person("Terry", 4))


We can run this with present-day Python, and we get

   John running
   Michael running
   Terry running
   John running
   Michael running
   Terry running
   Michael running
   Terry running
   Terry running

Waiting for Resources

Things get more interesting when our threads do something non-trivial.

Let's turn our people into dining philosophers. For this we'll need a
way to represent forks (the eating kind, not the unix kind) and a way
for a thread to wait for one to become available.

But before launching into that, let's add two more functions to our
scheduler that will come in useful.

   def block(queue):

This removes the currently running thread from the ready list and
adds it to a list that you specify.

   def unblock(queue):
     if queue:
       g = queue.pop(0)

This removes the thread at the head of the specified list, if
any, and adds it to the ready list.

Now we can start implementing an eating utensil.

   class Utensil:

     def __init__(self, id):
       self.id = id
       self.available = True
       self.queue = []

The utensil has a flag indicating whether it's available, and a queue
of threads waiting to use it. To acquire a utensil, we first check to see
whether it is available. If not, we block the current thread on the
queue, and then yield. When we get to run again, it's our turn, so we
mark the utensil as being in use.

     def acquire(self):
       if not self.available:
       self.available = False

To release the utensil, we mark it as available and then unblock the
thread at the head of the queue, if any.

     def release(self):
       self.available = True

Next we need a life cycle for a philosopher.

   def philosopher(name, lifetime, think_time, eat_time, left_fork, right_fork):
     for i in xrange(lifetime):
       for j in xrange(think_time):
         print name, "thinking"
       print name, "waiting for fork", left_fork.id
       yield from left_fork.acquire()
       print name, "acquired fork", left_fork.id
       print name, "waiting for fork", right_fork.id
       yield from right_fork.acquire()
       print name, "acquired fork", right_fork.id
       for j in xrange(eat_time):
         # They're Python philosophers, so they eat spam rather than spaghetti
         print name, "eating spam"
       print name, "releasing forks", left_fork.id, "and", right_fork.id
     print name, "leaving the table"

Now we can set up a scenario.

   forks = [Utensil(i) for i in xrange(3)]
   schedule(philosopher("Plato", 7, 2, 3, forks[0], forks[1]))
   schedule(philosopher("Socrates", 8, 3, 1, forks[1], forks[2]))
   schedule(philosopher("Euclid", 5, 1, 4, forks[2], forks[0]))

We can't run this in current Python, because of the yield-froms.
However, we can test it by substituting for-loops such as

       for _ in left_fork.acquire(): yield

After doing this, the output is

   Plato thinking
   Socrates thinking
   Euclid thinking
   Plato thinking
   Socrates thinking
   Euclid waiting for fork 2
   Euclid acquired fork 2
   Euclid waiting for fork 0
   Euclid acquired fork 0
   Euclid eating spam

Waiting for External Events

So far our thread system has been completely self-absorbed and unable
to deal with the outside world. Let's arrange things so that, if there
are no threads ready to run, the scheduler will wait for some file
to become readable or writable using select(). It's easiest to do
this by writing a new main loop that builds on the previous one.

   def run2():
     while 1:
       if not wait_for_event():

We will need a data structure to hold threads waiting for files. Each
file needs two queues associated with it, for threads waiting to read
and write respectively.

   class FdQueues:

     def __init__(self):
       self.readq = []
       self.writeq = []

We will keep a mapping from file objects to their associated FdQueue

   fd_queues = {}

The following function retrieves the queues for a given fd, creating
new ones if they don't already exist.

   def get_fd_queues(fd):
     q = fd_queues.get(fd)
     if not q:
       q = FdQueues()
       fd_queues[fd] = q
     return q

Now we can write a new pair of scheduling primitives to block on
a file.

   def block_for_reading(fd):

   def block_for_writing(fd):

It's expected that the thread calling these will immediately yield
afterwards. We could incorporate the yield into these functions, but
we'll be building higher level functions on top of these shortly, and
it will be more convenient to do the yield there.

We'll also want a way of removing a file from the fd_queues when
we've finished with it, so we'll add a function to close it and
clean up.

   def close_fd(fd):
     if fd in fd_queues:
       del fd_queues[fd]

Now we can write wait_for_event(). It's a bit longwinded, but fairly
straightforward. We build lists of file objects having nonempty read
or write queues, pass them to select(), and for each one that's ready,
we unblock the thread at the head of the relevant queue. If there are
no threads waiting on any files, we return False to tell the scheduler
there's no more work to do.

   def wait_for_event():
     from select import select
     read_fds = []
     write_fds = []
     for fd, q in fd_queues.iteritems():
       if q.readq:
       if q.writeq:
     if not (read_fds or write_fds):
       return False
     read_fds, write_fds, _ = select(read_fds, write_fds, [])
     for fd in read_fds:
     for fd in write_fds:
     return True

At this point we can try a quick test to see if everything works
so far.

   def loop():
     while 1:
       print "Waiting for input"
       print "Input is ready"
       line = stdin.readline()
       print "Input was:", repr(line)
       if not line:


Sample session:

   Waiting for input
   Input is ready
   Input was: 'asdf\n'
   Waiting for input
   Input is ready
   Input was: 'qwer\n'
   Waiting for input
   Input is ready
   Input was: ''

It's not a very convincing test yet, though, since there's only
one thread, so let's play around with some sockets and build a
multithreaded server.

A Spam Server

We're going to implement the following protocol. The client sends the
word "SPAM" followed by a number, and the server replies with "100
SPAM FOLLOWS" and the corresponding number of repetitions of the
phrase "spam glorious spam". If the requested number is not greater
than zero or the request is malformed, the server replies "400 WE ONLY

We could do with some higher-level functions for blocking operations
on sockets, so let's write a few. First, accepting a connection from
a listening socket.

   def sock_accept(sock):
     return sock.accept()

Now reading a line of text from a socket. We keep reading until the
data ends with a newline or EOF is reached. (We're assuming that the
client will wait for a reply before sending another line, so we don't
have to worry about reading too much.) We also close the socket on
EOF, since we won't be reading from it again after that.

   def sock_readline(sock):
     buf = ""
     while buf[-1:] != "\n":
       data = sock.recv(1024)
       if not data:
       buf += data
     if not buf:
     return buf

Writing data to a socket. We loop until all the data has been written.
We don't use sendall(), because it might block, and we don't want to
hold up other threads.

   def sock_write(sock, data):
     while data:
       n = sock.send(data)
       data = data[n:]

Now we're ready to write the main loop of the server. It will set up a
listening socket, then repeatedly accept connections and spawn a
thread to handle each one.

   port = 4200

   def listener():
     lsock = socket(AF_INET, SOCK_STREAM)
     lsock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
     lsock.bind(("", port))
     while 1:
       csock, addr = yield from sock_accept(lsock)
       print "Listener: Accepted connection from", addr

The handler function handles the interaction with one client session.

   def handler(sock):
     while 1:
       line = yield from sock_readline(sock)
       if not line:
         n = parse_request(line)
         yield from sock_write(sock, "100 SPAM FOLLOWS\n")
         for i in xrange(n):
           yield from sock_write(sock, "spam glorious spam\n")
       except BadRequest:
         yield from sock_write(sock, "400 WE ONLY SERVE SPAM\n")

The handler uses the following function to parse the request and check
it for validity.

   class BadRequest(Exception):

   def parse_request(line):
     tokens = line.split()
     if len(tokens) != 2 or tokens[0] != "SPAM":
       raise BadRequest
       n = int(tokens[1])
     except ValueError:
       raise BadRequest
     if n < 1:
       raise BadRequest
     return n

All we need to do now is spawn the main loop and run the scheduler.


At this point, I got fed up with expanding yield-from statements by
hand, and wrote a program to do it for me.

Here's a sample client session:

   % telnet localhost 4200
   Connected to localhost.
   Escape character is '^]'.
   SPAM 3
   spam glorious spam
   spam glorious spam
   spam glorious spam
   telnet> Connection closed.


The yield-from statement makes it possible to write thread code using
generators almost the same way as you would write ordinary code.

Whether it's any easier or clearer than using things like yield
Call(g(x)) and yield Return(x) is debatable. However, I think this
example does show that the implementation of a generator-based
scheduler can be very clean and simple when yield-from is available,
and if it is suitably optimised, probably more efficient as well.

I was going to attach a zip file containing the code I used for
testing all this, but the python-ideas mail server seems to be
bouncing the message when I do that. I'll try sending it in a
separate message.


More information about the Python-ideas mailing list