ResponseError (on LISTEN [(), {}]): Unknown response type o

[Nick Jennings]

Hey guys, I've had a tornado instance running on our dev server all
summer, with minimal issues. However today I tried to set up a few
more vhosts for developers and spawn of a few more instances
per-vhost. Once I did that I started getting all kinds of strange
errors and exceptions I'd never run into before. it's like the whole
thing just started to fail and I don't know why.

I've been getting this one a lot, seems to happen during the initial
HTTP request... Any ideas whats going on here? Sometimes it says
"Unknown response type n" instead of 'type o'.


Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/tornado/web.py", line
1021, in _stack_context_handle_exception
raise_exc_info((type, value, traceback))
File "/usr/local/lib/python2.7/dist-packages/tornado/stack_context.py",
line 258, in _nested
yield vars
File "/usr/local/lib/python2.7/dist-packages/tornado/stack_context.py",
line 228, in wrapped
callback(*args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",
line 163, in readline_callback
callback(*args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornado/stack_context.py",
line 233, in wrapped
callback(*args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
382, in inner
self.set_result(key, result)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
315, in set_result
self.run()
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 343, in run
yielded = self.gen.throw(*exc_info)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",
line 927, in listen
response = yield gen.Task(self.process_data, data, cmd_listen)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 366, in run
self.yield_point.start(self)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
241, in start
self.func(*self.args, **self.kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
120, in wrapper
runner.run()
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 345, in run
yielded = self.gen.send(next)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",
line 470, in process_data
cmd_line)
ResponseError: ResponseError (on LISTEN [(), {}]): Unknown response type o

[aliane abdelouahab]

it seems it's a database related error (Redis)

[Nick Jennings]

I'm also getting this error almost as much. Do you think it's also
related to Redis?

Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/tornado/web.py", line
1021, in _stack_context_handle_exception
raise_exc_info((type, value, traceback))

File "/usr/local/lib/python2.7/dist-packages/tornado/web.py", line
1139, in wrapper
return method(self, *args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
120, in wrapper
runner.run()

File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 343, in run
yielded = self.gen.throw(*exc_info)

File "/opt/belvedere/notify/listener.py", line 215, in queue_command
yield tornado.gen.Task(self.application.trdb.rpush, job_queue, job)

File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 366, in run
self.yield_point.start(self)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
241, in start
self.func(*self.args, **self.kwargs)

File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",
line 684, in rpush
self.execute_command('RPUSH', key, value, callback=callback)

File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
120, in wrapper
runner.run()
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 345, in run
yielded = self.gen.send(next)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",

line 416, in execute_command
raise e
IndexError: deque index out of range

[aliane abdelouahab]

yes, it seems that because there is a (index out of range) then there
is no answer about the state of the list?! (just suggestion).
there is the gen.py which is for async, do you use the asynchronous
library for redis? am sorry because i dont use Redis, i'm learning
MongoDB and in MongoDB the official driver is blocking (even it is
"shot and forget and needs the safe=True to confirme) and to use the
asnyc there is other libraries like Motor and AsyncMongo.
but from the traceball it is always Redis and Gen which is a problem
related to async calls to Redis DB.
i've found this:
http://stackoverflow.com/questions/11769965/how-to-do-asynchronous-writes-to-redis-in-tornado

[Nick Jennings]

Yeah, I've read that stackoverflow post before. I'm actually using
both redis-py and tornado-redis for different parts of my application,
but these errors keep coming up and it's extremely hard for me to
trace where they are coming from. The Traceback doesn't touch on any
part of the application code, just in the tornado internals, which
make it very hard for me to pintpoint even which redis request is the
one causing the errors.

I use tornado-redis when I need to do something like a
subscribe/listen because that's locking - and instead i can just set a
callback and carry on. In other cases I use the sync redis-py client
because I need to know that "right now" I'm adding something to a list
or popping it off. This take just a second so I don't care so much
about needing to make that quick operation async, and it should be
independant of other instances of IOLoop right?

I'm getting this one all the time now, and I never used to get it
before. Does anyone have any tips for how to track down the problem?

Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/tornado/web.py", line
1021, in _stack_context_handle_exception
raise_exc_info((type, value, traceback))
File "/usr/local/lib/python2.7/dist-packages/tornado/stack_context.py",
line 258, in _nested
yield vars
File "/usr/local/lib/python2.7/dist-packages/tornado/stack_context.py",
line 228, in wrapped
callback(*args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",

line 191, in read_callback

callback(*args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
382, in inner
self.set_result(key, result)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
315, in set_result
self.run()
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 343, in run
yielded = self.gen.throw(*exc_info)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",

line 938, in listen

cmd_listen)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 366, in run
self.yield_point.start(self)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
241, in start
self.func(*self.args, **self.kwargs)
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line
120, in wrapper
runner.run()
File "/usr/local/lib/python2.7/dist-packages/tornado/gen.py", line 345, in run
yielded = self.gen.send(next)
File "/usr/local/lib/python2.7/dist-packages/tornadoredis/client.py",

line 489, in process_data

cmd_line)
ResponseError: ResponseError (on LISTEN [(), {}]): Unknown response type

[Ronan Amicel]

What versions of redis, redis-py, tornado-redis are you using?

--
Ronan Amicel

«« Twitter overload?
»» Get your daily summary at http://focus.io/

[Nick Jennings]

I updated to the latest pip's for redis-py (2.6.2) and tornado-redis
(0.0.7) earlier today, and am still getting the error.
The server:
$ redis-server --version
Redis server version 2.2.12 (00000000:0)

[aliane abdelouahab]

what about library conflict? maybe because u're using two libraries
which are dependent try to use the redis-py alone without invoquing
the tornado-redis methods?

[Nick Jennings]

Does anyone have any suggestions as to how I could subscribe & set a
listen to a few pub/sub channels without blocking the client?

I have a listener and several workers. When the client connects the
listener subscribes to a channel just for that session, and another
channel which is anything for that client (ie. if there are several
connections from the same client). Then when the client sends
commands, the listener will push the command to a job queue (a redis
list), and the workers will pop a job off and process it, when
complete they send the results along the subscription channel
specified in the job. So the listener will get the 'job complete' data
either on the session channel or the user-specific channel, and send
it back to the client,

I can't figure out a way to establish those listeners without blocking
the connection, using the redis-py client.

Anyone have any suggestions?
Thanks
Nick

[aliane abdelouahab]

there is Google's Pubsubhubu, it seems that it will do the job without
using a database alerts.
https://code.google.com/p/pubsubhubbub/
sorry but i dont knoz the concept, but from a presentation, it seems
that they dont use Redis, so it may be a database independent.

[Nick Jennings]

Well, pubsubhubbub doesn't seem to fit all my use-cases, which Redis
does. I need the lists, the hsets and the pub/sub channels for
different purposes.

People have had to run into this problem before? Can't redis be used
in this pretty straight-forward (and I'm sure quite common)
architecture?

It was working fine for us with one listener instance. Now we have 3
and all sorts of bizarre errors are happening and it's really
difficult to debug, or even to know where to begin.

[Ronan Amicel]

On Wed, Sep 12, 2012 at 1:43 PM, Nick Jennings
<nick.sil...@gmail.com> wrote:
> Does anyone have any suggestions as to how I could subscribe & set a
> listen to a few pub/sub channels without blocking the client?
>
> I have a listener and several workers. When the client connects the
> listener subscribes to a channel just for that session, and another
> channel which is anything for that client (ie. if there are several
> connections from the same client). Then when the client sends
> commands, the listener will push the command to a job queue (a redis
> list), and the workers will pop a job off and process it, when
> complete they send the results along the subscription channel
> specified in the job. So the listener will get the 'job complete' data
> either on the session channel or the user-specific channel, and send
> it back to the client,
>
> I can't figure out a way to establish those listeners without blocking
> the connection, using the redis-py client.
>
> Anyone have any suggestions?

Based on what I understand you're trying to do, here is how I would do it:
– start a listener thread that subscribes to a specific Redis channel,
called e.g. 'commands' (using redis-py)
– the main thread runs the IOLoop and handles web requests ; when a
client starts a session, the main thread publishes a message on the
'commands' channel to notify the listener thread
– when receiving this notification message, the listener thread does a
Redis SUBSCRIBE to now also listen to additional channels
– the Redis connection used for pub/sub should NOT be used for other
commands ; if you need to issue Redis commands in the listener threads
(to dispatch jobs to workers), you should use a second connection for
those

Hope this helps.

[Nick Jennings]

Hi Roman, thanks for your reply. Questions below...

On Wed, Sep 12, 2012 at 2:29 PM, Ronan Amicel <ronan....@gmail.com> wrote:
> On Wed, Sep 12, 2012 at 1:43 PM, Nick Jennings
> <nick.sil...@gmail.com> wrote:
>> Does anyone have any suggestions as to how I could subscribe & set a
>> listen to a few pub/sub channels without blocking the client?
>>
>> I have a listener and several workers. When the client connects the
>> listener subscribes to a channel just for that session, and another
>> channel which is anything for that client (ie. if there are several
>> connections from the same client). Then when the client sends
>> commands, the listener will push the command to a job queue (a redis
>> list), and the workers will pop a job off and process it, when
>> complete they send the results along the subscription channel
>> specified in the job. So the listener will get the 'job complete' data
>> either on the session channel or the user-specific channel, and send
>> it back to the client,
>>
>> I can't figure out a way to establish those listeners without blocking
>> the connection, using the redis-py client.
>>
>> Anyone have any suggestions?
>
> Based on what I understand you're trying to do, here is how I would do it:
> – start a listener thread that subscribes to a specific Redis channel,
> called e.g. 'commands' (using redis-py)
> – the main thread runs the IOLoop and handles web requests ; when a
> client starts a session, the main thread publishes a message on the
> 'commands' channel to notify the listener thread

Can we clarify what specifically is the listener thread and the main
thread? If the main thread is the <'tornado.web.Application'> and the
listener thread is the <'tornado.websocket.WebSocketHandler'>, then
how could the main thread handle requests? The listener can only do
that, right?

Would it be possible to write out a very brief psuedo-code example?

> – when receiving this notification message, the listener thread does a
> Redis SUBSCRIBE to now also listen to additional channels
> – the Redis connection used for pub/sub should NOT be used for other
> commands ; if you need to issue Redis commands in the listener threads
> (to dispatch jobs to workers), you should use a second connection for
> those

Well, for the worker queue it's a redis list. The workers pop off a
task, process it, then broadcast it back to whomever is listening on
the pub/sub channels. So only active sessions would receive the
completed jobs. They would look for their session ID in the data of
the procesed job, and only send it back to the client if it was for
them.

[Ronan Amicel]

On Wed, Sep 12, 2012 at 4:26 PM, Nick Jennings
<nick.sil...@gmail.com> wrote:
>>
>> Based on what I understand you're trying to do, here is how I would do it:
>> – start a listener thread that subscribes to a specific Redis channel,
>> called e.g. 'commands' (using redis-py)
>> – the main thread runs the IOLoop and handles web requests ; when a
>> client starts a session, the main thread publishes a message on the
>> 'commands' channel to notify the listener thread
>
> Can we clarify what specifically is the listener thread and the main
> thread? If the main thread is the <'tornado.web.Application'> and the
> listener thread is the <'tornado.websocket.WebSocketHandler'>, then
> how could the main thread handle requests? The listener can only do
> that, right?

It sounds like you only have one thread right now. (A request handler
is not a thread.)

My suggestion is to offload the redis subscriber to a second thread,
as it needs to be blocking, and thus cannot run in the same thread as
the tornado IO loop.

> Would it be possible to write out a very brief psuedo-code example?

Here's some (incomplete) code, but it might help you:
https://gist.github.com/3707121

[Nick Jennings]

Thank you very much for this gist, it actually helped clear up some
misunderstandings I had surrounding the pubsub feature in general. I
will use your example as a basis of refactoring my code and hopefully
this new approach will get rid of the oddities I'm currently
experiencing.

Cheers
Nick

[Nick Jennings]

On Wed, Sep 12, 2012 at 4:48 PM, Ronan Amicel <ronan....@gmail.com> wrote:
> On Wed, Sep 12, 2012 at 4:26 PM, Nick Jennings
> <nick.sil...@gmail.com> wrote:
>>>
>>> Based on what I understand you're trying to do, here is how I would do it:
>>> – start a listener thread that subscribes to a specific Redis channel,
>>> called e.g. 'commands' (using redis-py)
>>> – the main thread runs the IOLoop and handles web requests ; when a
>>> client starts a session, the main thread publishes a message on the
>>> 'commands' channel to notify the listener thread
>>
>> Can we clarify what specifically is the listener thread and the main
>> thread? If the main thread is the <'tornado.web.Application'> and the
>> listener thread is the <'tornado.websocket.WebSocketHandler'>, then
>> how could the main thread handle requests? The listener can only do
>> that, right?
>
> It sounds like you only have one thread right now. (A request handler
> is not a thread.)
>
> My suggestion is to offload the redis subscriber to a second thread,
> as it needs to be blocking, and thus cannot run in the same thread as
> the tornado IO loop.
>

One question, how would the Subscriber thread communicate back to the
WebSocket listener? To pass information back to the client.

[Russ Weeks]

I think there's a race condition in redis-py where the first message on the queue can sometimes be interpreted as the return value of the "subscribe" command.

I'm doing something very similar, with redis pubsub messages being pushed across a websocket.  I took the same approach as Ronan suggested, offloading message subs to a separate thread.  I had to put in a bit of a hack to avoid the race condition.  Gist is here if you're interested: https://gist.github.com/3708161

-Russ

[Ronan Amicel]

On Wed, Sep 12, 2012 at 7:07 PM, Russ Weeks <rwe...@newbrightidea.com> wrote:
> I think there's a race condition in redis-py where the first message on the
> queue can sometimes be interpreted as the return value of the "subscribe"
> command.

I think this issue may have been fixed in redis-py 2.6.0 (or is this
something else?):
https://github.com/andymccurdy/redis-py/blob/master/CHANGES

[Russ Weeks]

Could be!  It seems I'm still on 2.4.  I will definitely check that out, thanks for the tip.

-Russ

[Nick Jennings]

Any ideas about that Ronan? Since it's spawned in a completely
different thread, the only way I can think to communicate with the
WebSocket listener is via. Redis, which would defeat the purpose,
wouldn't it?


On Wed, Sep 12, 2012 at 6:20 PM, Nick Jennings

[Russ Weeks]

Hi, Nick,

One question, how would the Subscriber thread communicate back to the
WebSocket listener? To pass information back to the client.

Your subscriber thread is (1) handling subscription requests from your websocket handlers and (2) responding to message notifications from the redis pubsub object.  In my case, the subscription requests are going to provide a channel identifier and a callback.  So my subscriber thread maintains a dict mapping channel ID to a list of callbacks.  Then when the subscriber thread receives a message notification it maps the channel property of the message to the callback, and invokes the callback like so:

        channel = msg['channel']

        data = msg['data']

        listeners = self._sub_cbs.get(channel,[])

        for listener in listeners: IOLoop.instance().add_callback(functools.partial(listener, data))

In my case, the callback is associated with a WebSocketHandler; it converts the data to JSON and pushes it down the socket.

-Russ

[Nick Jennings]

Hi Russ, thanks for the example. Makes sense, except I'm not sure how
you can pass a callback to a method in a WebSocket instance through
Redis. How are you doing it?

[Russ Weeks]

Hi, Nick,

In the gist example, the _rc_listen function is iterating over all the messages as they become available through the redis pubsub object.  This function is running in its own thread, effectively in an infinite loop.  It calls the _process_msg function on each message, which has logic to process 'subscribe' and 'unsubscribe' messages, and some hacky code to handle the aforementioned race condition which is probably fixed in 2.6.  But generally (the else part of the if condition in _process_msg), the message channel is used to lookup the callbacks for the message, which are then invoked in Tornado's IOLoop.

So, it isn't that I'm passing a callback to a method in a WebSocket instance to redis.  I'm processing the messages the regular way, by pulling them from the redis pubsub object in a blocking manner.  Then I'm invoking the correct websocket callback from the listener thread based on the channel ID.  Because it's going across threads, I have to use IOLoop.instance().add_callback.

Hope that clears things up,

-Russ

[Nick Jennings]

Hi Russ, what does the callback actually look like? I'm still confused
as to whats actually being sent to the add_callback method, what do I
need to make available so that instances in the IOLoop can tell the
external subscribe thread "hey get back in touch with me using
__this__". What is __this__ specifically? Maybe this is covered in the
docs somewhere? I couldn't find anything, but maybe I'm not using the
right search terms.

Other than that, your example makes perfect sense and I think I understand it.
-Nick

[Nick Jennings]

Oh, actually - let me take that back. I was actually referring more to
Ronans example earlier in the thread. He had the subscriber thread
completely outside of the IOLoop, it was listening for commands from
the listener threads and subscribing/unsubscribing to channels based
on those commands. When it got some info that needed to be passed back
to the original listener, it was prototyped to handle that in the
"do_something" function, however I wasn't sure how I could send data,
or issue a callback, to one of these listener threads that was in a
completely different area of the application.

In your example, for each port the application is instantiated on, you
spawn off an internal thread of the _rc_listen() function, correct? I
thought it was not recommended to use threading from within an IOLoop?
So, the class in your gist would be similar to my "Application" class
in this file:

https://bitbucket.org/silverbucket/belvedere/src/b118a32280f5/notify/listener.py

Except I would spawn a thread to handle the pub/sub functionality.
Correct? Then when a new listener is instantiated from a connection,
it would add itself to a self.application._listeners dict so the
application class can reference it and make calls to it?

On line 76:

for listener in listeners:
IOLoop.instance().add_callback(functools.partial(listener, data))

I have two questions:
1) what is the listener? Is it a reference to the listeners 'self' ?
2) where in the listener instance does this call arrive at? which
method in the listener

Thanks!
-Nick


On Thu, Sep 13, 2012 at 1:15 PM, Nick Jennings

[Nick Jennings]

Hi Guys,

Well after some more digging I finally got a working solution. I
ended up using Ronans example as a base. Here's how it's set up:

- When the listeners are created, the subscriber class is forked off
right afterward, and passes the instance of the listener to it during
initialization.

- During all subscribe/unsubscribe commands the websocket listeners
pass their SID as well, which the Subscriber keeps track of in it's
clients list

- when the Subscribe gets a message on the "public" subscription
channels (not the one for back-end communication) it passes it along
to it's listener object which it got during initialization.

- the listener object contains a reference to the instantiated IOLoop
instances, so the Subscribe can make the call:
self.listener.application.callback(sid, data)

- the application class has a record of all the websocket listeners
(they register and unregister themselves during on_open/close and give
the application class a callback function). So when it receives
callback(sid, data) it can then call:
self.listener_callbacks[sid](data)

- and the websocket instance now gets the callback with the data to
send to the client.


You can see the code in these two files:
(this contains the subscribe class, and the initial listener instantiation loop)
https://bitbucket.org/silverbucket/belvedere/src/2f593bf20856/notify/plugin.py

(this contains the application class and the websocket listener class itself)
https://bitbucket.org/silverbucket/belvedere/src/2f593bf20856/notify/listener.py

I've tested it and it's performing much better than previously, I can
have it running on multiple ports, with more load and the results are
consistent and there are so far no odd errors and weirdnesses
happening.
*sigh of relief*
Thanks everyone for your help!
-Nick

On Thu, Sep 13, 2012 at 2:24 PM, Nick Jennings