Repository: iterative/dvc

Issue:
'Errno 24 - Too many open files' on dvc push
### Version information
* DVC version: 0.58.1
* Platform: MacOS 10.14.6
* Method of installation: pip within a conda environment

### Description
When pushing to S3 a directory of ~100 files that have been added to DVC, I observe an Errno 24 error from the dvc process.

It looks like dvc is trying to open more files than the OS allows. Checking the file handles on for the dvc process I get:

```
$ lsof -p $DVC_PID | wc -l
412
```

Looking at the OS limits, a process is limited to having 256 open files.

```
$ ulimit -a
core file size          (blocks, -c) 0
data seg size           (kbytes, -d) unlimited
file size               (blocks, -f) unlimited
max locked memory       (kbytes, -l) unlimited
max memory size         (kbytes, -m) unlimited
open files                      (-n) 256
pipe size            (512 bytes, -p) 1
stack size              (kbytes, -s) 8192
cpu time               (seconds, -t) unlimited
max user processes              (-u) 4256
virtual memory          (kbytes, -v) unlimited
```

A workaround for this is to increase the max files per process to a larger number (say 4096) by running something like `ulimit -n 4096`, but I wonder if the ideal solution is for DVC to work within the OS configured limits by default?

Edit: Updated wording of workaround

Discussion:
pared: Note: 
If we will agree that is how we want to handle s3, I will have to prepare change to docs about `push` and `pull`.
shcheklein: sorry guys, missed the discussion in the ticket. @pared could you please summarize the changes, show a gif on how would it look like? 

My 2cs - it feels a bit that we are over-architecting this if we try to analyze the system in advance. Why don't we just handle this gracefully and show a message in logs/screen that "you are hitting the limit blah - blah ..."?
pared: @shcheklein the whole implementation is a result of our discussion in the original issue.
We wanted to achieve two goals:
1. Prevent `Too many open files` from occurring
2. The default jobs number is pretty high (16) and we would like it to stay, to make default experience better.

The problem here is that we cannot for 100% say that this error will occur, which heavily depends on the size of particular files and speed of the network. We know that this error might occur when the number of jobs exceeds some (possible to estimate) number. However, user might be aware of that and purposefully increases the number of jobs because he knows that for some reason (super-fast network) this does not apply to him. So what I want to introduce in this change is:

1. Decrease automatically default number of jobs if user did not provide `jobs`, just to make his life easier, example:
https://asciinema.org/a/286748
(added print that its actually decreasing)

2. If everything seems fine, just use default jobs number:
https://asciinema.org/a/286749
(also this print will not be there)

3. In case of user defining `jobs` specifically, warn him:
https://asciinema.org/a/286750
Suor: I don't like this approach. This is:
- fragile, if something is happening in the bg, like backup or torrents) then a user will still get "too many files" 
- bothers user with some warnings 
- still stops on errors, this makes it non-reliable to run automatically unless you use conservative `jobs`.

So I would prefer dvc to retry on errors, adjust jobs automatically. Ideally we won't need `jobs` param at all.
pared: @Suor I don't like the retrying approach, because you cannot determine when and if the operation will fail. So it is possible, that someone will get an error after a few minutes of operation execution. And I think that would be really bad because the user would need to reduce number of jobs/increase fds limit, retry and hope for the best. I think that could be frustrating if faced a few times.
shcheklein: @pared thanks for the clarifications! Even though I tend to agree with @Suor that our future direction should be aggressive (and do not fail, but adjust itself dynamically), I think we should move forward (especially since it was discussed  by a few members of the team) with this and move @Suor 's general suggestion to a separate discussion and approach holistically.

@efiop just to clarify (since I missed the discussion), what's your take on this?
efiop: > fragile, if something is happening in the bg, like backup or torrents) then a user will still get "too many files"

@Suor ulimit is per-session, your torrents (aye-aye, captain Alex! ⚓️ 🚢 🔫 ☠️) and backups are in a different session, so they won't affect your dvc pull.

> bothers user with some warnings
> still stops on errors, this makes it non-reliable to run automatically unless you use conservative jobs.

Agreed on these two. The warning can be replaced with a post-crash one (see part about it below).

> So I would prefer dvc to retry on errors, adjust jobs automatically. Ideally we won't need jobs param at all.

Ideally yes, but that is a different ticket. With ulimit, it is hard to retry, as you can't really be sure if it won't fail again or after 10 minutes of wait time. I haven't tested it, so not 100% sure if that would be suitable for us or not, I guess it depends on where we are going to catch that exception.

@pared We've agreed to definitely catch that exception in `main.py` and show a meaningful warning to start with, right? I don't think anyone has any objections about that part that should come as a separate PR.
pared: @Suor @shcheklein @efiop 
Ok, so to sum it up:
1. Lets create new issue where we will handle OsError 24 and post-fail communication, and maybe automatic adjustment of jobs, post fail
2. Modify current change so that it does not log warnings, but leave adjusting default jobs number be.
How does it sound? 
pared: Note: I think I did not initially understand @Suor point of view, and imagined it as following workflow:
push -> fail -> push again (manually).
I was wrong here, right? In the end, we should let it fail and retry (automatically) upon "Too many open files" error. That should not penalize the whole operation, because push/pull operation as a whole is not success or fail operation, some of the files should be pushed/pulled.

Suor: @pared the retry thing may be definitely separated and might need to be handled along with retries for other network related reasons.

We can merge this as is or remove warning, both are fine with me.

@pared yes, I meant automatic retries.

efiop: @pared It still feels like catching the error and printing a message to increase the ulimit brings more value here. I'm worried about automatic adjustment like this because it is a lot of heuristics to handle. Maybe it is indeed the only way to do that, but it feels like if I was the user, if dvc told me "increase your ulimit" it would be more useful than the warning and the auto-throttle. Especially since we are talking mostly about mac, where ulimit for files is simply ridiculously low, so it makes total sense to just make user increase it.
pared: It seems that handling exceptions coming from parallel execution of tasks is causing some confusion.
1. We have ugly `try/ catch Exception` in [`_download_file`](https://github.com/iterative/dvc/blob/6e22769f3d8c940a5f2cf7e9855a8370ad4c2772/dvc/remote/base.py#L617) and in [`upload`](https://github.com/iterative/dvc/blob/6e22769f3d8c940a5f2cf7e9855a8370ad4c2772/dvc/remote/base.py#L540)
2. Till now we used to throw generic `[Download/Upload]Error` after all operations if some of them failed
3. I think that on `Too many open files` we could raise exception immediately.
4. Trying to introduce separate `Too many open files` error handling in `RemoteBASE` caused me to scatter logic responsible for error handling between `RemoteBASE` and `RemoteLOCAL` which obfuscates what is actually happening with the errors.

That brings me to the conclusion that we might need to introduce some kind of error handler to whom we could delegate logic related to error handling.

@iterative/engineering  What do you think about this idea?
efiop: @pared So for example, `_download` from RemoteS3 would catch its specific boto3.ClientError and maybe some other stuff that is not critical and so other workers can proceed working, while we in `download` in RemoteBASE would catch some generic issues (not sure which ones from the top of my head) and main.py would catch OSError for ulimit issue, right?
pared: @efiop yes, that's what I have in mind.

I also can't come up some common errors for all remotes, but handling "Unspecified" exceptions is common to all classes (raise `[Upload/ Download]Error` after all operations are performed) 
efiop: @pared Should be quite straight forward. E.g. for s3 it is ClientError, for other remotes it is also something specific that we can get straight from the docs. There will be learning curve though, but I think that is reasonable for the sake of proper error handling and not this current lazy "except Exception"(my fault originally).
pared: @efiop I agree that there will be a learning curve, though I believe that the current way of handling exceptions also requires some insight.
efiop: @pared Agreed. Ok, so on a quick google for upload/download:
s3 - ClientError
azure - ClientException
gs - ClientError
etc
So looks like it is pretty standardized for those clouds, which is pretty nice. As a first step, it would be reasonable to count those as recoverable ones (meaning that they would be logged and `return 1`, same as we do right now with all exceptions in workers) and the rest consider non-recoverable, that way we could catch ulimit error in main.py, as discussed earlier. What do you think?
pared: @efiop I would actually leave error as recoverable by default, and implement logic for those that we think should fail the whole operation, that seems to go along with logic we applied until this issue came up.
efiop: @pared Right, that is what you are basically doing right now. Makes sense. Seems like we agree on the throttling magic being too fragile, so maybe let's repurpose this PR to only properly handle the ulimit issue? If so, one thing I would do is to not create that exception, but rather catch that as an OSError in main.py and `logger.error()` it instead with that message. Doesn't look like we really need a special exception for it.

Patch:
diff --git a/dvc/main.py b/dvc/main.py
index 86e1a9e0d..89c9dc775 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, please increase your `ulimit`"
+            )
         else:
             logger.exception("unexpected error")
         ret = 255
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index f72e56127..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
@@ -516,6 +519,16 @@ class RemoteBASE(object):
             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 @@ class RemoteBASE(object):
                 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 @@ class RemoteBASE(object):
             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)
 


Test patch:
diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py
index 23f318861..f3140b5f3 100644
--- a/tests/func/test_remote.py
+++ b/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 test_partial_push_n_pull(dvc_repo, repo_dir, caplog):
 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