Cleanup incomplete upload upon error

Since we do not have resuming support, we should simply cleanup failed transfers.
minio · Dec 7, 2017 · 1255abe · 1255abe
1 parent 978fb46
commit 1255abe
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 7 deletions.
diff --git a/minio/api.py b/minio/api.py
@@ -1527,7 +1527,15 @@ def _stream_put_object(self, bucket_name, object_name,
             parts_to_upload.append((bucket_name, object_name, upload_id, part_number, part_data))
 
         # Run parts upload in parallel
-        pool.parallel_run(self._upload_part_routine, parts_to_upload)
+        try:
+            pool.parallel_run(self._upload_part_routine, parts_to_upload)
+        except:
+            # Any exception that occurs sends an abort on the
+            # on-going multipart operation.
+            self._remove_incomplete_upload(bucket_name,
+                                           object_name,
+                                           upload_id)
+            raise
 
         # Update uploaded_parts with the part uploads result
         # and check total uploaded data.
@@ -1546,14 +1554,27 @@ def _stream_put_object(self, bucket_name, object_name,
         if total_uploaded != content_size:
             msg = 'Data uploaded {0} is not equal input size ' \
                   '{1}'.format(total_uploaded, content_size)
+            # cleanup incomplete upload upon incorrect upload
+            # automatically
+            self._remove_incomplete_upload(bucket_name,
+                                           object_name,
+                                           upload_id)
             raise InvalidSizeError(msg)
 
         # Complete all multipart transactions if possible.
-        mpart_result = self._complete_multipart_upload(bucket_name,
-                                                       object_name,
-                                                       upload_id,
-                                                       uploaded_parts,
-                                                       metadata=metadata)
+        try:
+            mpart_result = self._complete_multipart_upload(bucket_name,
+                                                           object_name,
+                                                           upload_id,
+                                                           uploaded_parts,
+                                                           metadata=metadata)
+        except:
+            # Any exception that occurs sends an abort on the
+            # on-going multipart operation.
+            self._remove_incomplete_upload(bucket_name,
+                                           object_name,
+                                           upload_id)
+
         # Return etag here.
         return mpart_result.etag
 

diff --git a/minio/thread_pool.py b/minio/thread_pool.py
@@ -51,6 +51,7 @@ def run(self):
                 except Exception as e:
                     self.exceptions_queue.put(e)
                     fast_quit = True
+
             # Mark this task as done, whether an exception happened or not
             self.tasks_queue.task_done()
 
@@ -82,4 +83,3 @@ def parallel_run(self, func, args_list):
     def result(self):
         """ Return the result of all called tasks """
         return self.results_queue
-