cogen and greenlets

I was playing these days with greenlets and I finally managed to find some time to make and coroutine implementation for cogen that uses the seamless stack switching of greenlets instead of the usual generator yield-based switching.

The best part about this is that cogen could support any sort of old code that uses the usual file or socket apis in the standard python library.

Unfortunately greenlets don’t work with python 2.6 right now; let me shamelessly quote this very nice reply I got from one of py.lib’s (greenlets are part of py.lib) maintainers:

“I mean come on – it’s open source, if it works for me on 2.5 why I would bother too much?”

For the love of god, is this the right attitude to run a project these days ?! I mean come on …

Anyhow, nevermind that.

So back to those coroutines using greenlets – I named them corolets – feels like a good name given the fact the the regular generator stuff is named with coroutine and there’s some need for distinction. I also added some socket wrappers in a module socketlets (not very inspired). It has a Socket class that provides regular blocking interface and makes use of the good old _fileobject class from the socket module in stdlib for the makefile stuff.

I got those 2 pieces in cogen.magic.corolets and cogen.magic.socketlets. For those who didn’t understood much here’s an example:

import sys

from cogen.core import schedulers
from cogen.magic.corolets import corolet, yield_
from cogen.magic import socketlets

@corolet
def server():
    srv = socketlets.Socket()
    adr = ('0.0.0.0', len(sys.argv)>1 and int(sys.argv[1]) or 1200)
    srv.bind(adr)
    srv.listen(64)
    while 1:
        print "Listening on", adr
        conn, addr = srv.accept()
        print "Connection from %s:%s" % addr
        m.add(handler, args=(conn, addr))

@corolet
def handler(sock, addr):
    fh = sock.makefile()
    fh.write("WELCOME TO ECHO SERVER !\r\n")
    fh.flush()

    while 1:
        line = fh.readline(1024)
        if line.strip() == 'exit':
            fh.write("GOOD BYE")
            fh.close()
            sock.close()
            return
        fh.write(line)
        fh.flush()

m = schedulers.Scheduler()
m.add(server)
m.run()

Compare that to the regular generator based example (don’t miss those yields):

import sys

from cogen.core import sockets
from cogen.core import schedulers
from cogen.core.coroutines import coroutine

@coroutine
def server():
    srv = sockets.Socket()
    adr = ('0.0.0.0', len(sys.argv)>1 and int(sys.argv[1]) or 1200)
    srv.bind(adr)
    srv.listen(64)
    while 1:
        print "Listening on", adr
        conn, addr = yield srv.accept()
        print "Connection from %s:%s" % addr
        m.add(handler, args=(conn, addr))

@coroutine
def handler(sock, addr):
    fh = sock.makefile()
    yield fh.write("WELCOME TO ECHO SERVER !\r\n")
    yield fh.flush()

    while 1:
        line = yield fh.readline(1024)
        if line.strip() == 'exit':
            yield fh.write("GOOD BYE")
            yield fh.close()
            sock.close()
            return
        yield fh.write(line)
        yield fh.flush()

m = schedulers.Scheduler()
m.add(server)
m.run()

It’s pretty rough right now – I still need to add tests and check the performance, but I wanted to give an idea of what’s going on for the braver ones :)

ctypes based iocp proactor for cogen

I’ve finally mannaged to pull off a set of wrappers for using the raw windows iocp api using ctypes, namely wrappers for GetQueuedCompletionStatus, WSARecv, WSASend, AcceptEx, ConnectEx, TransmitFile and the whatever structs and utility calls they need.

Writing the wrappers wasn’t that hard – but if you aren’t careful with ctypes you are in a world of gruesome crashes and at least 2 wasted evenings (read nights) debugging why the hell some random allocation fails. Well, for a fact, I still have no idea to why do I own those to wasted nights but thank god it doesn’t crash on my unittests anymore – I think I could have figured out why some missing object cleanup resulted in weird crashes if I knew some ctypes internals but eh, I don’t :(

Take a look at the evil code here and here, and punish me for doing such an evil thing with ctypes *malefic laugh* =]

Here’s a gem:

#~ BOOL = SOCKET hSocket, HANDLE hFile, DWORD nNumberOfBytesToWrite, DWORD nNumberOfBytesPerSend, LPOVERLAPPED lpOverlapped,LPTRANSMIT_FILE_BUFFERS lpTransmitBuffers, DWORD dwFlags
TransmitFileType = WINFUNCTYPE(BOOL, SOCKET, HANDLE, DWORD, DWORD, POINTER(OVERLAPPED), POINTER(TRANSMIT_FILE_BUFFERS), DWORD)

def _GetTransmitFilePtr(given_socket=None):
    from socket import socket
    bogus_sock = given_socket or socket()
    bogus_bytes = DWORD()
    TransmitFile = TransmitFileType(0)
    ret = WSAIoctl(
        bogus_sock.fileno(), SIO_GET_EXTENSION_FUNCTION_POINTER, byref(WSAID_TRANSMITFILE), sizeof(WSAID_TRANSMITFILE),
        byref(TransmitFile), sizeof(TransmitFile), byref(bogus_bytes), None, None
    )
    return TransmitFile

TransmitFile = _GetTransmitFilePtr()

So far the new proactor works pretty nice in cogen.

On the future of cogen coroutine library

The major refactor of the socket stuff is finally in the trunk. The Socket is more in line with the standard socket module in python (there’s a makefile and a _fileobject for readline just like the standard library). And it looks like performance is still pretty good.

The only things I can think of right now are the unittests for the socket stuff – they lack good coverage and they were actually made for the previous incompatible cogen socket api. They will be fixed eventually.

Another thing is that i’m trying to push some patches in the pywin32 module for win32file.TransmitFile and win32file.ConnectEx. But it’s going so slow that I’m thinking it isn’t going to work out and I’d better write a separate module for the GetQueueCompletionStatus, CreateIoCompletionPort, WSASend, WSARecv, ConnectEx, TransmitFile and avoid all that insane overlapped struct hacking that happens in the win32file module. Wadda ya think ?

Rewrite temptation

Well, not exactly a rewrite, merely a major internal refactor.

The thought of cleaning up the socket and reactor internals really nags me. I would move the platform dependant code from the sockets module in the reactors – so, say, the reactors would have the platform specific code for recv, send, connect, accept instead of each socket operation class. Also, the fact is that the iocp was added later and the code was patched to support a what was a design scheme based on a reactor pattern – so adding support for another proactor based api (say, linux’s aio) would really mudge the internals. Also, moving the readline stuff out of the socket operations in a coroutine would make things real fancy and easy to follow.

Meh, on second thought, i would not have real gain, besides making more easy to add support for another proactor api like aio (wich btw just sucks, it wasn’t made for sockets – no support for accept or connect) and priding myself for the fancier internals i would probably get lower performance – and I still have yet to fine tune the performance part.

Now that I have relieved myself I’d better concentrate myself on that diploma thesis. =]

comet chat in pylons (with cogen)

There is something very cool about wsgi: asynchronicity at it’s core ! The spec was made with this in mind – I absolutely love wsgi.

I’ve been playing recently with pylons and i’ve made a example chat app – just a proof-of-concept comet application with long pooling in a pylons app using a custom wsgi server (cogen.wsgi).

Here’s the controller code that give me a feeling I will be burned to the stake for too much magic. =]

import logging

from pylons import request, response, session
from pylons import tmpl_context as c
from pylons.controllers.util import abort, redirect_to, url_for

from chatapp.lib.base import BaseController
# import chatapp.model as model

log = logging.getLogger(__name__)
from cogen.core import queue, events
from cogen.core.coroutines import coro
from cogen.core.pubsub import PublishSubscribeQueue
pubsub = PublishSubscribeQueue()

class Client:
    def __init__(self):
        self.messages = queue.Queue(10)
        self.dead = False
    @coro
    def watch(self):
        """This is a coroutine that runs permanently for each participant to the
        chat. If the participant has more than 10 unpulled messages this
        coroutine will die.

        `pubsub` is a queue that hosts the messages from all the
        participants.
          * subscribe registers this coro to the queue
          * fetch pulls the recent messages from the queue or waits if there
        are no new ones.

        self.messages is another queue for the frontend comet client (the
        pull action from the ChatController will pop messages from this queue)
        """
        yield pubsub.subscribe()
        while 1:
            messages = yield pubsub.fetch()
            try:
                yield self.messages.put_nowait(messages)
            except:
                print 'Client %s is dead.' % self
                self.dead = True
                break
class ChatController(BaseController):

    def push(self):
        """This action puts a message in the global queue that all the clients
        will get via the 'pull' action."""
        yield request.environ['cogen.call'](pubsub.publish)(
            "%X: %s" % (id(session['client']), request.body)
        )
        # the request.environ['cogen.*'] objects are the the asynchronous
        # wsgi extensions offered by cogen - basicaly they do some magic to
        # make the code here work as a coroutine and still work with any
        # middleware
        yield str(request.environ['cogen.wsgi'].result)

    def pull(self):
        """This action does some state checking (adds a object in the session
        that will identify this chat participant and adds a coroutine to manage
        it's state) and gets new messages or bail out in 10 seconds if there are
        no messages."""
        if not 'client' in session or session['client'].dead:
            client = Client()
            print 'Adding new client:', client
            session['client'] = client
            session.save()
            yield request.environ['cogen.core'].events.AddCoro(client.watch)
        else:
            client = session['client']

        yield request.environ['cogen.call'](client.messages.get)(timeout=10)

        if isinstance(request.environ['cogen.wsgi'].result, events.OperationTimeout):
            pass
        elif isinstance(request.environ['cogen.wsgi'].result, Exception):
            import traceback
            traceback.print_exception(*request.environ['cogen.wsgi'].exception)
        else:
            yield "%s\r\n"% '\r\n'.join(request.environ['cogen.wsgi'].result)

The frontend code is fairly simple:

    <h1>chat powered by <a href="http://code.google.com/p/cogen/">cogen</a></h1>
    <form name="chat">
    <textarea id="ouput" name="output" rows="10" cols="80" readonly="readonly"></textarea><br/>
    <input id="tosend" name="tosend" value="" size="80"/>
    </form>
<script language="javascript">
    function xhr(url, callback, data) {
        var req = window.XMLHttpRequest?new XMLHttpRequest():new ActiveXObject('Microsoft.XMLHTTP');
        req.onreadystatechange = function() {
            if (req.readyState == 4) {
                if (callback) callback(req);
            }
        }
        req.open(data?"POST":"GET", url, true);
        req.send(data);
    }
    function pull(req) {
        if(req.status == 200) {
            document.chat.output.value = req.responseText + document.chat.output.value;
            xhr('/chat/pull', pull)
        } else {
            alert(req.responseText);
        }
    }
    xhr('/chat/pull', pull);
    document.chat.tosend.onkeydown = function (event) {
        event = event || window.event;
        var key = event.which || event.keyCode;
        if (key == 13) {
            xhr('/chat/push', null, document.chat.tosend.value);
            document.chat.tosend.value = '';
        }
    }
    document.chat.onsubmit = function() { return false; }
</script>

I think the javascript part looks fairly obvious, but the controller uses a bunch of magic – here’s the workflow:

• a client load the page and calls the pull page via xhr
• the pull page waits 10 sencond for a new message to be posted in the messages queue
• another client sends a message via push page via xhr
• the push page adds the message in the pubsub queue
• the coroutines (the ‘watch’ methods) are waken up and get the new message from the fetch() calls
• each ‘watch’ coroutine posts that message in the client’s queue (via messages.put_nowait)

also, the messages queue is limited to 10 messages so if there are 10 updates and a client doesn’t get them it’s associated ‘watch’ coroutine will die

• each loading pull page gets that message (via messages.get(timeout=10)) and send it to the client

The fairly obscure cogen docs might also be helpfull :)

Almost forgot, to run this you need a trunk version of cogen (easy_install cogen==dev or check out http://cogen.googlecode.com/svn/trunk/)
and the ChatApp example (you can view all the code here and get it from here)

Once you have installed cogen (easy_install or setup.py develop), install the ChatApp (setup.py develop) and start the app with paster serve test.ini. And try finding some bugs (just joking) at http://127.0.0.1:5000/

Qt reactor in cogen

I’ve implemented a reactor to integrate the cogen loop in the qt main loop.

You whould use it like this:

from PyQt4 import QtGui

app = QtGui.QApplication([])
hello = QtGui.QPushButton( "Hello world!" )
hello.resize(100, 30)
hello.show()

from cogen.core import reactors, schedulers
m = schedulers.Scheduler(reactor=reactors.QtReactor)

def lorem_ipsum_app(environ, start_response):
    start_response('200 OK', [('Content-type','text/plain'), ('Content-Length','19')])
    return ['Lorem ipsum dolor..']

server = wsgi.WSGIServer(
  ('0.0.0.0', 9001),
  lorem_ipsum_app,
  m,
)
m.add(server.serve)
m.run()
# voila ! this actualy runs the QApplication loop with the cogen hooks

Though the reactor isn’t tested, well, unittested – i have a bunch of tests to keep the code sane.
The problem is that Qt is a big nasty monolithic framework, well, it’s nice if you don’t need to do any weird stuff – like running the main application loop in a second thread.

I just don’t understand why they made this ugly limitation in qt: the application loop must run in the main thread. I’ll fix those tests somehow, but in the meanwhile i’ll just continue cursing Qt for making me feel miserable about my unittests.

cogen and pylons can play !

Just so you know, cogen is a coroutine framework (based on the bidirectional generators from python 2.5) that has a wsgi server with some async extensions.

Straight to the point, i’m going to show the basics by building the backend of a web based (with comet-style ajax) irc client. This is a proof of concept and the javascript part of the app will come in a future blog post.

So a async app with cogen works like a streaming app in pylons. Everything is fine if you don’t use middleware that consumes the appiter (sadly, the ErrorHandler middleware does just that – and it can’t be fixed becouse you can’t report a error after you sent a part of the response to the client).

Grab the latest TIP from pylons (get Mercurial first):

hg clone https://www.knowledgetap.com/hg/pylons-dev Pylons

Grab the latest trunk from paste:

svn co http://svn.pythonpaste.org/Paste/trunk Paste

Grab the latest trunk from cogen:

svn co http://cogen.googlecode.com/svn/trunk/ cogen

Ok, so install them (usualy a python setup.py develop). Next thing you make a pylons app (paster create -t pylons). And you need to make just 2 changes to de middeware config:

In middleware.py comment out the error middleware like this:

# Handle Python exceptions
#~ app = ErrorHandler(app, global_conf, **config['pylons.errorware'])

and change the Registry to work with streaming apps:

# Establish the Registry for this application
app = RegistryManager(app, streaming=True)

You need to change the development.ini, just replace the server:main section with this:

[server:main]
use = egg:cogen#http
host = 0.0.0.0
port = 5000

Allright, now add a controller (paster controller irc, or something like that). So we’ll make something extremely simple: we’ll write a relay that just passes the messages from the server to the client in json format with a timeout of 30 seconds when there no messages arrive (that’s our comet long-poll update).

Just put this code in controllers/irc.py:

import logging

from pylons import request, response, session
from pylons import tmpl_context as c
from pylons.controllers.util import abort, redirect_to, url_for
from pylons.decorators import validate
from formencode import validators
from cogenircapp.lib.base import BaseController
import cogenircapp.model as model

log = logging.getLogger(__name__)

from cogen.core.coroutines import coro, debug_coroutine
from cogen.core import events, sockets
from cogen.core.util import priority
from cogen.core import queue
from cogen.web import async

import simplejson

def parsemsg(s): # stolen from twisted.words
    """Breaks a message from an IRC server into its prefix, command, and arguments.
    """
    prefix = ''
    trailing = []
    if not s:
        raise Exception("Empty line.")
    if s[0] == ':':
        prefix, s = s[1:].split(' ', 1)
    if s.find(' :') != -1:
        s, trailing = s.split(' :', 1)
        args = s.split()
        args.append(trailing)
    else:
        args = s.split()
    command = args.pop(0)
    return prefix, command, args

class Connection:
    def __init__(self, server, reconnect_interval=60, sock_timo=15):
        self.server = server
        self.reconnect_interval = reconnect_interval
        self.connected = False
        self.sock_timo = sock_timo
        self.events = queue.Queue(25) # Max 25 pending events, well, messages
                    # from the server. After that we'll lose the connection.

    @coro
    def pull(self):
        """This coroutine handles the server connection, does a basic parse on
        the received messages and put them in a queue named events.
        The controllers pull method will take the messages from that queue.
        """
        self.sock = sockets.Socket()
        while not self.connected:
            try:
                addr = self.server.split(':')
                if len(addr) < 2:
                    addr.append(6667)
                else:
                    addr[1] = int(addr[1])
                yield self.events.put_nowait(('', 'CONNECTING', ''))
                yield self.sock.connect(tuple(addr), timeout=self.sock_timo)
                self.connected = True
            except events.OperationTimeout, e:
                yield self.events.put_nowait(('', 'CONNECT_TIMEOUT', str(e)))
                yield events.Sleep(self.reconnect_interval)

        yield self.events.put_nowait(('', 'CONNECTED', ''))
        while 1:
            try:
                line = yield self.sock.readline(8192)
                prefix, command, params = parsemsg(line)
                yield self.events.put_nowait((prefix, command, params))
            except Exception, e:
                yield self.events.put_nowait(('', 'ERROR', str(e)))
                break

from pylons.templating import render_mako as render

class IrcController(BaseController):
    """
    This controller supports multiple server connections.
    """
    def index(self):
        if 'connections' not in session:
            session['connections'] = {}
            session.save()
        return render('index.mako')

    def push(self, id):
        "Sends a message to the specified connection (id)"
        conn = session['connections'].get(id, None)
        if conn:
            yield request.environ['cogen.core'].sockets.WriteAll(conn.sock,
                            request.environ['wsgi.input'].read() +'\r\n')
            if isinstance(request.environ['cogen.wsgi'].result, Exception):
                yield simplejson.dumps(('', 'ERROR', str(e)))
            else:
                yield simplejson.dumps(('', 'PUSH_OK', ''))
        else:
            yield simplejson.dumps(('', 'ERROR', 'Invalid connection id.'))

    def connect(self, server):
        "Connects to a server and return a connection id."
        conns = session['connections']
        id = str(len(conns))
        conn = Connection(server)
        conns[id] = conn
        yield request.environ['cogen.core'].events.AddCoro(conn.pull)
        yield id

    def pull(self, id):
        """Take the messages from the queue and if there are none wait 30
        seconds till returning an empty message.

        Also, cogen's wsgi async extensions are in the environ and prefixed with
        'cogen.'
        """
        conn = session['connections'].get(id, None)
        if conn:
            ev_list = []
            while 1:
                # ok, so this might look a bit ugly but the concept is very simple
                #  you yield a special object from the environ that does some magic
                #  and the wsgi server will resume the app when it has the result
                yield request.environ['cogen.call'](conn.events.get_nowait)()
                event = request.environ['cogen.wsgi'].result
                # also, we can't have better exception handling in this wsgi
                # contraption and we need to check the result for exceptions
                if isinstance(event, queue.Empty):
                    break
                elif isinstance(event, Exception):
                    ev_list.append(('', 'ERROR', str(event)))
                    break
                else:
                    ev_list.append(event)
            if ev_list:
                yield simplejson.dumps(ev_list)
            else:
                # if we don't have any updates atm, we'll wait 30 secs for one
                yield request.environ['cogen.call'](conn.events.get)(timeout=30)
                event = request.environ['cogen.wsgi'].result
                if isinstance(event, events.OperationTimeout):
                    yield simplejson.dumps([])
                elif isinstance(event, Exception):
                    yield simplejson.dumps([('', 'ERROR', str(event))])
                else:
                    yield simplejson.dumps([event])
        else:
            yield simplejson.dumps(('', 'ERROR', 'Invalid connection id.'))

For the frontend i’m thinking to use YUI. I’ll have it finished in a couple of days.

LE: sources are at: https://cogen.googlecode.com/svn/trunk/examples/cogen-irc though the javascript interface needs polishing (a lot of it to get blog-post worthy).

cogen: python coroutine library introduction

Ok, introductory stuff: before python 2.5 generators were just a unidirectional computation structure. That means one could get values out of the generator. In python 2.5 we have the enhancements from PEP 342: Coroutines via Enhanced Generators – generators have 3 extra methods: sent, throw and close and the yield statement is a expression now. So the idea is that the generators are now a bidirectional computation structure: we can get values out and in the generator.
Here’s a very simple example of what you can do now:

>>> def somegenerator():
...     print (yield 1)
...     print (yield 2)
...
>>> g = somegenerator()
>>> print g.next()
1
>>> print g.send('a')
a
2
>>>

Coroutines are program components that generalize subroutines to allow multiple entry points and suspending and resuming of execution at certain locations, to quote wikipedia. So we could look at a generator as a coroutine and the yield expressions as suspend/resume points. Take this example:

def somecoroutine():
    data = (yield nonblocking_read(my_socket, nbytes))

Well, this example is actually from the PEP342 spec.Here’s an example that uses cogen:

from cogen.core.coroutine import coroutine
from cogen.core.schedulers import Scheduler
from cogen.core.sockets import Socket
from cogen.core.reactors import SelectReactor

@coroutine
def somecoroutine():
    mysocket = Socket() # cogen's socket wrapper
    yield mysocket.connect(('www.google.com',80))
    yield mysocket.writeall("GET / HTTP/1.1\r\nHost: www.google.com\r\n\r\n")
    result = yield mysocket.read(10240)
    print result

sched = Scheduler(reactor=SelectReactor)
sched.add(somecoroutine)
sched.run() # this is the main loop

• coroutine is a special decorator that wrapps generators and functions alike in a special Coroutine class. So for example, if you decorate a function that isn’t actualy a generator nothing bad will happen.
• Socket is a special wrapper that looks like the regular socket object (well, we actualy have a different flavour for usual recv and send, namely: read, readall, readline, write, writeall) but returns some special objects that we call operations. These operations instruct the cogen scheduler what to do with the coroutine.
• reactor=SelectReactor – there are actually 5 reactors: SelectReactor, PollReactor, KQueueReactor, EpollReactor, IOCPProactor. For KQueueReactor and EpollReactor you need C extension modules (that are provided in the lib directory) and for IOCPProactor you need the pywin32 extensions – also, IOCPProactor is in development atm. The reactor coupled with the socket operations handles all the nasty asynchronous networking code.

Also, there are a bunch of examples here.

Calls to other coroutines need to be made through the scheduler. To call another coroutine you yield a Call operation. Here’s an example:

>>> from cogen.core.coroutine import coroutine
>>> from cogen.core.schedulers import Scheduler
>>> from cogen.core import events
>>>
>>> @coroutine
... def foo():
...     print 'foo'
...     result = yield events.Call(bar, args=("ham",))
...     print result
...
>>> @coroutine
... def bar(what):
...     print 'bar'
...     raise StopIteration("spam, %s and eggs" % what)
...
>>> sched = Scheduler()
>>> sched.add(foo)
<foo Coroutine instance at 0x00D1E6B8 wrapping <function foo at 0x00C01B30>, state: NOTSTARTED>
>>> sched.run()
foo
bar
spam, ham and eggs
>>>

cogen has also other usefull operations. Take for example this using signals:

>>> from cogen.core.coroutine import coroutine
>>> from cogen.core.schedulers import Scheduler
>>> from cogen.core import events
>>>
>>> @coroutine
... def foo():
...     yield events.Signal("bar", 'spam')
...     yield events.Signal("bar", 'ham')
...     yield events.Signal("bar", 'eggs')
...     yield events.Sleep(3)
...
>>> @coroutine
... def bar():
...     print (yield events.WaitForSignal("bar"))
...     print (yield events.WaitForSignal("bar"))
...     print (yield events.WaitForSignal("bar"))
...     try:
...         print (yield events.WaitForSignal("bar", timeout=2))
...     except events.OperationTimeout:
...         print 'No more stuff !'
...
>>> sched = Scheduler()
>>> sched.add(bar)
<bar Coroutine instance at 0x00D1E6B8 wrapping <function bar at 0x00C551F0>, state: NOTSTARTED>
>>> sched.add(foo)
<foo Coroutine instance at 0x00D1EC90 wrapping <function foo at 0x00C01B30>, state: NOTSTARTED>
>>> sched.run()
spam
ham
eggs
No more stuff !
>>>

• events.Signal(name, value) – name is an object that needs to be immutable (strings, tuples, numbers and frozen sets are immutable for example); value is what is yield events.WaitForSignal going to return.
• cogen‘s scheduler main loop (run) exits when there is no more stuff to run (no active coroutines, no socket operations waiting etc) – events.WaitForSignal does not qualify, otherwise the scheduler would run endlessly and nothing would happen (no other coroutines to raise the signal), that is why there is a events.Sleep(3) in foo.




There’s a lot more to talk about, so check the docs on events and sockets, the examples.