Merge "Shutdown backend EC connection contexts on disconnect"

swift/proxy/controllers/base.py

@@ -926,6 +926,7 @@ class ResumingGetter(object):
                         if nchunks % 5 == 0:
                             sleep()
 
+            part_iter = None
             try:
                 while True:
                     start_byte, end_byte, length, headers, part = \
@@ -939,6 +940,9 @@ class ResumingGetter(object):
                     self.pop_range()
             except StopIteration:
                 req.environ['swift.non_client_disconnect'] = True
+            finally:
+                if part_iter:
+                    part_iter.close()
 
         except ChunkReadTimeout:
             self.app.exception_occurred(node[0], _('Object'),

swift/proxy/controllers/obj.py

@@ -1118,6 +1118,11 @@ class ECAppIter(object):
         self.stashed_iter = None
 
     def close(self):
+        # close down the stashed iter first so the ContextPool can
+        # cleanup the frag queue feeding coros that may be currently
+        # executing the internal_parts_iters.
+        if self.stashed_iter:
+            self.stashed_iter.close()
         for it in self.internal_parts_iters:
             close_if_possible(it)
 

test/unit/proxy/test_server.py

@@ -5827,23 +5827,24 @@ class TestObjectController(unittest.TestCase):
         exp = 'HTTP/1.1 201'
         self.assertEqual(headers[:len(exp)], exp)
 
-        # get object
+        with mock.patch.object(_test_servers[0], 'client_timeout', new=5):
-        fd.write('GET /v1/a/ec-discon/test HTTP/1.1\r\n'
+            # get object
-                 'Host: localhost\r\n'
+            fd.write('GET /v1/a/ec-discon/test HTTP/1.1\r\n'
-                 'Connection: close\r\n'
+                     'Host: localhost\r\n'
-                 'X-Storage-Token: t\r\n'
+                     'Connection: close\r\n'
-                 '\r\n')
+                     'X-Storage-Token: t\r\n'
-        fd.flush()
+                     '\r\n')
-        headers = readuntil2crlfs(fd)
+            fd.flush()
-        exp = 'HTTP/1.1 200'
+            headers = readuntil2crlfs(fd)
-        self.assertEqual(headers[:len(exp)], exp)
+            exp = 'HTTP/1.1 200'
+            self.assertEqual(headers[:len(exp)], exp)
 
-        # read most of the object, and disconnect
+            # read most of the object, and disconnect
-        fd.read(10)
+            fd.read(10)
-        sock.fd._sock.close()
+            sock.fd._sock.close()
-        condition = \
+            condition = \
-            lambda: _test_servers[0].logger.get_lines_for_level('warning')
+                lambda: _test_servers[0].logger.get_lines_for_level('warning')
-        self._sleep_enough(condition)
+            self._sleep_enough(condition)
 
         # check for disconnect message!
         expected = ['Client disconnected on read'] * 2