remote: upload/download: handle OSError

dvc/exceptions.py

@@ -344,3 +344,12 @@ class NoOutputInExternalRepoError(DvcException):
 class HTTPError(DvcException):
     def __init__(self, code, reason):
         super(HTTPError, self).__init__("'{} {}'".format(code, reason))
+
+
+class TooManyOpenFilesException(DvcException):
+    def __init__(self):
+        super(TooManyOpenFilesException, self).__init__(
+            "Operation failed due to too many open file descriptors. reduce "
+            "the number of jobs or increase open file descriptors limit to "
+            "prevent this."
+        )

dvc/remote/base.py

@@ -13,15 +13,11 @@ from operator import itemgetter
 
 from shortuuid import uuid
 
-from funcy import cached_property
-
 import dvc.prompt as prompt
 from dvc.config import Config
-from dvc.exceptions import (
-    DvcException,
-    ConfirmRemoveError,
-    DvcIgnoreInCollectedDirError,
-)
+from dvc.exceptions import ConfirmRemoveError, TooManyOpenFilesException
+from dvc.exceptions import DvcException
+from dvc.exceptions import DvcIgnoreInCollectedDirError
 from dvc.ignore import DvcIgnore
 from dvc.path_info import PathInfo, URLInfo
 from dvc.progress import Tqdm
@@ -521,6 +517,16 @@ class RemoteBASE(object):
             return
         self._save_file(path_info, checksum)
 
+    def _handle_transfer_exception(
+        self, from_info, to_info, exception, operation
+    ):
+        if isinstance(exception, OSError) and exception.errno == 24:
+            raise TooManyOpenFilesException()
+
+        msg = "failed to {} '{}' to '{}'".format(operation, from_info, to_info)
+        logger.exception(msg)
+        return 1
+
     def upload(self, from_info, to_info, name=None, no_progress_bar=False):
         if not hasattr(self, "_upload"):
             raise RemoteActionNotImplemented("upload", self.scheme)
@@ -542,10 +548,10 @@ class RemoteBASE(object):
                 name=name,
                 no_progress_bar=no_progress_bar,
             )
-        except Exception:
-            msg = "failed to upload '{}' to '{}'"
-            logger.exception(msg.format(from_info, to_info))
-            return 1  # 1 fail
+        except Exception as e:
+            return self._handle_transfer_exception(
+                from_info, to_info, e, "upload"
+            )
 
         return 0
 
@@ -619,10 +625,10 @@ class RemoteBASE(object):
             self._download(
                 from_info, tmp_file, name=name, no_progress_bar=no_progress_bar
             )
-        except Exception:
-            msg = "failed to download '{}' to '{}'"
-            logger.exception(msg.format(from_info, to_info))
-            return 1  # 1 fail
+        except Exception as e:
+            return self._handle_transfer_exception(
+                from_info, to_info, e, "download"
+            )
 
         move(tmp_file, to_info, mode=file_mode)
 

dvc/remote/s3.py

@@ -348,9 +348,6 @@ class RemoteS3(RemoteBASE):
         return TransferConfig()
 
     def adjust_jobs(self, jobs=None):
-        jobs_declared = bool(jobs)
-        jobs = jobs or self.JOBS
-
         if os.name == "nt":
             import win32file
 
@@ -363,6 +360,9 @@ class RemoteS3(RemoteBASE):
         fds_per_thread = 2  # file and socket
         safety_margin = 10
 
+        jobs_declared = bool(jobs)
+        jobs = jobs or self.JOBS
+
         estimated_descriptors_num = jobs * threads_per_job * fds_per_thread
         if estimated_descriptors_num <= descriptor_limit - safety_margin:
             return jobs
@@ -370,16 +370,6 @@ class RemoteS3(RemoteBASE):
         safe_jobs_number = (descriptor_limit - safety_margin) // (
             threads_per_job * fds_per_thread
         )
-        safe_descriptors_limit = estimated_descriptors_num + safety_margin
-        if jobs_declared:
-            logger.warning(
-                "Provided jobs number '{}' might result in 'Too many open "
-                "files error'. Consider decreasing jobs number to '{}' or "
-                "increasing "
-                "file descriptors limit to '{}'.".format(
-                    jobs, safe_jobs_number, safe_descriptors_limit
-                )
-            )
-        else:
+        if not jobs_declared:
             jobs = safe_jobs_number
         return jobs

tests/unit/remote/test_s3.py

@@ -2,7 +2,6 @@ import logging
 import os
 
 import pytest
-from mock import patch
 from dvc.remote import RemoteS3
 from tests.utils import empty_caplog