Merge pull request #1305 from pika/lrb-gh-1286

PR 1288 continued

pika/adapters/utils/io_services_utils.py

@@ -1018,6 +1018,10 @@ def write(self, data):
         assert data, ('_AsyncPlaintextTransport.write(): empty data from user.',
                       data, self._state)
 
+        # pika/pika#1286
+        # NOTE: Modify code to write data to buffer before setting writer.
+        # Otherwise a race condition can occur where ioloop executes writer 
+        # while buffer is still empty. 
         tx_buffer_was_empty = self.get_write_buffer_size() == 0
 
         self._buffer_tx_data(data)
@@ -1156,6 +1160,10 @@ def write(self, data):
         assert data, ('_AsyncSSLTransport.write(): empty data from user.',
                       data, self._state)
 
+        # pika/pika#1286
+        # NOTE: Modify code to write data to buffer before setting writer.
+        # Otherwise a race condition can occur where ioloop executes writer 
+        # while buffer is still empty. 
         tx_buffer_was_empty = self.get_write_buffer_size() == 0
 
         self._buffer_tx_data(data)

pika/adapters/utils/selector_ioloop_adapter.py

@@ -325,13 +325,16 @@ def set_writer(self, fd, on_writable):
         else:
             if callbacks.writer is None:
                 assert callbacks.reader is not None
+                # NOTE: Set the writer func before setting the mask!
+                # Otherwise a race condition can occur where ioloop tries to
+                # call writer when it is still None.
+                callbacks.writer = on_writable
                 self._loop.update_handler(
                     fd, self._readable_mask | self._writable_mask)
                 LOGGER.debug('set_writer(%s, _) updated handler RdWr', fd)
             else:
                 LOGGER.debug('set_writer(%s, _) replacing writer', fd)
-
-            callbacks.writer = on_writable
+                callbacks.writer = on_writable
 
     def remove_writer(self, fd):
         """Implement