[Stackless] Stackless based replacement

[Arnar Birgisson]

Hi Larry,

On Fri, Oct 3, 2008 at 00:06, Larry Dickson <ldickson at cuttedge.com> wrote:
> If I'm following you right, asynchronous IO is not needed. All that is
> needed is a loop on a non-blocking select

Non-blocking select is one form of asynchronous IO, right? Maybe we
differ in terminology here.

> (or, to avoid busyness, a select
> against the IO and one tick later than the last time noted by the loop). If
> you can put code into the virtual machine, it can be even cleaner: an
> unconditional swap out of the tasklet; whenever its turn comes up in the
> round robin, an unblocking select on its IO; and whenever all tasklets are
> waiting on IO, a blocking select on all outstanding IO.
>
> By dealing with the selects alone, you can emulate the occam ALT primitive
> which says "swap out until any of this list come ready." Old Inmos
> documentation shows how to get the state machine exactly right for that,
> including timers.

Instead of reinventing the wheel here, I'm in favour of using
mechanism that already handles this kind of thing. libevent and its
Python bindings pyevent is one. Regular files and network streams can
be (rudimentarily) emulated with some simple wrappers on channels (not
that this does timers too):

import time

import stackless
import event

def _loop():
    while _loop.running:
        event.loop(True)
        if stackless.getruncount() == 1:
            time.sleep(0.0001)
        stackless.schedule()
_loop.running = False

def init(interval=0.0001):
    """Initializes the libevent system and starts a tasklet that executes
    the libevent dispatcher on every round."""
    if _loop.running: return
    _loop.running = True
    stackless.tasklet(_loop)()

def stop():
    """Stops the libevent dispatcher tasklet."""
    if not _loop.running: return
    _loop.running = False

def sleep(seconds):
    """Sleeps the current tasklet for the specified number of seconds"""
    ch = stackless.channel()
    event.timeout(seconds, ch.send, None)
    ch.receive()

class ssfile(file):

    def __init__(self, *args, **kwargs):
        self.rch = stackless.channel()
        self.wch = stackless.channel()
        super(ssfile, self).__init__(*args, **kwargs)

    def read(self, *args):
        def cb():
            self.rch.send(super(ssfile, self).read(*args))
        event.read(self.fileno(), cb)
        return self.rch.receive()

    def write(self, *args):
        def cb():
            self.wch.send(super(ssfile, self).write(*args))
        event.write(self.fileno(), cb)
        return self.wch.receive()

stdfile = file
file = ssfile

import socket as stdsocket
class ssocket(stdsocket.socket):

    def __init__(self, *args, **kwargs):
        self.sendch = stackless.channel()
        self.recvch = stackless.channel()
        super(ssocket, self).__init__(*args, **kwargs)

    def send(self, *args, **kwargs):
        def cb():
            self.sendch.send(super(ssocket, self).send(*args, **kwargs))
        event.write(self.fileno(), cb)
        return self.sendch.receive()

    def recv(self, *args, **kwargs):
        def cb():
            self.recvch.send(super(ssocket, self).recv(*args, **kwargs))
        event.read(self.fileno(), cb)
        return self.recvch.receive()

    def accept(self, *args, **kwargs):
        def cb():
            self.recvch.send(super(ssocket, self).accept(*args, **kwargs))
        event.read(self.fileno(), cb)
        return self.recvch.receive()

cheers,
Arnar