diff --git a/dvc/exceptions.py b/dvc/exceptions.py index 1373179f5..1beb8d6e1 100644 --- a/dvc/exceptions.py +++ b/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, - ) diff --git a/dvc/main.py b/dvc/main.py index 86e1a9e0d..0b4f95344 100644 --- a/dvc/main.py +++ b/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 diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 5f184e4f3..57a1846a6 100644 --- a/dvc/remote/base.py +++ b/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) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index c98a87f90..9c68e3a80 100644 --- a/dvc/remote/local.py +++ b/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, diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 8d9fccb47..6366fc426 100644 --- a/dvc/remote/s3.py +++ b/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 diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py index cf41a5f26..f3140b5f3 100644 --- a/tests/func/test_remote.py +++ b/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 diff --git a/tests/unit/remote/test_s3.py b/tests/unit/remote/test_s3.py index ddcf5e7f8..57fcfcec7 100644 --- a/tests/unit/remote/test_s3.py +++ b/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 diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py index d4569049a..a40a91f2c 100644 --- a/tests/utils/__init__.py +++ b/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 == ""