Writing a server with Python's asyncore module

The Python asyncore and aynchat modules

The Python standard library provides two modules---asyncore and
asynchat---to help in writing concurrent network servers using
event-based designs. The documentation does not give good examples,
so I am making some notes.

Overview

The basic idea behind the asyncore module is that:

• there is a function, asyncore.loop() that does select() on a bunch of 'channels'. Channels are thin wrappers around sockets.
• when select reports an event on any socket, loop() examines the event and the socket's state to create a higher level event;
• it then calls a method on the channel corresponding to the higher level event.

asyncore provides a low-level, but flexible API to build network
servers. asynchat builds upon asyncore and provides an API that is
more suitable for request/response type of protocols.

aysncore

The asyncore module's API consists of:

• the loop method, to be called by a driver program;
• the dispatcher class, to be subclassed to do useful stuff. The dispatcher class is what is called 'channel' elsewhere.

+-------------+           +--------+
| driver code |---------> | loop() |
+-------------+           +--------+
      |                      |
      |                      | loop-dispatcher API (a)
      |                      |
      |                  +--------------+
      |                  | dispatcher   |
      +----------------->| subclass     |
                         +--------------+
                             |
                             | dispatcher-logic API (b)
                             |
                         +--------------+
                         | server logic |
                         +--------------+

This is all packaged nicely in an object oriented way. So, we have
the dispatcher class, that extends/wraps around the socket class (from
the socket module in the Python standard library). It provides all
the socket class' methods, as well as methods to handle the higher
level events. You are supposed to subclass dispatcher and implement
the event handling methods to do something useful.


The loop-dispatcher API

The loop function looks like this:

loop( [timeout[, use_poll[, map[,count]]]])

What is the map? It is a dictionary whose keys are the
file-descriptors, or fds, of the socket (i.e., socket.fileno()), and
whose values are the dispatcher objects.

When we create a dispatcher object, it automatically gets added to a
global list of sockets. The loop() function does a select() on this
list unless we provide an explicit map. (Hmm... we might always want
to use explicit maps; then our loop calls will be thread safe and we
will be able to launch multiple threads, each calling loop on
different maps.)

Methods a dispatcher subclass should implement

loop() needs some methods from the dispatcher object:

• readable(): should return True, if you want the fd to be observed for read events;
• writable(): should return True, if you want the fd to be observed for write events;

If either read or write is true, the corresponding fd will be examined
for errors also. Obviously, it makes no sense to have a dispatcher
which returns False for both readable() and writable().

• handle_read: socket is readable; dispatcher.recv() can be used to actually get the data
• handle_write: socket is writable; dispatcher.send(data) can be used to actually send the data
• handle_error: socket encountered an error
• handle_expt: socket received OOB data (not really used in practice)
• handle_close: socket was closed remotely or locally

Server sockets get one more event.

• handle_accept: a new incoming connection can be accept()ed. Call the accept() method really accept the connection. To create a server socket, call the bind() and listen() methods on it first.
  

Client sockets get this event:

• handle_connect: connection to remote endpoint has been made. To initiate the connection, first call the connect() method on it.

Other socket methods are available in dispatch: create_socket(),
close(), set_resue_addr().


How to write a server using asyncore

The standard library documentation gives a client example, but not a
server example. Here are some notes on the latter.

1. Subclass dispatched to create a listening socket
2. In its handle_accept method, create new dispatchers. They'll get added to the global socket map.

Note: the handlers must not block or take too much time... or the
server won't be concurrent.

These socket-like functions that dispatcher extends should not be bypassed. They do funky things to detect higher level events. For e.g., how does asyncore figure out that the socket is closed? If I remember correctly, there are two ways to detect whether a non-blocking socket is closed:

• select() returns a read event, but when you call recv()/read() you get zero bytes;
• you call send()/write() and it fails with an error (sending zero bytes is not an error).

(I wish I had a copy of Unix Network Programming by Stevens handy
right now.)

Will look at asynchat in another post.

The code for the server is below:


asyncore_echo_server.py

import logging
import asyncore
import socket

logging.basicConfig(level=logging.DEBUG, format="%(created)-15s %(msecs)d %(levelname)8s %(thread)d %(name)s %(message)s")
log                     = logging.getLogger(__name__)

BACKLOG                 = 5
SIZE                    = 1024

class EchoHandler(asyncore.dispatcher):

    
    def __init__(self, conn_sock, client_address, server):
        self.server             = server
        self.client_address     = client_address
        self.buffer             = ""

        # We dont have anything to write, to start with
        self.is_writable        = False

        # Create ourselves, but with an already provided socket
        asyncore.dispatcher.__init__(self, conn_sock)
        log.debug("created handler; waiting for loop")

    def readable(self):
        return True     # We are always happy to read


    def writable(self):
        return self.is_writable # But we might not have
                                # anything to send all the time


    def handle_read(self):
        log.debug("handle_read")
        data = self.recv(SIZE)
        log.debug("after recv")
        if data:
            log.debug("got data")
            self.buffer += data
            self.is_writable = True  # sth to send back now
        else:
            log.debug("got null data")

    def handle_write(self):
        log.debug("handle_write")
        if self.buffer:
            sent = self.send(self.buffer)
            log.debug("sent data")
            self.buffer = self.buffer[sent:]
        else:
            log.debug("nothing to send")
        if len(self.buffer) == 0:
            self.is_writable = False


    # Will this ever get called?  Does loop() call
    # handle_close() if we called close, to start with?
    def handle_close(self):
        log.debug("handle_close")
        log.info("conn_closed: client_address=%s:%s" % \
                     (self.client_address[0],
                      self.client_address[1]))
        self.close()
        #pass

            
class EchoServer(asyncore.dispatcher):

    allow_reuse_address         = False
    request_queue_size          = 5
    address_family              = socket.AF_INET
    socket_type                 = socket.SOCK_STREAM


    def __init__(self, address, handlerClass=EchoHandler):
        self.address            = address
        self.handlerClass       = handlerClass

        asyncore.dispatcher.__init__(self)
        self.create_socket(self.address_family,
                               self.socket_type)

        if self.allow_reuse_address:
            self.set_resue_addr()

        self.server_bind()
        self.server_activate()


    def server_bind(self):
        self.bind(self.address)
        log.debug("bind: address=%s:%s" % (self.address[0], self.address[1]))


    def server_activate(self):
        self.listen(self.request_queue_size)
        log.debug("listen: backlog=%d" % self.request_queue_size)


    def fileno(self):
        return self.socket.fileno()


    def serve_forever(self):
        asyncore.loop()


    # TODO: try to implement handle_request()

    # Internal use
    def handle_accept(self):
        (conn_sock, client_address) = self.accept()
        if self.verify_request(conn_sock, client_address):
            self.process_request(conn_sock, client_address)


    def verify_request(self, conn_sock, client_address):
        return True


    def process_request(self, conn_sock, client_address):
        log.info("conn_made: client_address=%s:%s" % \
                     (client_address[0],
                      client_address[1]))
        self.handlerClass(conn_sock, client_address, self)


    def handle_close(self):
        self.close()

and to use it:

    server = asyncore_echo_server.EchoServer((interface, port))
    server.serve_forever()