Asynchronous Stream Processing in Python

Question

Let's start with a simple example. A HTTP data stream comes in the following format:

MESSAGE_LENGTH, 2 bytes
MESSAGE_BODY, 
REPEAT...

Currently, I use urllib2 to retrieve and process streaming data as below:

length = response.read(2)
while True:
    data = response.read(length)
    DO DATA PROCESSING

It works, but since all messages are in size of 50-100 bytes, the above method limits buffer size each time it reads so it may hurt performance.

Is it possible to use seperate threads for data retrieval and processing?

Answer 1

Yes, can be done and is not that hard, if your format is essentially fixed.

I used it with httplib in Python 2.2.3 and found it had some abysmal performance in the way we hacked it together (basically monkey patching a select() based socket layer into httplib).

The trick is to get the socket and do the buffering yourself, so you do not fight over buffering with the intermediate layers (made for horrible performance when we had httplib buffer for chunked http decoding, the socket layer buffer for read()).

Then have a statemachine that fetches new data from the socket when needed and pushes completed blocks into a Queue.Queue that feeds your processing threads.

I use it to transfer files, checksum (zlib.ADLER32) them in an extra thread and write them to the filesystem in a third thread. Makes for about 40 MB/s sustained throughput on my local machine via sockets and with HTTP/chunked overhead.