remote: s3: adjust jobs only on default

dvc/remote/base.py

@@ -77,6 +77,7 @@ class RemoteBASE(object):
     scheme = "base"
     path_cls = URLInfo
     REQUIRES = {}
+    JOBS = 4 * cpu_count()
 
     PARAM_RELPATH = "relpath"
     CHECKSUM_DIR_SUFFIX = ".dir"
@@ -85,9 +86,8 @@ class RemoteBASE(object):
 
     state = StateNoop()
 
-    @cached_property
-    def jobs(self):
-        return cpu_count() * 4
+    def adjust_jobs(self, jobs=None):
+        return jobs or self.JOBS
 
     def __init__(self, repo, config):
         self.repo = repo
@@ -751,11 +751,6 @@ class RemoteBASE(object):
             return self._changed_dir_cache(checksum)
         return self.changed_cache_file(checksum)
 
-    def _adjust_jobs(self, jobs=None):
-        if not jobs:
-            jobs = self.jobs
-        return jobs
-
     def cache_exists(self, checksums, jobs=None, name=None):
         """Check if the given checksums are stored in the remote.
 
@@ -794,9 +789,7 @@ class RemoteBASE(object):
                 pbar.update_desc(str(path_info))
                 return ret
 
-            with ThreadPoolExecutor(
-                max_workers=self._adjust_jobs(jobs)
-            ) as executor:
+            with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor:
                 path_infos = map(self.checksum_to_path_info, checksums)
                 in_remote = executor.map(exists_with_progress, path_infos)
                 ret = list(itertools.compress(checksums, in_remote))

dvc/remote/local.py

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

dvc/remote/s3.py

@@ -277,6 +277,7 @@ 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,7 +291,11 @@ 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
+                from_info.bucket,
+                from_info.path,
+                to_file,
+                Callback=pbar.update,
+                Config=self.transfer_config,
             )
 
     def _generate_download_url(self, path_info, expires=3600):
@@ -337,19 +342,38 @@ class RemoteS3(RemoteBASE):
 
                 self.extra_args[extra_args_key] = config.get(grant_option)
 
-    def _adjust_jobs(self, jobs=None):
-        jobs = super(RemoteS3, self)._adjust_jobs(jobs)
+    @cached_property
+    def transfer_config(self):
+        from boto3.s3.transfer import TransferConfig
+
+        return TransferConfig()
+
+    def adjust_jobs(self, jobs=None):
+        jobs_declared = bool(jobs)
+        jobs = jobs or self.JOBS
 
         descriptor_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
-        estimated_descriptors_num = jobs * 20
-        if estimated_descriptors_num <= descriptor_limit - 10:
+        threads_per_job = self.transfer_config.max_request_concurrency
+        fds_per_thread = 2  # file and socket
+        safety_margin = 10
+
+        estimated_descriptors_num = jobs * threads_per_job * fds_per_thread
+        if estimated_descriptors_num <= descriptor_limit - safety_margin:
             return jobs
 
-        jobs = (descriptor_limit - 10) // 20
-        logger.warning(
-            "Parallelization reduced to '{}' jobs. Increase open "
-            "file descriptors limit to more than '{}' to prevent "
-            "the "
-            "reduction.".format(jobs, estimated_descriptors_num + 10)
+        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:
+            jobs = safe_jobs_number
         return jobs

dvc/remote/ssh/__init__.py

@@ -40,6 +40,7 @@ class RemoteSSH(RemoteBASE):
     scheme = Schemes.SSH
     REQUIRES = {"paramiko": "paramiko"}
 
+    JOBS = 4
     PARAM_CHECKSUM = "md5"
     DEFAULT_PORT = 22
     TIMEOUT = 1800
@@ -50,10 +51,6 @@ class RemoteSSH(RemoteBASE):
 
     DEFAULT_CACHE_TYPES = ["copy"]
 
-    @property
-    def jobs(self):
-        return 4
-
     def __init__(self, repo, config):
         super(RemoteSSH, self).__init__(repo, config)
         url = config.get(Config.SECTION_REMOTE_URL)
@@ -328,10 +325,9 @@ class RemoteSSH(RemoteBASE):
             def exists_with_progress(chunks):
                 return self.batch_exists(chunks, callback=pbar.update_desc)
 
-            jobs = self._adjust_jobs(jobs)
-            with ThreadPoolExecutor(max_workers=jobs) as executor:
+            with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor:
                 path_infos = [self.checksum_to_path_info(x) for x in checksums]
-                chunks = to_chunks(path_infos, num_chunks=jobs)
+                chunks = to_chunks(path_infos, num_chunks=self.JOBS)
                 results = executor.map(exists_with_progress, chunks)
                 in_remote = itertools.chain.from_iterable(results)
                 ret = list(itertools.compress(checksums, in_remote))

tests/unit/remote/test_s3.py

@@ -2,7 +2,7 @@ import pytest
 
 from dvc.config import ConfigError
 from dvc.remote.s3 import RemoteS3
-
+from tests.utils import empty_caplog
 
 bucket_name = "bucket-name"
 prefix = "some/prefix"
@@ -54,3 +54,34 @@ 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)]
+)
+@patch("resource.getrlimit", return_value=(256, 1024))
+def test_adjust_default_jobs_number(
+    _, caplog, default_jobs_number, expected_result
+):
+    remote = RemoteS3(None, {})
+
+    with empty_caplog(caplog), patch.object(
+        remote, "JOBS", default_jobs_number
+    ):
+        assert remote.adjust_jobs() == expected_result
+
+
+@patch("resource.getrlimit", return_value=(256, 1024))
+def test_warn_on_too_many_jobs(_, caplog):
+    remote = RemoteS3(None, {})
+
+    with caplog.at_level(logging.INFO, "dvc"):
+        assert remote.adjust_jobs(64) == 64
+    assert len(caplog.messages) == 1
+    assert caplog.messages[0] == (
+        "Provided jobs number '64' might result "
+        "in 'Too many open files error'. "
+        "Consider decreasing jobs number to '12' "
+        "or increasing file descriptors limit to "
+        "'1290'."
+    )

tests/utils/__init__.py

@@ -45,3 +45,14 @@ 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 == ""