Cothread Architecture Notes#

Some notes on key architectural issues in the cothread scheduler.

Coroutine Switching#

There are precisely 7 calls to switch in the cothread library, and it’s crucial that they work together correctly. These calls are (in order of appearance in the code).

During scheduler creation (in the .create method of _Scheduler) we create a separate scheduler task and switch to that. This should be called from the main parent task (ie, when it ends, the application terminates).
As soon as the scheduler has been initialised, control is returned to the main task. This is in .__scheduler, which carries a top level loop for restarting the scheduler.
If the main scheduler loop .__schedule_loop raises an exception, control is switched to the main parent task with a special .__WAKEUP_INTERRUPT return code. The main task will at this point normally be waiting for a normal scheduler wake up.
The .__tick dispatcher switches to all ready tasks in turn passing a wakeup code, either .__WAKEUP_NORMAL or .__WAKEUP_TIMEOUT.
If .poll_scheduler has been called, and thus .__poll_callback is set, the routine .__poll_suspend will switch back to the suspended .poll_scheduler call. In this case special polling arguments are passed (this should be a list of (descriptor, flags) pairs, together with a timeout), and a list of active descriptors should be returned.
.poll_scheduler defines the other side of this switch: a list of ready descriptors is passed in a switch to the scheduler task, and the polling arguments described above are returned from the .poll_scheduler call – unless a .__WAKEUP_INTERRUPT switch has occurred!
The normal place where control is switched to the scheduler is in the middle of .wait_until. We have to pass an empty list (to be compatible with the .poll_scheduler switch), and a wakeup reason should be returned.

Callback#

The new Callback mechanism raises some issues.

It is possible for the callback queue to grow without limit, and this is invisible to the poor users. This can happen if too much time is spent in processing updates.
The callback queue is processed without yielding. If the callback queue is very long this can cause unresponsive behaviour. Unfortunately inserting a yield on every callback is surprisingly costly.
We lack any mechanism for observing what’s going on inside the library.

Some notes for emulating select on Windows#

Want to look at libevent. I suspect this, being a network API, is largely socket based, but there may be clues. Source file win32select.c, function win32_dispatch(), looks vaguely interesting. Also poll.c.

Also should look at twistedmatrix.com, and there’s a further page of notes in poll_win32.py

Anyhow, building on Windows is a pain and I dislike the platform anyway…

Questions on Stack Overflow.#

http://stackoverflow.com/questions/3021424/select-on-a-named-pipe

The answer here is use the named pipe APIs using the overlapped I/O model and WaitForMultipleObjects(), example

http://msdn.microsoft.com/en-us/library/aa365603(VS.85).aspx

http://stackoverflow.com/questions/3911799/windows-poll-or-select-on-named-pipe

Answer is need to use overlapped I/O or I/O completion ports, references

http://en.wikipedia.org/wiki/Iocp

Another interesting suggestion is to use socketpair(2) instead of pipe(2) to create the asynchronous event notifier in _Callback and ThreadedEventQueue. This still leaves stdin handling for the input hook, but at least would restore some level of function. The socketpair function is in the Python socket module.

Ho ho ho. Turns out that Windows doesn’t have socketpair anyway! There’s a reasonable looking example here:

http://code.activestate.com/recipes/525487-extending-socketsocketpair-to-work-on-windows/

which does this:

def SocketPair(family = AF_INET; type_ = SOCK_STREAM; proto = IPPROTO_IP):
    listensock = socket(family, type_, proto)
    listensock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    listensock.bind( ('localhost', 0) )
    iface, ephport = listensock.getsockname()
    listensock.listen(1)

    sock1 = socket(family, type_, proto)
    connthread = threading.Thread(target=pairConnect, args=[sock1, ephport])
    connthread.setDaemon(1)
    connthread.start()
    sock2, sock2addr = listensock.accept()
    listensock.close()
    return (sock1, sock2)

def pairConnect(sock, port):
    sock.connect( ('localhost', port) )