~vijaykumar

[ Home | Feed | Twitter | Vector Art | Ascii Art | Tutorials ]

Protothreads in Python

Mon, 28 Mar 2011

I did a lightning talk on Protothreads at the ChennaiPy user group meeting last week. This post contains the code examples demonstrated during the talk.

The code was written to show how Protothreads work, but it is not the best way to implement Protothreads. For one, the examples use 100% of the CPU, because the scheduler does not block. For a more complete implementation, check the source code of the Kiwi project.

BTW, the term Protothreads comes from a co-operative multi-threading implementation in C by Adam Dunkels.

Threads waiting on timeouts
import time

class WaitForTimeout(object):
    def __init__(self, timeout):
        self.start = self.now()
        self.end = self.start + timeout

    def now(self):
        return time.time()

    def runnable(self):
        return self.now() > self.end

class WaitForNone(object):
    def runnable(self):
        return True

def thread1():
    while True:
        print "Hello World"
        yield WaitForTimeout(1)

def thread2():
    while True:
        print "Goodbye World"
        yield WaitForTimeout(5)

t1 = thread1()
t2 = thread2()

c1 = WaitForNone()
c2 = WaitForNone()

while True:
    if c1.runnable():
        c1 = t1.next()

    if c2.runnable():
        c2 = t2.next()
Generic scheduler
import time

class WaitForTimeout(object):
    def __init__(self, timeout):
        self.start = self.now()
        self.end = self.start + timeout

    def now(self):
        return time.time()

    def runnable(self):
        return self.now() > self.end

class WaitForNone(object):
    def runnable(self):
        return True

def thread1():
    while True:
        print "Hello World"
        yield WaitForTimeout(1)

def thread2():
    while True:
        print "Goodbye World"
        yield WaitForTimeout(5)

def schedule(threads):
    cond = [WaitForNone()] * len(threads)

    while True:
        for i in range(len(threads)):
            if cond[i] and cond[i].runnable():
                try:
                    cond[i] = threads[i].next()
                except StopIteration, e:
                    cond[i] = None

threads = [thread1(), thread2()]
schedule(threads)
Threads waiting on fds
import time
import urllib
import select
import sys

class WaitForTimeout(object):
    def __init__(self, timeout):
        self.start = self.now()
        self.end = self.start + timeout

    def now(self):
        return time.time()

    def runnable(self):
        return self.now() > self.end

class WaitForIO(object):
    def __init__(self, fd, eventmask):
        self.poll = select.poll()
        self.poll.register(fd, eventmask)

    def runnable(self):
        if self.poll.poll(0):
            return True
        else:
            return False

class WaitForNone(object):
    def runnable(self):
        return True

size = 0
total = 1
def thread1():
    global total, size
    fp = urllib.urlopen("http://www.zilogic.com/")
    msg = fp.info()
    total = int(msg["Content-Length"])

    size = 0
    while True:
        yield WaitForIO(fp.fileno(), select.POLLIN)
        data = fp.read(1024)
        size += len(data)
        if data == "":
            break

def thread2():
    while True:
        percent = (size * 100.0 / total)
        progress = "["
        progress += "*" * (int(percent) / 2)
        progress += " " * (int(100 - percent) / 2)
        progress += "]"
        print progress,
        print "%.2f%%\r" % (size * 100.0 / total),
        sys.stdout.flush()
        yield WaitForTimeout(1)

def schedule(threads):
    cond = [WaitForNone()] * len(threads)

    while True:
        for i in range(len(threads)):
            if cond[i] and cond[i].runnable():
                try:
                    cond[i] = threads[i].next()
                except StopIteration, e:
                    cond[i] = None

threads = [thread1(), thread2()]
schedule(threads)
Threads waiting on socket and input fd
import time
import urllib
import select
import sys
import tty
import termios

class WaitForTimeout(object):
    def __init__(self, timeout):
        self.start = self.now()
        self.end = self.start + timeout

    def now(self):
        return time.time()

    def runnable(self):
        return self.now() > self.end

class WaitForIO(object):
    def __init__(self, fd, eventmask=select.POLLIN):
        self.poll = select.poll()
        self.poll.register(fd, eventmask)

    def runnable(self):
        if self.poll.poll(0):
            return True
        else:
            return False

class WaitForNone(object):
    def runnable(self):
        return True

class Or(object):
    def __init__(self, *cond_list):
        self.cond_list = cond_list

    def runnable(self):
        for cond in self.cond_list:
            if cond.runnable():
                return True
        return False

size = 0
total = 1
def thread1():
    global total, size
    fp = urllib.urlopen("http://www.zilogic.com/releases/zkit-51-sw-1.1.iso")
    msg = fp.info()
    total = int(msg["Content-Length"])

    size = 0
    while True:
        yield WaitForIO(fp.fileno())
        data = fp.read(1024)
        size += len(data)
        if data == "":
            break

def thread2():
    fd = sys.stdin.fileno()
    attrs = termios.tcgetattr(fd)
    tty.setraw(fd)

    while True:
        percent = (size * 100.0 / total)
        progress = "["
        progress += "*" * (int(percent) / 2)
        progress += " " * (int(100 - percent) / 2)
        progress += "]"
        print progress,
        print "%.2f%%\r" % (size * 100.0 / total),
        sys.stdout.flush()

        timeout = WaitForTimeout(1)
        io = WaitForIO(fd)
        yield Or(timeout, io)

        if io.runnable():
            op = sys.stdin.read(1)
            if op == 's':
                termios.tcsetattr(fd, termios.TCSAFLUSH, attrs)
                sys.exit(0)

def schedule(threads):
    cond = [WaitForNone()] * len(threads)

    while True:
        for i in range(len(threads)):
            if cond[i] and cond[i].runnable():
                try:
                    cond[i] = threads[i].next()
                except StopIteration, e:
                    cond[i] = None

threads = [thread1(), thread2()]
schedule(threads)

Permalink | Add Comment | Share: Twitter, Facebook, Buzz, ... | Tags: python

blog comments powered by Disqus

Powered by Python | Made with PyBlosxom | Valid XHTML 1.1 | Best Viewed With Any Browser | Icon Credits | CC-BY-SA