(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 thread. 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): ready_list.append(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 try: g.next() except StopIteration: unschedule(g) else: expire_timeslice(g) 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] ready_list.append(g) When the thread finishes, we use the following function to remove it from the scheduling system. def unschedule(g): if g in ready_list: ready_list.remove(g) We've got enough so far to try a simple test. def person(name, count): for i in xrange(count): print name, "running" yield schedule(person("John", 2)) schedule(person("Michael", 3)) schedule(person("Terry", 4)) run() 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): queue.append(current) unschedule(current) 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) schedule(g) 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: block(self.queue) yield 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 unblock(self.queue) 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" yield 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" yield print name, "releasing forks", left_fork.id, "and", right_fork.id left_fork.release() right_fork.release() 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])) run() 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 ...etc... 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: run() if not wait_for_event(): return 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 instances. 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): block(get_fd_queues(fd).readq) def block_for_writing(fd): block(get_fd_queues(fd).writeq) 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] fd.close() 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: read_fds.append(fd) if q.writeq: write_fds.append(fd) if not (read_fds or write_fds): return False read_fds, write_fds, _ = select(read_fds, write_fds, []) for fd in read_fds: unblock(fd_queues[fd].readq) for fd in write_fds: unblock(fd_queues[fd].writeq) 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" block_for_reading(stdin) yield print "Input is ready" line = stdin.readline() print "Input was:", repr(line) if not line: break schedule(loop()) run2() Sample session: Waiting for input asdf Input is ready Input was: 'asdf\n' Waiting for input qwer 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 SERVE SPAM". 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): block_for_reading(sock) yield 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": block_for_reading(sock) yield data = sock.recv(1024) if not data: break buf += data if not buf: close_fd(sock) 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: block_for_writing(sock) yield 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)) lsock.listen(5) while 1: csock, addr = yield from sock_accept(lsock) print "Listener: Accepted connection from", addr schedule(handler(csock)) The handler function handles the interaction with one client session. def handler(sock): while 1: line = yield from sock_readline(sock) if not line: break try: 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): pass def parse_request(line): tokens = line.split() if len(tokens) != 2 or tokens[0] != "SPAM": raise BadRequest try: 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. schedule(listener()) run2() 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 Trying 127.0.0.1... Connected to localhost. Escape character is '^]'. SPAM 3 100 SPAM FOLLOWS spam glorious spam spam glorious spam spam glorious spam EGGS 400 WE ONLY SERVE SPAM ^] telnet> Connection closed. % Conclusions ----------- 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. -- Greg