pika/pika — Issue #107: Socket exception handling is wrong
Author: za-creature
Created: 2011-11-21 17:36:21+00:00
State: open

We've had some issues with infinite loops when using transactions (or any other blocking rpc for that matter):

In adapters/blocking_connection.py, the BlockingConnection class has the following method:

def _flush_outbound(self):
try:
self._handle_write()
self._socket_timeouts = 0
except socket.timeout:
self._socket_timeouts += 1
if self._socket_timeouts > SOCKET_TIMEOUT_THRESHOLD:
log.error(SOCKET_TIMEOUT_MESSAGE)
self._handle_disconnect()

The issue with this is that it internally calls socket.send(). which can succeed on a dropped connection. From the unix man page for send:

No indication of failure to deliver is implicit in a send(). Locally detected errors are indicated by a return value of -1.

It also never raises a timeout error; writes always succeed, even without a network interface: we tested connect, then ifdown eth0, then send and it returned the number of bytes written, 12 in our case. Had the same results when pulling out the lan cable, or with service network stop. The file descriptor in /proc is a broken symlink, however socket send still works on it.
This is because send just pushes the buffer onto the tcp stack buffer, it never sends the data on the network (not even with TCP_NODELAY set).
Instead recv should be used to determine a socket's state, and indeed, whenever recv returns 0, python raises an exception.

The problem is that pika's BlockingConnection uses the same counter for both read and write errors, and it resets that counter after a successful write (which as I said before, I'm inclined to believe is always the case)

So the following code from BlockingConnection will enter an infinite loop:

def process_data_events(self):
# Make sure we're open, if not raise the exception
if not self.is_open:
raise AMQPConnectionError #this needs _handle_disconnect to trigger so it can change

# Write our data
self._flush_outbound() # this write will always suceed, even though most of the times it's called with a 0 length buffer

# Read data
try:
self._handle_read() # will fail
self._socket_timeouts = 0
except socket.timeout:
self._socket_timeouts += 1 # will always be 1, since it's reset on _flush_outbound
if self._socket_timeouts > SOCKET_TIMEOUT_THRESHOLD: # never happens unless threshold is 1
log.error(SOCKET_TIMEOUT_MESSAGE)
self._handle_disconnect()

While BlockingConnection is deprecated, I'm reporting this because other adapters might be affected by the same behavior.