Race condition on subscribe

[adrianriley]

Looks like there is a race condition in the Jedis publish/subscribe implementation. To set up a subscription, call JedisConnection.subscribe(). That thread reads the responses from subsequent subscribe/unsubscribe commands sent on the same connection using JedisSubscription in order to maintain its view of the current subscription. But before the read on the connection's input stream, that does a flush on the output stream. See JedisPubSub.process(), which calls Connection.getObjectMultiBulkReply, which calls flush().

Another thread may concurrently be using JedisSubscription to send a command on the same stream. The result may be that the command is sent twice, or a corrupted command is sent.

To reproduce the problem, set up a subscription using JedisConnection.subscribe, then have many threads calling JedisSubscription.subscribe, JedisConnection.unsubscribe. Eventually the race condition will occur.

[adrianriley]

I found this commit comment:

commit f11c1622de1d529cad509b7793977714229f4883
Author: Darren Shepherd 
Date:   Thu Jan 16 21:57:22 2014 -0700
    Allow safe multi-threaded access to JedisPubSub
    
    If Thread A calls a subscribe method on Jedis it will block on a socket read
    call waiting for messages or subscription notifications.  Thread B is now free
    to call additional methods on JedisPubSub to change the current subscriptions
    that thread A is waiting for.  Essentially Thread A will do reads on the
    socket and Thread B will do writes.
    
    An issue occurs in that while Thread A is doing reads, in the
    getObjectMultiBulkReply() method there is an implicit flush() call.  This
    means both Thread A and Thread B may do a write to the socket.  Under this
    situation if Thread A does a flush while Thread B is writing the internal
    buffer will be corrupted.  The fix is to make thread A never call flush().
    This allows Thread A to be solely reads and Thread B to be solely writes.
    
    Additionally since Thread B is sending commands, the internal pipeline count
    is incremented and never decremented.  So when Thread A terminates it's read
    it resets the pipeline count.

That pretty much says the same thing as I'm saying, but in fact the code change on that commit has not removed the call to flush().

[itugs]

This issue is resolved by #998.
Above PR(#509) only fixed JedisPubSub and #998 also fixed BinaryJedisPubSub
@marcosnils Please confirm this.

[marcosnils]

@itugs I don't think this issue is resolved by #998. #998 only fixes sequentially usage but in order to make it reliably multi-threaded #997 / #196 must be implemented.

[itugs]

@marcosnils OK, I see. I thought this is related with #509.
Then we may merge this with #997.

[marcosnils]

Moved to #997