Merge pull request #364 from Pylons/feature/flush-from-app-threadmaster

Flush data from the application thread

1 files changed, 38 insertions, 8 deletions

diff --git a/src/waitress/channel.py b/src/waitress/channel.py
index 7d1f385..948b498 100644
--- a/src/waitress/channel.py
+++ b/src/waitress/channel.py
@@ -78,6 +78,7 @@ class HTTPChannel(wasyncore.dispatcher):
         may occasionally check if the client has disconnected and interrupt
         execution.
         """
+
         return not self.connected
 
     def writable(self):
@@ -116,23 +117,30 @@ class HTTPChannel(wasyncore.dispatcher):
             #    right now.
             flush = None
 
+        self._flush_exception(flush)
+
+        if self.close_when_flushed and not self.total_outbufs_len:
+            self.close_when_flushed = False
+            self.will_close = True
+
+        if self.will_close:
+            self.handle_close()
+
+    def _flush_exception(self, flush):
         if flush:
             try:
-                flush()
+                return (flush(), False)
             except OSError:
                 if self.adj.log_socket_errors:
                     self.logger.exception("Socket error")
                 self.will_close = True
+
+                return (False, True)
             except Exception:  # pragma: nocover
                 self.logger.exception("Unexpected exception when flushing")
                 self.will_close = True
 
-        if self.close_when_flushed and not self.total_outbufs_len:
-            self.close_when_flushed = False
-            self.will_close = True
-
-        if self.will_close:
-            self.handle_close()
+                return (False, True)
 
     def readable(self):
         # We might want to read more requests. We can only do this if:
@@ -190,6 +198,7 @@ class HTTPChannel(wasyncore.dispatcher):
         Receives input asynchronously and assigns one or more requests to the
         channel.
         """
+
         if not data:
             return False
 
@@ -201,6 +210,7 @@ class HTTPChannel(wasyncore.dispatcher):
 
                 # if there are requests queued, we can not send the continue
                 # header yet since the responses need to be kept in order
+
                 if (
                     self.request.expect_continue
                     and self.request.headers_finished
@@ -215,6 +225,7 @@ class HTTPChannel(wasyncore.dispatcher):
 
                     if not self.request.empty:
                         self.requests.append(self.request)
+
                         if len(self.requests) == 1:
                             # self.requests was empty before so the main thread
                             # is in charge of starting the task. Otherwise,
@@ -363,7 +374,14 @@ class HTTPChannel(wasyncore.dispatcher):
                 self.total_outbufs_len += num_bytes
 
                 if self.total_outbufs_len >= self.adj.send_bytes:
-                    self.server.pull_trigger()
+                    (flushed, exception) = self._flush_exception(self._flush_some)
+
+                    if (
+                        exception
+                        or not flushed
+                        or self.total_outbufs_len >= self.adj.send_bytes
+                    ):
+                        self.server.pull_trigger()
 
             return num_bytes
 
@@ -374,6 +392,17 @@ class HTTPChannel(wasyncore.dispatcher):
 
         if self.total_outbufs_len > self.adj.outbuf_high_watermark:
             with self.outbuf_lock:
+                (_, exception) = self._flush_exception(self._flush_some)
+
+                if exception:
+                    # An exception happened while flushing, wake up the main
+                    # thread, then wait for it to decide what to do next
+                    # (probably close the socket, and then just return)
+                    self.server.pull_trigger()
+                    self.outbuf_lock.wait()
+
+                    return
+
                 while (
                     self.connected
                     and self.total_outbufs_len > self.adj.outbuf_high_watermark
@@ -460,6 +489,7 @@ class HTTPChannel(wasyncore.dispatcher):
             # Add new task to process the next request
             with self.requests_lock:
                 self.requests.pop(0)
+
                 if self.connected and self.requests:
                     self.server.add_task(self)
                 elif (