kqueue_hub.py

import sys
import time
import traceback
import select

from eventlet.support import greenlets as greenlet
from eventlet.hubs.hub import BaseHub, FdListener, READ, WRITE

WAIT_TIMER_INDENT = 0x10001234

class Hub(BaseHub):
    def __init__(self, *args, **kwds):
        super(Hub, self).__init__(*args, **kwds)
        
        self.kqueue = select.kqueue()
    
    def add(self, evtype, fileno, cb):
        listener = super(Hub, self).add(evtype, fileno, cb)
        
        changes = []
        if evtype == self.READ:
            changes.append(
                select.kevent(fileno, select.KQ_FILTER_READ, select.KQ_EV_ADD)
            )
        if evtype == self.WRITE:
            changes.append(
                select.kevent(fileno, select.KQ_FILTER_WRITE, select.KQ_EV_ADD)
            )

        self.kqueue.control(changes, 0)

        return listener
    
    def remove(self, listener):
        super(Hub, self).remove(listener)
        
        if listener.evtype == self.READ:
            filter = select.KQ_FILTER_READ
        elif listener.evtype == self.WRITE:
            filter = select.KQ_FILTER_WRITE
        else:
            return
        
        self.kqueue.control([select.kevent(listener.fileno, filter, select.KQ_EV_DELETE)], 0)

    def remove_descriptor(self, fileno):
        super(Hub, self).remove_descriptor(fileno)

        changes = []
        if fileno in self.listeners[self.READ]:
            changes.append(
                select.kevent(fileno, select.KQ_FILTER_READ, select.KQ_EV_DELETE)
            )
        if fileno in self.listeners[self.WRITE]:
            changes.append(
                select.kevent(fileno, select.KQ_FILTER_WRITE, select.KQ_EV_DELETE)
            )

        self.kqueue.control(changes, 0)


    def wait(self, seconds=None):
        readers = self.listeners[self.READ]
        writers = self.listeners[self.WRITE]
        
        changelist = []
        if seconds:
            wait = int(seconds * 1000)
            if wait <= 0:
                wait = 1
            changelist.append(
                select.kevent(
                    WAIT_TIMER_INDENT, select.KQ_FILTER_TIMER,
                    select.KQ_EV_ADD | select.KQ_EV_ONESHOT,
                    data=wait
                )
            )
        
        events = self.kqueue.control(changelist, 10)
        
        for event in events:
            if event.filter == select.KQ_FILTER_TIMER:
                assert event.ident == WAIT_TIMER_INDENT
                continue

            try:
                if event.filter == select.KQ_FILTER_READ:
                    readers[event.ident][0](event.ident)
                if event.filter == select.KQ_FILTER_WRITE:
                    writers[event.ident][0](event.ident)
            except KeyError:
                pass
        
            if event.flags & select.KQ_EV_EOF:
                self.remove_descriptor(event.ident)