Python kqueue with multiple events

Question

I'm trying to write a little program to send and receive UDP traffic and receive commands via an HTTP interface. The HTTP server lives in one multiprocessing.Process; the UDP server lives in another. The two processes communicate via a python multiprocessing.Pipe. I've attached the complete code below.

I have 2 related problems:

• How do I handle multiple file descriptors/kevents with kqueue in python (socket file descriptor works, pipe file descriptor doesn't appear to- not sure if the pipe I'm using is equivalent to a file)?
• How do I differentiate between these kevents so I can apply different functions when the pipe is to be read vs the socket?

Pseudo code for what I'd like my UDP server to do:

kq = new kqueue
udpEvent = kevent when socket read
pipeEvent = kevent when pipe read
while:
    for event in kq.conrol([udpEvent, pipeEvent]):
        if event == udpEvent:
             # do something
        elif event == pipeEvent:
             print "HTTP command via pipe:", pipe.recv()

Right now, the UDP server recognizes socket events and reads off the socket correctly. However, when I add the pipe kevent to the kqueue, the program spits out pipe events nonstop. I'm setting the filter as pipe has been written, but I assume either 1) this is wrong 2) more specifically, the python multiprocessing.Pipe is like a regular unix pipe and needs to be handled differently.

.....
<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 udata=0x4000000000000>
<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 udata=0x4000000000000>
<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 udata=0x4000000000000>
<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 udata=0x4000000000000>
<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 udata=0x4000000000000>
<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 ^C<select.kevent ident=4297866384 filter=-29216 flags=0x4000 fflags=0x1 data=0x16 udata=0x4000000000000>

---

main.py

import sys
from multiprocessing import Process, Pipe
# from userinterface import OSXstatusbaritem # use like so: OSXstatusbaritem.start(pipe)
from server import Server
import handler # UI thingy

# For UI, use simple HTTP server with various endpoints
# open a connection: localhost:[PORT]/open/[TARGET_IP]

def startServer(pipe):
    UDP_IP = "127.0.0.1"
    UDP_PORT = 9000

    print "starting server"
    s = Server(pipe)
    s.listen(UDP_IP, UDP_PORT)
    print "finishing server"

import BaseHTTPServer
def startUI(pipe):
    HTTP_PORT = 4567
    server_class = BaseHTTPServer.HTTPServer
    myHandler = handler.handleRequestsUsing(pipe)
    httpd = server_class(('localhost', 4567), myHandler)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
          pass
    httpd.server_close()

def main():
    # Named full duplex pipe for communicating between server process and UI
    pipeUI, pipeServer = Pipe()

    # Start subprocesses
    pServer = Process(target=startServer, args=(pipeServer,))
    pServer.start()
    startUI(pipeUI)
    pServer.join()

if __name__ == "__main__": sys.exit(main())

server.py (UDP)

import sys
import select # for kqueue
from socket import socket, AF_INET, SOCK_DGRAM
from multiprocessing import Process, Pipe

class Server:
    def __init__(self, pipe):
        self.pipe = pipe

    def listen (self, ipaddress, port):
        print "starting!"

        # Initialize listening UDP socket
        sock = socket(AF_INET, SOCK_DGRAM)
        sock.bind((ipaddress, port))

        # Configure kqueue
        kq = select.kqueue()
        # Event for UDP socket data available
        kevent0 = select.kevent( sock.fileno(),
                                 filter=select.KQ_FILTER_READ,
                                 flags=select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR)
        # Event for message queue from other processes (ui)
        kevent1 = select.kevent( self.pipe.fileno(),
                                 filter=select.KQ_FILTER_WRITE,
                                 flags=select.KQ_EV_ADD | select.KQ_EV_ENABLE)                        

        # TODO: Figure out how to handle multiple kevents on kqueue
        # TODO: Need an event for TUN data

        # Start kqueue      
        while True:
            revents = kq.control([kevent0, kevent1], 1, None)
            for event in revents:
                print event
        kq.close()
        # close file descriptors (os.close(fd))

handler.py (HTTP interface)

import BaseHTTPServer

# Simple HTTP endpoints for controlling prototype Phantom implementation.
# The following commands are supported:
# 1. Open a connection via /open/[IP]:[PORT]
# 2. ????

class RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    pipe = None

    def __init__(self, pipe, *args):
        RequestHandler.pipe = pipe
        BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args)

    def do_HEAD(s):
        s.send_response(200)
        s.send_header("Content-type", "application/json")
        s.end_headers()
    def do_GET(s):
        s.send_response(200)
        s.send_header("Content-type", "application/json")
        s.end_headers()

        # Open connection command
        if s.path.startswith('/open/'):
            addrStr = s.path[6:len(s.path)]
            (address, port) = tuple(filter(None, addrStr.split(':')))
            port = int(port)
            print "opening address: ", address, "port:", port
            RequestHandler.pipe.send(['open', address, port])

def handleRequestsUsing(logic):
    return lambda *args: RequestHandler(logic, *args)

---

UPDATE:

I rewrote the server listen method with select. For a slow little python prototype that won't use more than 3 or 4 fds, speed doesn't matter anyway. Kqueue will be the subject for another day.

def listen (self, ipaddress, port): print "starting!"

# Initialize listening non-blocking UDP socket
sock = socket(AF_INET, SOCK_DGRAM)
sock.setblocking(0)
sock.bind((ipaddress, port))

inputs = [sock, self.pipe] # stuff we read
outputs = [] # stuff we expect to write
while inputs:
    readable, writable, exceptional = select.select(inputs, outputs, inputs)

    for event in readable:
        if event is sock:
            self.handleUDPData( sock.recvfrom(1024) )
        if event is self.pipe:
            print "pipe event", self.pipe.recv()

Answer 1

I know this is an old question, but I can give you an example of kqueue socket polling that I am using for a multithreaded HTTP server, that I figured out after reading C source code and the man pages for kqueue.

#bsd socket polling
#I make all the relevant flags more C like to match the kqueue man pages
from select import kevent, kqueue
from select import KQ_EV_ADD as EV_ADD, KQ_EV_ONESHOT as EV_ONESHOT   
from select import KQ_EV_EOF as EV_EOF

from .common import Client_Thread #a parent class who's implementation is irrelevant to the question, lol

class BSD_Client(Client_Thread):
    def __init__(self, *args):
        Client_Thread.__init__(self, *args)
        #Make a kqueue object for the thread 
        kq = kqueue()
        #Make a one-shot kev for this kqueue for when the kill socket is
        #connected to. The connection is only made once, so why not tell
        #that to our kqueue? The default filter is EVFILT_READ, so we don't
        #need to specify that. The default flag is just EV_ADD.
        kill_kev = kevent(self.kill_fd, flags=EV_ADD|EV_ONESHOT)
        #using defaults for the client socket.
        client_kev = kevent(self.client_sock)
        #we only need to keep track of the kqueue's control func.
        #This also makes things prettier in the run func.
        self.control = kq.control
        #now, we add thel list of events we just made to our kqueue.
        #The first 0 means we want a list of at most 0 length in return.
        #the second 0 means we want no timeout (i.e. do this in a 
        #non-blocking way.) 
        self.control([client_kev, kill_kev], 0, 0)

    def run(self):
        while True:
            #Here we poll the kqueue object.
            #The empty list means we are adding no new events to the kqueue.
            #The one means we want a list of at most 1 element. Then None
            #Means we want block until an event is triggered.
            events = self.control([], 1, None)
            #If we have an event, and the event is for the kill socket 
            #(meaning somebody made a connection to it), then we break the 
            #loop and die.
            if events and events[0].ident == self.kill_fd:
                self.die()
                break
            #If all that is left is an EOF in our socket, then we break
            #the loop and die. Kqueues will keep returning a kevent
            #that has been read once, even when they are empty.
            if events and events[0].flags & EV_EOF:
                self.die()
                break
            #Finally, if we have an event that isn't for the kill socket and
            #does not have the EOF flag set, then there is work to do. If
            #the handle client function (defined in the parent class) returns
            #1, then we are done serving a page and we can die.
            if events and self.handle_client():
                self.die()
                break
        client.close()

All self.die does is puts the clients ip:port string onto a Queue used for messaging. A different thread gets that string from the queue, prints a message and joins the relevant thread object. Of course, I am not using an pipes for this, only sockets. I did find this on an online manpage for kqueue though

Fifos, Pipes
Returns when the there is data to read; data contains the number of
bytes available.

When the last writer disconnects, the filter will set EV_EOF in
flags. This may be cleared by passing in EV_CLEAR, at which point the
filter will resume waiting for data to become available before re-
turning

So perhaps in your udp server, where you loop through the revents list, you should do as the man page says? Actually, you don't even need to loop through a list that is a most 1 long. Perhaps your listen function should look something like this...

def listen(self, ip, port):
    print "Starting!"
    sock = socket.socket(AF_INET, SOCK_DGRAM)
    sock.bind((ip, port))
    kq = select.kqueue()
    kev0 = select.kevent(sock)
    kev1 = select.kevent(self.pipe)
    kq.control([kev0, kev1], 0, 0)
    while True: #this loop never breaks! so this whole function blocks forever like this
        revents = kq.control([], 1, None)
        if revents:
            event = revents[0]
            if event.flags & select.KQ_EV_EOF:
                new_event = select.kevent(event.ident, flags=select.KQ_EV_CLEAR)
                kq.control([new_event], 0, 0)
            else:
                print event

I really recommend importing the flags and functions the way I do though, It makes it more similar to the C based manpages you will have to compare to, and I think it looks prettier. I also want to point out that my class is a bit different from what you have, because each new client is going to get an instance of this, and each will run in it's own thread.