Shutdown IO queue when file can't be opened

boto3/compat.py

@@ -13,6 +13,18 @@
 import sys
 import os
 import errno
+import socket
+
+from botocore.vendored import six
+
+if six.PY3:
+    # In python3, socket.error is OSError, which is too general
+    # for what we want (i.e FileNotFoundError is a subclass of OSError).
+    # In py3 all the socket related errors are in a newly created
+    # ConnectionError
+    SOCKET_ERROR = ConnectionError
+else:
+    SOCKET_ERROR = socket.error
 
 
 if sys.platform.startswith('win'):

boto3/s3/transfer.py

@@ -148,6 +148,10 @@ def __call__(self, bytes_amount):
 
 MB = 1024 * 1024
 SHUTDOWN_SENTINEL = object()
+S3_RETRYABLE_ERRORS = (
+    socket.timeout, boto3.compat.SOCKET_ERROR,
+    ReadTimeoutError, IncompleteReadError
+)
 
 
 def random_file_extension(num_digits=8):
@@ -523,8 +527,7 @@ def _download_range(self, bucket, key, filename,
                         self._ioqueue.put((current_index, chunk))
                         current_index += len(chunk)
                     return
-                except (socket.timeout, socket.error,
-                        ReadTimeoutError, IncompleteReadError) as e:
+                except S3_RETRYABLE_ERRORS as e:
                     logger.debug("Retrying exception caught (%s), "
                                  "retrying request, (attempt %s / %s)", e, i,
                                  max_attempts, exc_info=True)
@@ -535,6 +538,15 @@ def _download_range(self, bucket, key, filename,
             logger.debug("EXITING _download_range for part: %s", part_index)
 
     def _perform_io_writes(self, filename):
+        try:
+            self._loop_on_io_writes(filename)
+        except Exception as e:
+            logger.debug("Caught exception in IO thread: %s",
+                         e, exc_info=True)
+            self._ioqueue.trigger_shutdown()
+            raise
+
+    def _loop_on_io_writes(self, filename):
         with self._os.open(filename, 'wb') as f:
             while True:
                 task = self._ioqueue.get()
@@ -543,15 +555,9 @@ def _perform_io_writes(self, filename):
                                  "shutting down IO handler.")
                     return
                 else:
-                    try:
-                        offset, data = task
-                        f.seek(offset)
-                        f.write(data)
-                    except Exception as e:
-                        logger.debug("Caught exception in IO thread: %s",
-                                     e, exc_info=True)
-                        self._ioqueue.trigger_shutdown()
-                        raise
+                    offset, data = task
+                    f.seek(offset)
+                    f.write(data)
 
 
 class TransferConfig(object):
@@ -699,10 +705,7 @@ def _get_object(self, bucket, key, filename, extra_args, callback):
             try:
                 return self._do_get_object(bucket, key, filename,
                                            extra_args, callback)
-            except (socket.timeout, socket.error,
-                    ReadTimeoutError, IncompleteReadError) as e:
-                # TODO: we need a way to reset the callback if the
-                # download failed.
+            except S3_RETRYABLE_ERRORS as e:
                 logger.debug("Retrying exception caught (%s), "
                              "retrying request, (attempt %s / %s)", e, i,
                              max_attempts, exc_info=True)

tests/integration/test_s3.py

@@ -484,6 +484,32 @@ def test_download_above_threshold(self):
                                download_path)
         assert_files_equal(filename, download_path)
 
+    def test_download_file_with_directory_not_exist(self):
+        transfer = self.create_s3_transfer()
+        self.client.put_object(Bucket=self.bucket_name,
+                                Key='foo.txt',
+                                Body=b'foo')
+        self.addCleanup(self.delete_object, 'foo.txt')
+        download_path = os.path.join(self.files.rootdir, 'a', 'b', 'c',
+                                     'downloaded.txt')
+        with self.assertRaises(IOError):
+            transfer.download_file(self.bucket_name, 'foo.txt', download_path)
+
+    def test_download_large_file_directory_not_exist(self):
+        transfer = self.create_s3_transfer()
+
+        filename = self.files.create_file_with_size(
+            'foo.txt', filesize=20 * 1024 * 1024)
+        with open(filename, 'rb') as f:
+            self.client.put_object(Bucket=self.bucket_name,
+                                   Key='foo.txt',
+                                   Body=f)
+            self.addCleanup(self.delete_object, 'foo.txt')
+        download_path = os.path.join(self.files.rootdir, 'a', 'b', 'c',
+                                     'downloaded.txt')
+        with self.assertRaises(IOError):
+            transfer.download_file(self.bucket_name, 'foo.txt', download_path)
+
     def test_transfer_methods_through_client(self):
         # This is really just a sanity check to ensure that the interface
         # from the clients work.  We're not exhaustively testing through

tests/unit/s3/test_transfer.py

@@ -398,7 +398,7 @@ def test_retry_on_failures_from_stream_reads(self):
         response_body = b'foobarbaz'
         stream_with_errors = mock.Mock()
         stream_with_errors.read.side_effect = [
-            socket.error("fake error"),
+            socket.timeout("fake error"),
             response_body
         ]
         client.get_object.return_value = {'Body': stream_with_errors}
@@ -429,7 +429,7 @@ def test_exception_raised_on_exceeded_retries(self):
         client = mock.Mock()
         response_body = b'foobarbaz'
         stream_with_errors = mock.Mock()
-        stream_with_errors.read.side_effect = socket.error("fake error")
+        stream_with_errors.read.side_effect = socket.timeout("fake error")
         client.get_object.return_value = {'Body': stream_with_errors}
         config = TransferConfig(multipart_threshold=4,
                                 multipart_chunksize=4)
@@ -459,6 +459,22 @@ def test_io_thread_failure_triggers_shutdown(self):
             downloader.download_file('bucket', 'key', 'filename',
                                      len(response_body), {})
 
+    def test_io_thread_fails_to_open_triggers_shutdown_error(self):
+        client = mock.Mock()
+        client.get_object.return_value = {
+            'Body': six.BytesIO(b'asdf')
+        }
+        os_layer = mock.Mock(spec=OSUtils)
+        os_layer.open.side_effect = IOError("Can't open file")
+        downloader = MultipartDownloader(
+            client, TransferConfig(),
+            os_layer, SequentialExecutor)
+        # We're verifying that the exception raised from the IO future
+        # propogates back up via download_file().
+        with self.assertRaisesRegexp(IOError, "Can't open file"):
+            downloader.download_file('bucket', 'key', 'filename',
+                                     len(b'asdf'), {})
+
     def test_download_futures_fail_triggers_shutdown(self):
         class FailedDownloadParts(SequentialExecutor):
             def __init__(self, max_workers):
@@ -619,7 +635,7 @@ def test_get_object_stream_is_retried_and_succeeds(self):
             'ContentLength': below_threshold}
         self.client.get_object.side_effect = [
             # First request fails.
-            socket.error("fake error"),
+            socket.timeout("fake error"),
             # Second succeeds.
             {'Body': six.BytesIO(b'foobar')}
         ]
@@ -636,7 +652,7 @@ def test_get_object_stream_uses_all_retries_and_errors_out(self):
         # Here we're raising an exception every single time, which
         # will exhaust our retry count and propogate a
         # RetriesExceededError.
-        self.client.get_object.side_effect = socket.error("fake error")
+        self.client.get_object.side_effect = socket.timeout("fake error")
         with self.assertRaises(RetriesExceededError):
             transfer.download_file('bucket', 'key', 'smallfile')