redis-py is not compatible with select.select()

[mattlong]

In a project involving websockets and redis, select.select() is being used to be notified when either a websocket or redis pub/sub has data ready for reading. For it to work at all, we have to peek under the covers of redis-py a bit to get at the socket's file descriptor and receive the messages:

r = redis.StrictRedis()
pubsub = r.pubsub()

# subscribe to a few channels...

fd = pubsub.connection._sock.fileno()
while True:
    ready = select.select([fd], [], [])
    response = pubsub.parse_response()
    # do something with response

This generally works pretty well. However, we've discovered that when publishing multiple messages to subscribed channels in quick succession, select() tends to only return once for the first message, but not for subsequent messages. After a lot of digging, it turns out this is due to redis-py using buffered reads on its pub/sub socket. In particular, redis.connection.PythonParser.read uses readline, which is a buffered read operation. Several messages are read off the socket by the single call to readline, but all except the first message are buffered until the next time readline (or another read operation) is called. Due to this buffereing, select() will not fire again since there isn't any additional data to be read off the socket.

It would be really powerful if redis-py could be made compatible with select.select(). As a proof of concept, I addressed the issue by monkey patching PythonParser's read method to do unbuffered reads as shown below. This is clearly less efficient than doing buffered reads, but it seems something along these lines would be needed for select() compatibility.

# excerpt from redis.connection.PythonParser.read

...

# no length, read a full line
# readline is buffered, so #badnewsbears
#return self._fp.readline()[:-2]

# let's do it unbuffered instead!
buf = BytesIO()
byte = None
while byte != '\n':
    byte = self._fp.read(1)
    buf.write(byte)
buf.seek(0)
return buf.read()[:-2]

...

Note that this isn't a valid solution by itself since hiredis also does buffered reads...

Does anyone have any thoughts on how redis-py could be adapted in a not-so-hacky way to make it play nice with select.select()?

[andymccurdy]

Hey @mattlong. I did some work to refactor pubsub a while back but never merged it in. It mostly needs tests and I believe there are some edge cases that still fail.

Anyway, I think that it solves your problem. Check out the changes I made to the Connection and Parser classes in order to perform non-blocking reads: f69b909#diff-b3be2d0fa4e0cf1f1c403f2de43ecdb4R65

I'd like to find some time to get that branch merged in. Let me know if you think it'll help your use case.

[jrief]

Any news about this?
It currently blocks this issue jrief/django-websocket-redis#2.

[andymccurdy]

@jrief I was hoping for some feedback on my proposal above. Since I don't actually need this functionality, I wanted to make sure my solution solved someone's problem prior to implementing it :)

[jrief]

@andymccurdy
Actually, I never was able to reproduce that kind of error. Apparently, I don't use websockets for gaming :)
@mattlong
Could you please check if the above patch for py-redis works under your conditions.

[andymccurdy]

@jrief does the above patch fix your bug?

[jrief]

@mattlong reported it.
As I said, I never was able to reproduce that kind of error.

[mattlong]

@andymccurdy, it looks like the intended way to perform non-blocking pubsub would be to use get_message() instead of parse_response(). Is that correct?

If so, for my use case, I would do something like:

r = redis.StrictRedis()
pubsub = r.pubsub()

# subscribe to a few channels...

fd = pubsub.connection._sock.fileno()
while True:
    ready = select.select([fd], [], [])
    while True:
        response = pubsub.get_message()
        if response is None:
            break
        # do something with response

This would still suffer from a race condition since get_message() still uses a buffered read on the socket.

Instead, I could not use select.select() at all and instead just check get_message() and sleep for a tenth of a second or something in a loop. While not ideal from a performance point of view, it would definitely be a step in the right direction.

[andymccurdy]

@mattlong

get_message() does the select.select() polling for you via Connection.can_read(). can_read() also does extra work to see if there's anything remaining the buffer.

With PythonParser, it examines the underlying file descriptor's buffer via .tell().

With the HiRedisParser, it uses the parser's .gets() routine and caches that value in the parser. That value is then directly returned via read_response().

So you should simply have to do:

r = redis.StrictRedis()
pubsub = r.pubsub()
# subscribe to some channels...

while True:
    message = pubsub.get_message()
    if message is None:
        continue
    # handle the message

[andymccurdy]

@mattlong I should mention that get_message() never blocks. You can add a time.sleep() to your loop if you desire it.

[mattlong]

Thanks for the explanation!

This is even simpler :) This seems like a reasonable solution to me. I'll try to apply your patch later tonight and see if everything works out alright...

[andymccurdy]

@mattlong Great, thanks! I just merged the latest master into the pubsub branch.

There are still a few pubsub related things I want to fix, but validating that is fixes this issue has been the biggest blocker.

[jrief]

@mattlong
Could you please describe how this error can be reproduced, with the old version of redis-py.
Maybe, then I can add a unit test to avoid regressions in the future.

[andymccurdy]

Hey guys,

I just pushed a bunch more changes to the pubsub branch in 97c50cc. I think I have all the kinks worked out. Still have to write some more tests but it's already significantly better.

[andymccurdy]

@mattlong @jrief

I just pushed what I believe are the final changes/fixes/tests to the pubsub branch. I'm of the belief that everything works as intended now. There's a lot of new bonus features too ;)

Would love to get some feedback before I merge to master if either of you have time.

[andymccurdy]

I added some docs to the README about PubSub. See here: https://github.com/andymccurdy/redis-py/blob/pubsub/README.rst

[ntki]

Hi all!

Please consider changing from select.select() to select.poll(). If you have a long running app that handles reconnections and all that, the underlying filedescriptor number of the socket can be 1024>= and select could easily crash because of that.
(in libhiredis too they changed that between 0.10-0.11)

Br

[andymccurdy]

@ntki select.poll is sadly not available on OSX.

[andymccurdy]

@ntki the implementation likely needs to be based on what capabilities the system has, e.g., select.poll, select.epoll, select.kqueue, select.kevent, select.select, etc.

I wonder if there's a library that abstracts these.

[ntki]

@andymccurdy
Well, I cannot find any. My guess would be just to use a simple if/elif sys.platform to choose.

[ntki]

@andymccurdy
Or even better: just try hasattr(select, "poll"), and fall back to select if False. Although it is probably worth checking for bugs in select with high FD on OSX too.

[andymccurdy]

redis-py 2.10 is out with the new pubsub system.

@ntki I've created a separate issue (#486) to track the change of moving from select.select() to something better.