write starvation fixes

pika/adapters/base_connection.py

@@ -66,6 +66,7 @@ def __init__(self,
         self.socket = None
         self.stop_ioloop_on_close = stop_ioloop_on_close
         self.write_buffer = None
+        self.buffer_hist = None
         super(BaseConnection, self).__init__(parameters,
                                              on_open_callback,
                                              on_open_error_callback,
@@ -105,7 +106,7 @@ def remove_timeout(self, timeout_id):
         self.ioloop.remove_timeout(timeout_id)
 
     def _adapter_connect(self):
-        """Connect to the RabbitMQ broker, returning True if connected
+        """Connect to the RabbitMQ broker, returning True if connected.
 
         :rtype: bool
 
@@ -123,6 +124,8 @@ def _adapter_connect(self):
         for sock_addr in addresses:
             error = self._create_and_connect_to_socket(sock_addr)
             if not error:
+                # Make the socket non-blocking after the connect
+                self.socket.setblocking(0)
                 return None
         # Failed to connect
         return error
@@ -239,7 +242,10 @@ def _get_error_code(self, error_value):
         return None
 
     def _flush_outbound(self):
-        """Call the state manager who will figure out that we need to write."""
+        """write early, if the socket will take the data why not get it out
+        there asap.
+        """
+        self._handle_write()
         self._manage_event_state()
 
     def _handle_disconnect(self):
@@ -301,6 +307,12 @@ def _handle_error(self, error_value):
         # Disconnect from our IOLoop and let Connection know what's up
         self._handle_disconnect()
 
+    def _handle_timeout(self):
+        """Handle a socket timeout in read or write.
+        We don't do anything in the non-blocking handlers because we
+        only have the socket in a blocking state during connect."""
+        pass
+
     def _handle_events(self, fd, events, error=None, write_only=False):
         """Handle IO/Event loop events, processing them.
 
@@ -337,9 +349,19 @@ def _handle_read(self):
                 data = self.socket.read(self._buffer_size)
             else:
                 data = self.socket.recv(self._buffer_size)
+
         except socket.timeout:
-            raise
+            self._handle_timeout()
+            return 0
+
+        except ssl.SSLWantReadError:
+            # ssl wants more data but there is nothing currently
+            # available in the socket, wait for it to become readable.
+            return 0
+
         except socket.error as error:
+            if error.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
+                return 0
             return self._handle_error(error)
 
         # Empty data, should disconnect
@@ -352,18 +374,35 @@ def _handle_read(self):
         return len(data)
 
     def _handle_write(self):
-        """Handle any outbound buffer writes that need to take place."""
+        """Try and write as much as we can, if we get blocked requeue 
+        what's left"""
         bytes_written = 0
-        if self.outbound_buffer:
-            frame = self.outbound_buffer.popleft()
-            try:
-                self.socket.sendall(frame)
-                bytes_written = len(frame)
-            except socket.timeout:
-                raise
-            except socket.error as error:
+        try:
+            while self.outbound_buffer:
+                frame = self.outbound_buffer.popleft()
+                bw = self.socket.send(frame)
+                bytes_written += bw
+                if bw < len(frame):
+                    LOGGER.debug("Partial write, requeing remaining data")
+                    self.outbound_buffer.appendleft(frame[bw:])
+                    break
+
+        except socket.timeout:
+            # Will only come here if the socket is blocking
+            LOGGER.debug("socket timeout, requeuing frame")
+            self.outbound_buffer.appendleft(frame)
+            self._handle_timeout()
+
+        except socket.error as error:
+            if error.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
+                LOGGER.debug("Would block, requeuing frame")
+                self.outbound_buffer.appendleft(frame)
+            else:
                 return self._handle_error(error)
-        return bytes_written
+
+        #LOGGER.debug("wrote %s bytes", bytes_written)
+        return bytes_written     
+
 
     def _init_connection_state(self):
         """Initialize or reset all of our internal state variables for a given

pika/adapters/blocking_connection.py

@@ -241,8 +241,6 @@ def process_data_events(self):
                 self._socket_timeouts = 0
         except AttributeError:
             raise exceptions.ConnectionClosed()
-        except socket.timeout:
-            self._handle_timeout()
         self._flush_outbound()
         self.process_timeouts()
 
@@ -373,11 +371,8 @@ def _check_state_on_disconnect(self):
     def _flush_outbound(self):
         """Flush the outbound socket buffer."""
         if self.outbound_buffer:
-            try:
-                if self._handle_write():
-                    self._socket_timeouts = 0
-            except socket.timeout:
-                return self._handle_timeout()
+            if self._handle_write():
+                self._socket_timeouts = 0
 
     def _on_connection_closed(self, method_frame, from_adapter=False):
         """Called when the connection is closed remotely. The from_adapter value

pika/adapters/tornado_connection.py

@@ -2,6 +2,8 @@
 from tornado import ioloop
 import logging
 import time
+import socket
+import errno
 
 from pika.adapters import base_connection
 
@@ -47,6 +49,7 @@ def __init__(self, parameters=None,
         """
         self.sleep_counter = 0
         self.ioloop = custom_ioloop or ioloop.IOLoop.instance()
+
         super(TornadoConnection, self).__init__(parameters,
                                                 on_open_callback,
                                                 on_open_error_callback,
@@ -56,7 +59,7 @@ def __init__(self, parameters=None,
 
     def _adapter_connect(self):
         """Connect to the remote socket, adding the socket to the IOLoop if
-        connected
+        connected. 
 
         :rtype: bool