From c167eb2675639f80d4154fd93f1a312e73941d1a Mon Sep 17 00:00:00 2001 From: pawel Date: Thu, 5 Dec 2019 14:45:13 +0100 Subject: [PATCH] remote: s3: adjust jobs only on default --- dvc/remote/base.py | 15 ++++-------- dvc/remote/local.py | 3 +-- dvc/remote/s3.py | 46 +++++++++++++++++++++++++++--------- dvc/remote/ssh/__init__.py | 10 +++----- tests/unit/remote/test_s3.py | 33 +++++++++++++++++++++++++- tests/utils/__init__.py | 11 +++++++++ 6 files changed, 86 insertions(+), 32 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 85f6401c4..81e22fa72 100644 --- a/dvc/remote/base.py +++ b/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)) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 40c4cd31b..c98a87f90 100644 --- a/dvc/remote/local.py +++ b/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, diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 329b9a14f..6366320fe 100644 --- a/dvc/remote/s3.py +++ b/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 diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index 87440672e..78da815e1 100644 --- a/dvc/remote/ssh/__init__.py +++ b/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)) diff --git a/tests/unit/remote/test_s3.py b/tests/unit/remote/test_s3.py index 57fcfcec7..94bc0ddbf 100644 --- a/tests/unit/remote/test_s3.py +++ b/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'." + ) \ No newline at end of file diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py index a40a91f2c..d4569049a 100644 --- a/tests/utils/__init__.py +++ b/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 == ""