libev: handle EAGAIN when message len matches buffer size

thobbs · thobbs · commit ec95133ef43e · 2014-02-18T15:07:43.000-06:00
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -9,6 +9,10 @@ Bug Fixes
 * Always close socket when defuncting error'ed connections to avoid a potential
   file descriptor leak
 * Handle "custom" types (such as the replaced DateType) correctly
+* With libevreactor, correctly handle EAGAIN/EWOULDBLOCK when the message from
+  Cassandra is a multiple of the read buffer size.  Previously, if no more data
+  became available to read on the socket, the message would never be processed,
+  resulting in an OperationTimedOut error.
 
 Other
 -----
diff --git a/cassandra/io/libevreactor.py b/cassandra/io/libevreactor.py
@@ -313,7 +313,7 @@ def handle_read(self, watcher, revents, errno=None):
         except socket.error as err:
             if err.args[0] not in NONBLOCKING:
                 self.defunct(err)
-            return
+                return
 
         if self._iobuf.tell():
             while True:
diff --git a/tests/unit/io/test_libevreactor.py b/tests/unit/io/test_libevreactor.py
@@ -4,6 +4,7 @@
     import unittest # noqa
 
 import errno
+import os
 from StringIO import StringIO
 from socket import error as socket_error
 
@@ -15,7 +16,7 @@
 
 from cassandra.decoder import (write_stringmultimap, write_int, write_string,
                                SupportedMessage, ReadyMessage, ServerError)
-from cassandra.marshal import uint8_pack, uint32_pack
+from cassandra.marshal import uint8_pack, uint32_pack, int32_pack
 
 try:
     from cassandra.io.libevreactor import LibevConnection
@@ -85,6 +86,40 @@ def test_successful_connection(self, *args):
         c.handle_read(None, 0)
 
         self.assertTrue(c.connected_event.is_set())
+        return c
+
+    def test_egain_on_buffer_size(self, *args):
+        # get a connection that's already fully started
+        c = self.test_successful_connection()
+
+        header = '\x00\x00\x00\x00' + int32_pack(20000)
+        responses = [
+            header + ('a' * (4096 - len(header))),
+            'a' * 4096,
+            socket_error(errno.EAGAIN),
+            'a' * 100,
+            socket_error(errno.EAGAIN)]
+
+        def side_effect(*args):
+            response = responses.pop(0)
+            if isinstance(response, socket_error):
+                raise response
+            else:
+                return response
+
+        c._socket.recv.side_effect = side_effect
+        c.handle_read(None, 0)
+        self.assertEquals(c._total_reqd_bytes, 20000 + len(header))
+        # the EAGAIN prevents it from reading the last 100 bytes
+        c._iobuf.seek(0, os.SEEK_END)
+        pos = c._iobuf.tell()
+        self.assertEquals(pos, 4096 + 4096)
+
+        # now tell it to read the last 100 bytes
+        c.handle_read(None, 0)
+        c._iobuf.seek(0, os.SEEK_END)
+        pos = c._iobuf.tell()
+        self.assertEquals(pos, 4096 + 4096 + 100)
 
     def test_protocol_error(self, *args):
         c = self.make_connection()
