remote: s3: revert jobs adjustment, pass OSError to main

dvc/exceptions.py

@@ -344,13 +344,3 @@ class NoOutputInExternalRepoError(DvcException):
 class HTTPError(DvcException):
     def __init__(self, code, reason):
         super(HTTPError, self).__init__("'{} {}'".format(code, reason))
-
-
-class TooManyOpenFilesError(DvcException):
-    def __init__(self, cause):
-        super(TooManyOpenFilesError, 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.",
-            cause=cause,
-        )

dvc/main.py

@@ -1,6 +1,7 @@
 """Main entry point for dvc CLI."""
 from __future__ import unicode_literals
 
+import errno
 import logging
 
 from dvc import analytics
@@ -64,6 +65,10 @@ def main(argv=None):
                 "unicode is not supported in DVC for Python 2 "
                 "(end-of-life January 1, 2020), please upgrade to Python 3"
             )
+        elif isinstance(exc, OSError) and exc.errno == errno.EMFILE:
+            logger.exception(
+                "too many open files error, please increase you `ulimit`"
+            )
         else:
             logger.exception("unexpected error")
         ret = 255

dvc/remote/base.py

@@ -1,4 +1,7 @@
 from __future__ import unicode_literals
+
+import errno
+
 from dvc.utils.compat import basestring, FileNotFoundError, str, urlparse
 
 import itertools
@@ -19,7 +22,6 @@ from dvc.exceptions import (
     DvcException,
     ConfirmRemoveError,
     DvcIgnoreInCollectedDirError,
-    TooManyOpenFilesError,
 )
 from dvc.ignore import DvcIgnore
 from dvc.path_info import PathInfo, URLInfo
@@ -85,9 +87,6 @@ class RemoteBASE(object):
 
     state = StateNoop()
 
-    def adjust_jobs(self, jobs=None):
-        return jobs or self.JOBS
-
     def __init__(self, repo, config):
         self.repo = repo
 
@@ -523,8 +522,8 @@ class RemoteBASE(object):
     def _handle_transfer_exception(
         self, from_info, to_info, exception, operation
     ):
-        if isinstance(exception, OSError) and exception.errno == 24:
-            raise TooManyOpenFilesError(exception)
+        if isinstance(exception, OSError) and exception.errno == errno.EMFILE:
+            raise exception
 
         msg = "failed to {} '{}' to '{}'".format(operation, from_info, to_info)
         logger.exception(msg)

dvc/remote/local.py

@@ -338,7 +338,8 @@ class RemoteLOCAL(RemoteBASE):
             func = remote.upload
             status = STATUS_NEW
 
-        jobs = remote.adjust_jobs(jobs)
+        if jobs is None:
+            jobs = remote.JOBS
 
         status_info = self.status(
             named_cache,

dvc/remote/s3.py

@@ -276,7 +276,6 @@ class RemoteS3(RemoteBASE):
                 to_info.path,
                 Callback=pbar.update,
                 ExtraArgs=self.extra_args,
-                Config=self.transfer_config,
             )
 
     def _download(self, from_info, to_file, name=None, no_progress_bar=False):
@@ -290,11 +289,7 @@ class RemoteS3(RemoteBASE):
             disable=no_progress_bar, total=total, bytes=True, desc=name
         ) as pbar:
             self.s3.download_file(
-                from_info.bucket,
-                from_info.path,
-                to_file,
-                Callback=pbar.update,
-                Config=self.transfer_config,
+                from_info.bucket, from_info.path, to_file, Callback=pbar.update
             )
 
     def _generate_download_url(self, path_info, expires=3600):
@@ -340,36 +335,3 @@ class RemoteS3(RemoteBASE):
                     )
 
                 self.extra_args[extra_args_key] = config.get(grant_option)
-
-    @cached_property
-    def transfer_config(self):
-        from boto3.s3.transfer import TransferConfig
-
-        return TransferConfig()
-
-    def adjust_jobs(self, jobs=None):
-        if os.name == "nt":
-            import win32file
-
-            descriptor_limit = win32file._getmaxstdio()
-        else:
-            import resource
-
-            descriptor_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
-        threads_per_job = self.transfer_config.max_request_concurrency
-        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
-
-        safe_jobs_number = (descriptor_limit - safety_margin) // (
-            threads_per_job * fds_per_thread
-        )
-        if not jobs_declared:
-            jobs = safe_jobs_number
-        return jobs

tests/func/test_remote.py

@@ -1,3 +1,4 @@
+import errno
 import os
 import shutil
 
@@ -6,7 +7,6 @@ import pytest
 from mock import patch
 
 from dvc.config import Config
-from dvc.exceptions import TooManyOpenFilesError
 from dvc.main import main
 from dvc.path_info import PathInfo
 from dvc.remote import RemoteLOCAL, RemoteConfig
@@ -265,11 +265,12 @@ def test_raise_on_too_many_open_files(tmp_dir, dvc, tmp_path_factory, mocker):
 
     tmp_dir.dvc_gen({"file": "file content"})
 
-    too_many_open_files_error = OSError()
-    mocker.patch.object(too_many_open_files_error, "errno", 24)
     mocker.patch.object(
-        RemoteLOCAL, "_upload", side_effect=too_many_open_files_error
+        RemoteLOCAL,
+        "_upload",
+        side_effect=OSError(errno.EMFILE, "Too many open files"),
     )
 
-    with pytest.raises(TooManyOpenFilesError):
+    with pytest.raises(OSError) as e:
         dvc.push()
+        assert e.errno == errno.EMFILE

tests/unit/remote/test_s3.py

@@ -1,10 +1,8 @@
-import os
-
 import pytest
 
 from dvc.config import ConfigError
-from dvc.remote import RemoteS3
-from tests.utils import empty_caplog
+from dvc.remote.s3 import RemoteS3
+
 
 bucket_name = "bucket-name"
 prefix = "some/prefix"
@@ -56,21 +54,3 @@ def test_grants_mutually_exclusive_acl_error(grants):
 
         with pytest.raises(ConfigError):
             RemoteS3(None, config)
-
-
-@pytest.mark.parametrize(
-    "default_jobs_number,expected_result", [(10, 10), (13, 12)]
-)
-def test_adjust_default_jobs_number(
-    mocker, caplog, default_jobs_number, expected_result
-):
-    remote = RemoteS3(None, {})
-
-    if os.name == "nt":
-        mocker.patch("win32file._getmaxstdio", return_value=256)
-    else:
-        mocker.patch("resource.getrlimit", return_value=(256, 1024))
-    mocker.patch.object(remote, "JOBS", default_jobs_number)
-
-    with empty_caplog(caplog):
-        assert remote.adjust_jobs() == expected_result

tests/utils/__init__.py

@@ -45,14 +45,3 @@ def trees_equal(dir_path_1, dir_path_2):
 
 def to_posixpath(path):
     return path.replace("\\", "/")
-
-
-class empty_caplog(object):
-    def __init__(self, caplog):
-        self.caplog = caplog
-
-    def __enter__(self):
-        self.caplog.clear()
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        assert self.caplog.text == ""