Remove all DLO segments on upload of replacement

Previously, only the first container-listing's worth of segments was
deleted, which would leave behind orphaned segments when the object was
very large with small segments or the server's container_listing_limit
was small.

In addition, process DLO and SLO deletions on the segment thread pool,
rather than the object thread pool.

Change-Id: I1587375261a6237fa55a9cb96bda8dae918cc795
Related-Bug: #1418007

swiftclient/service.py

@@ -1728,19 +1728,19 @@ class SwiftService(object):
 
             if old_manifest or old_slo_manifest_paths:
                 drs = []
+                delobjsmap = {}
                 if old_manifest:
                     scontainer, sprefix = old_manifest.split('/', 1)
                     scontainer = unquote(scontainer)
                     sprefix = unquote(sprefix).rstrip('/') + '/'
-                    delobjs = []
-                    for delobj in conn.get_container(scontainer,
-                                                     prefix=sprefix)[1]:
-                        delobjs.append(delobj['name'])
-                    for dr in self.delete(container=scontainer,
-                                          objects=delobjs):
-                        drs.append(dr)
+                    delobjsmap[scontainer] = []
+                    for part in self.list(scontainer, {'prefix': sprefix}):
+                        if not part["success"]:
+                            raise part["error"]
+                        delobjsmap[scontainer].extend(
+                            seg['name'] for seg in part['listing'])
+
                 if old_slo_manifest_paths:
-                    delobjsmap = {}
                     for seg_to_delete in old_slo_manifest_paths:
                         if seg_to_delete in new_slo_manifest_paths:
                             continue
@@ -1749,10 +1749,18 @@ class SwiftService(object):
                         delobjs_cont = delobjsmap.get(scont, [])
                         delobjs_cont.append(sobj)
                         delobjsmap[scont] = delobjs_cont
-                    for (dscont, dsobjs) in delobjsmap.items():
-                        for dr in self.delete(container=dscont,
-                                              objects=dsobjs):
-                            drs.append(dr)
+
+                del_segs = []
+                for dscont, dsobjs in delobjsmap.items():
+                    for dsobj in dsobjs:
+                        del_seg = self.thread_manager.segment_pool.submit(
+                            self._delete_segment, dscont, dsobj,
+                            results_queue=results_queue
+                        )
+                        del_segs.append(del_seg)
+
+                for del_seg in interruptable_as_completed(del_segs):
+                    drs.append(del_seg.result())
                 res['segment_delete_results'] = drs
 
             # return dict for printing

tests/unit/test_shell.py

@@ -489,11 +489,11 @@ class TestShell(unittest.TestCase):
         expected_delete_calls = [
             mock.call(
                 b'container1', b'old_seg1',
-                query_string=None, response_dict={}
+                response_dict={}
             ),
             mock.call(
                 b'container2', b'old_seg2',
-                query_string=None, response_dict={}
+                response_dict={}
             )
         ]
         self.assertEqual(
@@ -538,9 +538,11 @@ class TestShell(unittest.TestCase):
         ]
         connection.return_value.get_container.side_effect = [
             [None, [{'name': 'prefix_a', 'bytes': 0,
-                     'last_modified': '123T456'},
-                    {'name': 'prefix_b', 'bytes': 0,
-                     'last_modified': '123T456'}]]
+                     'last_modified': '123T456'}]],
+            # Have multiple pages worth of DLO segments
+            [None, [{'name': 'prefix_b', 'bytes': 0,
+                     'last_modified': '123T456'}]],
+            [None, []]
         ]
         connection.return_value.put_object.return_value = (
             'd41d8cd98f00b204e9800998ecf8427e')
@@ -555,11 +557,11 @@ class TestShell(unittest.TestCase):
         expected_delete_calls = [
             mock.call(
                 'container1', 'prefix_a',
-                query_string=None, response_dict={}
+                response_dict={}
             ),
             mock.call(
                 'container1', 'prefix_b',
-                query_string=None, response_dict={}
+                response_dict={}
             )
         ]
         self.assertEqual(