Repository: rwth-i6/sisyphus
Issue category: Performance 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.


Functions to edit:
sisyphus/aws_batch_engine.py:AWSBatchEngine.system_call
sisyphus/load_sharing_facility_engine.py:LoadSharingFacilityEngine.system_call
sisyphus/simple_linux_utility_for_resource_management_engine.py:SimpleLinuxUtilityForResourceManagementEngine.system_call
sisyphus/son_of_grid_engine.py:SonOfGridEngine.system_call

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_ = []