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

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, please increase your `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
@@ -516,6 +519,16 @@ def _save(self, path_info, checksum):
             return
         self._save_file(path_info, checksum)
 
+    def _handle_transfer_exception(
+        self, from_info, to_info, exception, operation
+    ):
+        if isinstance(exception, OSError) and exception.errno == errno.EMFILE:
+            raise exception
+
+        msg = "failed to {} '{}' to '{}'".format(operation, from_info, to_info)
+        logger.exception(msg)
+        return 1
+
     def upload(self, from_info, to_info, name=None, no_progress_bar=False):
         if not hasattr(self, "_upload"):
             raise RemoteActionNotImplemented("upload", self.scheme)
@@ -537,10 +550,10 @@ def upload(self, from_info, to_info, name=None, no_progress_bar=False):
                 name=name,
                 no_progress_bar=no_progress_bar,
             )
-        except Exception:
-            msg = "failed to upload '{}' to '{}'"
-            logger.exception(msg.format(from_info, to_info))
-            return 1  # 1 fail
+        except Exception as e:
+            return self._handle_transfer_exception(
+                from_info, to_info, e, "upload"
+            )
 
         return 0
 
@@ -614,10 +627,10 @@ def _download_file(
             self._download(
                 from_info, tmp_file, name=name, no_progress_bar=no_progress_bar
             )
-        except Exception:
-            msg = "failed to download '{}' to '{}'"
-            logger.exception(msg.format(from_info, to_info))
-            return 1  # 1 fail
+        except Exception as e:
+            return self._handle_transfer_exception(
+                from_info, to_info, e, "download"
+            )
 
         move(tmp_file, to_info, mode=file_mode)
 

tests/func/test_remote.py

@@ -1,14 +1,17 @@
+import errno
 import os
 import shutil
 
 import configobj
+import pytest
 from mock import patch
 
 from dvc.config import Config
 from dvc.main import main
 from dvc.path_info import PathInfo
-from dvc.remote import RemoteLOCAL
+from dvc.remote import RemoteLOCAL, RemoteConfig
 from dvc.remote.base import RemoteBASE
+from dvc.utils.compat import fspath
 from tests.basic_env import TestDvc
 from tests.remotes import get_local_url, get_local_storagepath
 
@@ -253,3 +256,21 @@ def unreliable_upload(self, from_file, to_info, name=None, **kwargs):
 def get_last_exc(caplog):
     _, exc, _ = caplog.records[-2].exc_info
     return exc
+
+
+def test_raise_on_too_many_open_files(tmp_dir, dvc, tmp_path_factory, mocker):
+    storage = tmp_path_factory.mktemp("test_remote_base")
+    remote_config = RemoteConfig(dvc.config)
+    remote_config.add("local_remote", fspath(storage), default=True)
+
+    tmp_dir.dvc_gen({"file": "file content"})
+
+    mocker.patch.object(
+        RemoteLOCAL,
+        "_upload",
+        side_effect=OSError(errno.EMFILE, "Too many open files"),
+    )
+
+    with pytest.raises(OSError) as e:
+        dvc.push()
+        assert e.errno == errno.EMFILE