Repository: rwth-i6/sisyphus
Issue category: Performance Issue

Issue:
Too many open file descriptors
Hi, I was using sisyphus today for a big recipe and I got an error in my worker which claimed `too many open files`:

```
OSError: [Errno 24] Unable to synchronously open file (unable to open file: name = <filename>, errno = 24, error message = 'Too many open files', flags = 0, o_flags = 0)
```

However, not only had the worker crashed, but also the manager crashed with this error. Moreover, the terminal (tmux pane) entered a state where every character I typed was converted into a newline character. As a result, I couldn't type any instruction, so I ended up killing the tmux pane and recreating it.

I got into an investigation of what was happening and I developed a really small test:

```
# test.py

def py():
    pass
```

I found out that running the sisyphus manager on the test (bare; without any `settings.py`) opened ~3k files, from my baseline of 524 opened files to 3254 opened files after running `sis m test.py`, according to `lsof | grep <user> | wc`.

Besides that, I had the issue that every job triggered by the manager was adding exactly 105 opened files to the list of open file descriptors. However, I can't reproduce this starting from scratch, which leads me to think that it might be a problem about how our code base interacts with sisyphus (or just from our code base). I'll keep investigating and keep you tuned.

Is this issue about opening too many files by sisyphus intended because of some sisyphus caching strategy or related work? Was this ever addressed?

If you need more details, I'll be glad to provide them. Thank you in advance.


Hints:
I had this happen again. It was with a relatively big setup, but I'm not sure what causes the issue yet since my manager shouldn't be opening many files, if any. Please find attached the corresponding stack trace from the manager 
[here](https://github.com/user-attachments/files/15509563/manager_too_many_open_files_github.txt).

Note that the last newlines from the stack trace are relevant, since these represent me trying to write anything at all, and any character becoming a newline:

> Moreover, the terminal (tmux pane) entered a state where every character I typed was converted into a newline character. As a result, I couldn't type any instruction, so I ended up killing the tmux pane and recreating it.

I think this could be an interaction with the manager prompting me for my SSH password many times because I had left the tmux session, and then crashing, thus leaving the prompt in an unstable state (i.e. whatever you write when writing your SSH password key is transparent).
Analyzing the stack trace I found that both issues (too many open files, and ssh key prompt) could be related. What sisyphus seems to be trying to do after each password prompt is running a subprocess with the `squeue` command (I'm running in SLURM, but this also used to happen in SGE as well, so it should be cluster-independent). Right now I'm running my setup with the `gateway="..."` option in `settings.py`, but I recall it could have happened without such an option.

This happens for each 30 seconds, which is the time my sisyphus is configured to scan the queue. With an open file cap of 1024 in the manager (assuming sisyphus doesn't open any other files and any are opened to begin with), the time needed to reach the cap would be 1024 * 30 = 30k seconds = 8.5 hours. Even though the time cap is practically lower because there are more files opened in the manager, it makes sense given the lengths of duration in which I abandoned the ssh/tmux session (evening/night).

I'll try to solve it on my end, but I think it could also make sense to try to fix it in sisyphus. How can we tell sisyphus to wait before running a new queue scan job? Maybe storing the last queue command issued and setting it to `None` after completion?

Edit: a pretty easy solution would probably be setting the timeout of the corresponding SSH command queue scan to the number of seconds it takes for sisyphus to run a new queue command.

Patch:
diff --git a/sisyphus/aws_batch_engine.py b/sisyphus/aws_batch_engine.py
index 4b0173f..80f454e 100644
--- a/sisyphus/aws_batch_engine.py
+++ b/sisyphus/aws_batch_engine.py
@@ -1,4 +1,4 @@
-""" This is an experimental implementation for the aws batch engine.
+"""This is an experimental implementation for the aws batch engine.
 
 WARNING: After running some setups I can currently not recommend using aws batch with Sisyphus.
 AWS parallelcluster (https://aws.amazon.com/blogs/opensource/aws-parallelcluster/) looks like a easy way how
@@ -88,10 +88,13 @@ def system_call(self, command, send_to_stdin=None):
         system_command = command
 
         logging.debug("shell_cmd: %s" % " ".join(system_command))
-        p = subprocess.Popen(system_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         if send_to_stdin:
             send_to_stdin = send_to_stdin.encode()
-        out, err = p.communicate(input=send_to_stdin, timeout=30)
+        try:
+            p = subprocess.run(system_command, input=send_to_stdin, capture_output=True, timeout=30)
+        except subprocess.TimeoutExpired:
+            logging.warning("Timeout expired for command: %s" % " ".join(system_command))
+            return [], ["TimeoutExpired"], -1
 
         def fix_output(o):
             """
@@ -105,9 +108,9 @@ def fix_output(o):
                 assert False
             return o[:-1]
 
-        out = fix_output(out)
-        err = fix_output(err)
-        retval = p.wait(timeout=30)
+        out = fix_output(p.stdout)
+        err = fix_output(p.stderr)
+        retval = p.returncode
 
         return out, err, retval
 
diff --git a/sisyphus/load_sharing_facility_engine.py b/sisyphus/load_sharing_facility_engine.py
index 2b63318..5a24ad1 100644
--- a/sisyphus/load_sharing_facility_engine.py
+++ b/sisyphus/load_sharing_facility_engine.py
@@ -56,10 +56,13 @@ def system_call(self, command, send_to_stdin=None):
             system_command = command
 
         logging.debug("shell_cmd: %s" % " ".join(system_command))
-        p = subprocess.Popen(system_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         if send_to_stdin:
             send_to_stdin = send_to_stdin.encode()
-        out, err = p.communicate(input=send_to_stdin, timeout=30)
+        try:
+            p = subprocess.run(system_command, input=send_to_stdin, capture_output=True, timeout=30)
+        except subprocess.TimeoutExpired:
+            logging.warning("Timeout expired for command: %s" % " ".join(system_command))
+            return [], ["TimeoutExpired"], -1
 
         def fix_output(o):
             # split output and drop last empty line
@@ -69,9 +72,9 @@ def fix_output(o):
                 assert False
             return o[:-1]
 
-        out = fix_output(out)
-        err = fix_output(err)
-        retval = p.wait(timeout=30)
+        out = fix_output(p.stdout)
+        err = fix_output(p.stderr)
+        retval = p.returncode
 
         # Check for ssh error
         err_ = []
diff --git a/sisyphus/simple_linux_utility_for_resource_management_engine.py b/sisyphus/simple_linux_utility_for_resource_management_engine.py
index a918928..48ad8db 100644
--- a/sisyphus/simple_linux_utility_for_resource_management_engine.py
+++ b/sisyphus/simple_linux_utility_for_resource_management_engine.py
@@ -86,16 +86,21 @@ def system_call(self, command, send_to_stdin=None):
         """
         if self.gateway:
             escaped_command = [shlex.quote(s) for s in command]  # parameters need to be shell safe when sending via ssh
-            system_command = ["ssh", "-x", self.gateway] + [" ".join(["cd", os.getcwd(), "&&"] + escaped_command)]
+            system_command = ["ssh", "-x", self.gateway, "-o", "BatchMode=yes"] + [
+                " ".join(["cd", os.getcwd(), "&&"] + escaped_command)
+            ]
         else:
             # no gateway given, skip ssh local
             system_command = command
 
         logging.debug("shell_cmd: %s" % " ".join(system_command))
-        p = subprocess.Popen(system_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         if send_to_stdin:
             send_to_stdin = send_to_stdin.encode()
-        out, err = p.communicate(input=send_to_stdin, timeout=30)
+        try:
+            p = subprocess.run(system_command, input=send_to_stdin, capture_output=True, timeout=30)
+        except subprocess.TimeoutExpired:
+            logging.warning("Timeout expired for command: %s" % " ".join(system_command))
+            return [], ["TimeoutExpired"], -1
 
         def fix_output(o):
             """
@@ -109,9 +114,9 @@ def fix_output(o):
                 assert False
             return o[:-1]
 
-        out = fix_output(out)
-        err = fix_output(err)
-        retval = p.wait(timeout=30)
+        out = fix_output(p.stdout)
+        err = fix_output(p.stderr)
+        retval = p.returncode
 
         # Check for ssh error
         err_ = []
diff --git a/sisyphus/son_of_grid_engine.py b/sisyphus/son_of_grid_engine.py
index acda4f0..6682aba 100644
--- a/sisyphus/son_of_grid_engine.py
+++ b/sisyphus/son_of_grid_engine.py
@@ -87,16 +87,21 @@ def system_call(self, command, send_to_stdin=None):
         :rtype: list[bytes], list[bytes], int
         """
         if self.gateway:
-            system_command = ["ssh", "-x", self.gateway] + [" ".join(["cd", os.getcwd(), "&&"] + command)]
+            system_command = ["ssh", "-x", self.gateway, "-o", "BatchMode=yes"] + [
+                " ".join(["cd", os.getcwd(), "&&"] + command)
+            ]
         else:
             # no gateway given, skip ssh local
             system_command = command
 
         logging.debug("shell_cmd: %s" % " ".join(system_command))
-        p = subprocess.Popen(system_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         if send_to_stdin:
             send_to_stdin = send_to_stdin.encode()
-        out, err = p.communicate(input=send_to_stdin, timeout=30)
+        try:
+            p = subprocess.run(system_command, input=send_to_stdin, capture_output=True, timeout=30)
+        except subprocess.TimeoutExpired:
+            logging.warning("Timeout expired for command: %s" % " ".join(system_command))
+            return [], ["TimeoutExpired"], -1
 
         def fix_output(o):
             """
@@ -110,9 +115,9 @@ def fix_output(o):
                 assert False
             return o[:-1]
 
-        out = fix_output(out)
-        err = fix_output(err)
-        retval = p.wait(timeout=30)
+        out = fix_output(p.stdout)
+        err = fix_output(p.stderr)
+        retval = p.returncode
 
         # Check for ssh error
         err_ = []


Test patch: