remote: s3: adjust jobs number basing on file descriptors number

Author: pared

dvc/remote/base.py

@@ -13,6 +13,8 @@
 
 from shortuuid import uuid
 
+from funcy import cached_property
+
 import dvc.prompt as prompt
 from dvc.config import Config
 from dvc.exceptions import (
@@ -75,7 +77,6 @@ class RemoteBASE(object):
     scheme = "base"
     path_cls = URLInfo
     REQUIRES = {}
-    JOBS = 4 * cpu_count()
 
     PARAM_RELPATH = "relpath"
     CHECKSUM_DIR_SUFFIX = ".dir"
@@ -84,6 +85,10 @@ class RemoteBASE(object):
 
     state = StateNoop()
 
+    @cached_property
+    def jobs(self):
+        return cpu_count() * 4
+
     def __init__(self, repo, config):
         self.repo = repo
 
@@ -746,6 +751,11 @@ def changed_cache(self, checksum):
             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.
 
@@ -784,7 +794,9 @@ def exists_with_progress(path_info):
                 pbar.update_desc(str(path_info))
                 return ret
 
-            with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor:
+            with ThreadPoolExecutor(
+                max_workers=self._adjust_jobs(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

@@ -339,7 +339,7 @@ def _process(
             status = STATUS_NEW
 
         if jobs is None:
-            jobs = remote.JOBS
+            jobs = remote.jobs
 
         status_info = self.status(
             named_cache,

dvc/remote/s3.py

@@ -3,6 +3,7 @@
 
 import logging
 import os
+import resource
 import threading
 
 from funcy import cached_property, wrap_prop
@@ -335,3 +336,20 @@ def _append_aws_grants_to_extra_args(self, config):
                     )
 
                 self.extra_args[extra_args_key] = config.get(grant_option)
+
+    def _adjust_jobs(self, jobs=None):
+        jobs = super(RemoteS3, self)._adjust_jobs(jobs)
+
+        descriptor_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
+        estimated_descriptors_num = jobs * 20
+        if estimated_descriptors_num <= descriptor_limit - 10:
+            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)
+        )
+        return jobs

dvc/remote/ssh/__init__.py

@@ -40,7 +40,6 @@ class RemoteSSH(RemoteBASE):
     scheme = Schemes.SSH
     REQUIRES = {"paramiko": "paramiko"}
 
-    JOBS = 4
     PARAM_CHECKSUM = "md5"
     DEFAULT_PORT = 22
     TIMEOUT = 1800
@@ -51,6 +50,10 @@ 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)
@@ -325,9 +328,10 @@ def cache_exists(self, checksums, jobs=None, name=None):
             def exists_with_progress(chunks):
                 return self.batch_exists(chunks, callback=pbar.update_desc)
 
-            with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor:
+            jobs = self._adjust_jobs(jobs)
+            with ThreadPoolExecutor(max_workers=jobs) as executor:
                 path_infos = [self.checksum_to_path_info(x) for x in checksums]
-                chunks = to_chunks(path_infos, num_chunks=self.JOBS)
+                chunks = to_chunks(path_infos, num_chunks=jobs)
                 results = executor.map(exists_with_progress, chunks)
                 in_remote = itertools.chain.from_iterable(results)
                 ret = list(itertools.compress(checksums, in_remote))