improve shutdown logic by having channel.cancel() interrupt the app_iter

4 files changed, 12 insertions, 79 deletions

diff --git a/waitress/channel.py b/waitress/channel.py
index 199ce2a..84abce5 100644
--- a/waitress/channel.py
+++ b/waitress/channel.py
@@ -53,7 +53,6 @@ class HTTPChannel(wasyncore.dispatcher, object):
     close_when_flushed = False   # set to True to close the socket when flushed
     requests = ()                # currently pending requests
     sent_continue = False        # used as a latch after sending 100 continue
-    force_flush = False          # indicates a need to flush the outbuf
 
     #
     # ASYNCHRONOUS METHODS (including __init__)
@@ -107,15 +106,6 @@ class HTTPChannel(wasyncore.dispatcher, object):
             #    because it's either data left over from task output
             #    or a 100 Continue line sent within "received".
             flush = self._flush_some
-        elif self.force_flush:
-            # 1. There's a running task, so we need to try to lock
-            #    the outbuf before sending
-            # 2. This is the last chunk sent by the Nth of M tasks in a
-            #    sequence on this channel, so flush it regardless of whether
-            #    it's >= self.adj.send_bytes.  We need to do this now, or it
-            #    won't get done.
-            flush = self._flush_some_if_lockable
-            self.force_flush = False
         elif (self.total_outbufs_len >= self.adj.send_bytes):
             # 1. There's a running task, so we need to try to lock
             #    the outbuf before sending
@@ -387,15 +377,13 @@ class HTTPChannel(wasyncore.dispatcher, object):
                     request = self.requests.pop(0)
                     request.close()
 
-        self.force_flush = True
-        self.server.pull_trigger()
+        if self.connected:
+            self.server.pull_trigger()
         self.last_activity = time.time()
 
     def cancel(self):
-        """ Cancels all pending requests """
-        self.force_flush = True
+        """ Cancels all pending / active requests """
+        self.will_close = True
+        self.connected = False
         self.last_activity = time.time()
         self.requests = []
-
-    def defer(self):
-        pass
diff --git a/waitress/task.py b/waitress/task.py
index 315a9c2..15aa771 100644
--- a/waitress/task.py
+++ b/waitress/task.py
@@ -122,11 +122,6 @@ class ThreadedTaskDispatcher(object):
                 self.queue_lock.notify(to_stop)
 
     def add_task(self, task):
-        try:
-            task.defer()
-        except:
-            task.cancel()
-            raise
         with self.queue_lock:
             self.queue.append(task)
             self.queue_lock.notify()
@@ -203,12 +198,6 @@ class Task(object):
                     self.status.startswith('304')
                     )
 
-    def cancel(self):
-        self.close_on_finish = True
-
-    def defer(self):
-        pass
-
     def build_response_header(self):
         version = self.version
         # Figure out whether the connection should be closed.
diff --git a/waitress/tests/test_channel.py b/waitress/tests/test_channel.py
index df20dab..1eada4b 100644
--- a/waitress/tests/test_channel.py
+++ b/waitress/tests/test_channel.py
@@ -119,20 +119,6 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertEqual(inst.last_activity, 0)
         self.assertTrue(outbuf.closed)
 
-    def test_handle_write_no_requests_force_flush(self):
-        inst, sock, map = self._makeOneWithMap()
-        inst.requests = [True]
-        inst.outbufs = [DummyBuffer(b'abc')]
-        inst.will_close = False
-        inst.force_flush = True
-        inst.last_activity = 0
-        result = inst.handle_write()
-        self.assertEqual(result, None)
-        self.assertEqual(inst.will_close, False)
-        self.assertTrue(inst.outbuf_lock.acquired)
-        self.assertEqual(inst.force_flush, False)
-        self.assertEqual(sock.sent, b'abc')
-
     def test_handle_write_no_requests_outbuf_gt_send_bytes(self):
         inst, sock, map = self._makeOneWithMap()
         inst.requests = [True]
@@ -456,7 +442,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst.requests = []
         inst.service()
         self.assertEqual(inst.requests, [])
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
 
     def test_service_with_one_request(self):
@@ -507,7 +493,7 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertTrue(request.serviced)
         self.assertEqual(inst.requests, [])
         self.assertEqual(len(inst.logger.exceptions), 1)
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
         self.assertFalse(inst.will_close)
         self.assertEqual(inst.error_task_class.serviced, True)
@@ -526,7 +512,7 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertTrue(request.serviced)
         self.assertEqual(inst.requests, [])
         self.assertEqual(len(inst.logger.exceptions), 1)
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
         self.assertTrue(inst.close_when_flushed)
         self.assertEqual(inst.error_task_class.serviced, False)
@@ -547,7 +533,7 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertFalse(inst.will_close)
         self.assertEqual(inst.requests, [])
         self.assertEqual(len(inst.logger.exceptions), 1)
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
         self.assertEqual(inst.error_task_class.serviced, True)
         self.assertTrue(request.closed)
@@ -565,7 +551,7 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertTrue(request.serviced)
         self.assertEqual(inst.requests, [])
         self.assertEqual(len(inst.logger.exceptions), 1)
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
         self.assertTrue(inst.close_when_flushed)
         self.assertTrue(request.closed)
@@ -585,7 +571,7 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertTrue(request.serviced)
         self.assertEqual(inst.requests, [])
         self.assertEqual(len(inst.logger.warnings), 1)
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
         self.assertFalse(inst.will_close)
         self.assertEqual(inst.error_task_class.serviced, False)
@@ -611,7 +597,7 @@ class TestHTTPChannel(unittest.TestCase):
         self.assertEqual(inst.requests, [])
         self.assertEqual(len(inst.logger.exceptions), 1)
         self.assertEqual(len(inst.logger.warnings), 0)
-        self.assertTrue(inst.force_flush)
+        self.assertTrue(inst.server.trigger_pulled)
         self.assertTrue(inst.last_activity)
         self.assertFalse(inst.will_close)
         self.assertEqual(inst.task_class.serviced, True)
@@ -630,10 +616,6 @@ class TestHTTPChannel(unittest.TestCase):
         inst.cancel()
         self.assertEqual(inst.requests, [])
 
-    def test_defer(self):
-        inst, sock, map = self._makeOneWithMap()
-        self.assertEqual(inst.defer(), None)
-
 class DummySock(object):
     blocking = False
     closed = False
diff --git a/waitress/tests/test_task.py b/waitress/tests/test_task.py
index f3f14d0..e1415ef 100644
--- a/waitress/tests/test_task.py
+++ b/waitress/tests/test_task.py
@@ -72,7 +72,6 @@ class TestThreadedTaskDispatcher(unittest.TestCase):
         inst = self._makeOne()
         inst.add_task(task)
         self.assertEqual(len(inst.queue), 1)
-        self.assertTrue(task.deferred)
 
     def test_log_queue_depth(self):
         task = DummyTask()
@@ -83,18 +82,6 @@ class TestThreadedTaskDispatcher(unittest.TestCase):
         inst.add_task(task)
         self.assertEqual(len(inst.queue_logger.logged), 2)
 
-    def test_add_task_defer_raises(self):
-        class BadDummyTask(DummyTask):
-            def defer(self):
-                super(BadDummyTask, self).defer()
-                raise ValueError
-        task = BadDummyTask()
-        inst = self._makeOne()
-        self.assertRaises(ValueError, inst.add_task, task)
-        self.assertEqual(len(inst.queue), 0)
-        self.assertTrue(task.deferred)
-        self.assertTrue(task.cancelled)
-
     def test_shutdown_one_thread(self):
         inst = self._makeOne()
         inst.threads.add(0)
@@ -130,15 +117,6 @@ class TestTask(unittest.TestCase):
         inst = self._makeOne(request=request)
         self.assertEqual(inst.version, '1.0')
 
-    def test_cancel(self):
-        inst = self._makeOne()
-        inst.cancel()
-        self.assertTrue(inst.close_on_finish)
-
-    def test_defer(self):
-        inst = self._makeOne()
-        self.assertEqual(inst.defer(), None)
-
     def test_build_response_header_bad_http_version(self):
         inst = self._makeOne()
         inst.request = DummyParser()
@@ -1478,15 +1456,11 @@ class DummyError(object):
 
 class DummyTask(object):
     serviced = False
-    deferred = False
     cancelled = False
 
     def service(self):
         self.serviced = True
 
-    def defer(self):
-        self.deferred = True
-
     def cancel(self):
         self.cancelled = True