Fix asyncio message chunks can be processed discontinuously

Author: aboudreault

CHANGELOG.rst

@@ -18,6 +18,7 @@ Features
 Bug Fixes
 ---------
 * re-raising the CQLEngineException will fail on Python 3 (PYTHON-1166)
+* asyncio message chunks can be processed discontinuously (PYTHON-1185)
 
 Others
 ------

build.yaml

@@ -46,8 +46,8 @@ schedules:
       EXCLUDE_LONG=1
     matrix:
       exclude:
-        - python: [2.7, 3.4, 3.6, 3.7]
-        - cassandra: ['2.0', '2.1', '2.2', '3.0', 'test-dse']
+        - python: [2.7, 3.4, 3.7]
+        - cassandra: ['2.0', '2.1', '2.2', '3.0', 'test-dse', dse-4.8', 'dse-5.0']
 
   release_test:
     schedule: per_commit

cassandra/io/asyncioreactor.py

@@ -83,6 +83,7 @@ class AsyncioConnection(Connection):
     _loop_thread = None
 
     _write_queue = None
+    _write_queue_lock = None
 
     def __init__(self, *args, **kwargs):
         Connection.__init__(self, *args, **kwargs)
@@ -91,6 +92,7 @@ def __init__(self, *args, **kwargs):
         self._socket.setblocking(0)
 
         self._write_queue = asyncio.Queue(loop=self._loop)
+        self._write_queue_lock = asyncio.Lock(loop=self._loop)
 
         # see initialize_reactor -- loop is running in a separate thread, so we
         # have to use a threadsafe call
@@ -157,20 +159,28 @@ def _close(self):
     def push(self, data):
         buff_size = self.out_buffer_size
         if len(data) > buff_size:
+            chunks = []
             for i in range(0, len(data), buff_size):
-                self._push_chunk(data[i:i + buff_size])
+                chunks.append(data[i:i + buff_size])
         else:
-            self._push_chunk(data)
+            chunks = [data]
 
-    def _push_chunk(self, chunk):
         if self._loop_thread.ident != get_ident():
             asyncio.run_coroutine_threadsafe(
-                self._write_queue.put(chunk),
+                self._push_msg(chunks),
                 loop=self._loop
             )
         else:
             # avoid races/hangs by just scheduling this, not using threadsafe
-            self._loop.create_task(self._write_queue.put(chunk))
+            self._loop.create_task(self._push_msg(chunks))
+
+    @asyncio.coroutine
+    def _push_msg(self, chunks):
+        # This lock ensures all chunks of a message are sequential in the Queue
+        with (yield from self._write_queue_lock):
+            for chunk in chunks:
+                self._write_queue.put_nowait(chunk)
+
 
     @asyncio.coroutine
     def handle_write(self):

tests/integration/__init__.py

@@ -355,6 +355,7 @@ def _id_and_mark(f):
 requiredse = unittest.skipUnless(DSE_VERSION, "DSE required")
 requirescloudproxy = unittest.skipIf(CLOUD_PROXY_PATH is None, "Cloud Proxy path hasn't been specified")
 
+libevtest = unittest.skipUnless(EVENT_LOOP_MANAGER=="libev", "Test timing designed for libev loop")
 
 def wait_for_node_socket(node, timeout):
     binary_itf = node.network_interfaces['binary']

tests/integration/simulacron/test_connection.py

@@ -27,7 +27,7 @@
 from cassandra.policies import HostStateListener, RoundRobinPolicy
 
 from tests import connection_class, thread_pool_executor_class
-from tests.integration import requiressimulacron
+from tests.integration import requiressimulacron, libevtest
 from tests.integration.util import assert_quiescent_pool_state, late
 # important to import the patch PROTOCOL_VERSION from the simulacron module
 from tests.integration.simulacron import SimulacronBase, PROTOCOL_VERSION
@@ -194,6 +194,7 @@ def test_callbacks_and_pool_when_oto(self):
         callback.assert_not_called()
 
     @cythontest
+    @libevtest
     def test_heartbeat_defunct_deadlock(self):
         """
         Ensure that there is no deadlock when request is in-flight and heartbeat defuncts connection