download_file hangs on windows if directory does not exist
On windows with python 3.4.3 when running in IDLE the download_file function will not return (or at least not in a reasonable time) when the filepath argument contains a non-existent directory.

The following code:

``` python
import boto3
from os import path
s3_client = boto3.client('s3')
fp = r"C:\\Users\\AUser\\I_dont_exist"
print("Go get that key!")
s3_client.download_file('bucket',
                             'key',path.join(fp,"key.exn"))
print("I got that key")
```

results in:

``` python
>>> 
Go get that key!
```

Adding a call to os.mkdir:

``` python
import boto3
from os import path, mkdir
s3_client = boto3.client('s3')
fp = r"C:\\Users\\AUser\\I_dont_exist"
mkdir(fp)
print("Go get that key!")
s3_client.download_file('bucket',
                             'key',path.join(fp,"key.exn"))
print("I got that key")
```

results in:

``` python
>>> 
Go get that key!
I got that key
```

At the very least this call to download_file should return an Exception.

diff --git a/boto3/compat.py b/boto3/compat.py
index 324b60a2..3a7a832f 100644
--- a/boto3/compat.py
+++ b/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'):
diff --git a/boto3/s3/transfer.py b/boto3/s3/transfer.py
index 1707dc2c..096c9434 100644
--- a/boto3/s3/transfer.py
+++ b/boto3/s3/transfer.py
@@ -148,6 +148,10 @@ queue = six.moves.queue
 
 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 @@ class MultipartDownloader(object):
                         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 @@ class MultipartDownloader(object):
             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 @@ class MultipartDownloader(object):
                                  "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 @@ class S3Transfer(object):
             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)
diff --git a/tests/integration/test_s3.py b/tests/integration/test_s3.py
index 4eec2eb5..583bfd83 100644
--- a/tests/integration/test_s3.py
+++ b/tests/integration/test_s3.py
@@ -484,6 +484,32 @@ class TestS3Transfers(unittest.TestCase):
                                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
diff --git a/tests/unit/s3/test_transfer.py b/tests/unit/s3/test_transfer.py
index 5a053ce1..f9ec75f9 100644
--- a/tests/unit/s3/test_transfer.py
+++ b/tests/unit/s3/test_transfer.py
@@ -398,7 +398,7 @@ class TestMultipartDownloader(unittest.TestCase):
         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 @@ class TestMultipartDownloader(unittest.TestCase):
         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 @@ class TestMultipartDownloader(unittest.TestCase):
             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 @@ class TestS3Transfer(unittest.TestCase):
             '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 @@ class TestS3Transfer(unittest.TestCase):
         # 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')
 
['boto3/compat.py', 'boto3/s3/transfer.py', 'tests/unit/s3/test_transfer.py', 'tests/integration/test_s3.py']