Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add timeout to SSH queue scan command #191

Merged
merged 11 commits into from
Jun 18, 2024
14 changes: 8 additions & 6 deletions sisyphus/aws_batch_engine.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -88,10 +88,12 @@ 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))

def fix_output(o):
"""
Expand All @@ -105,9 +107,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

Expand Down
12 changes: 7 additions & 5 deletions sisyphus/load_sharing_facility_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,10 +56,12 @@ 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))

def fix_output(o):
# split output and drop last empty line
Expand All @@ -69,9 +71,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_ = []
Expand Down
16 changes: 10 additions & 6 deletions sisyphus/simple_linux_utility_for_resource_management_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,16 +86,20 @@ 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:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we catch the subprocess.TimeoutExpired exception here and return, then the whole logic about retrying in submit_helper

while True:
try:
out, err, retval = self.system_call(sbatch_call)
except subprocess.TimeoutExpired:
logging.warning(self._system_call_timeout_warn_msg(command))
time.sleep(gs.WAIT_PERIOD_SSH_TIMEOUT)
continue
break
is obsolet.
If this was intentional, then maybe also remove this logic? and gs.WAIT_PERIOD_SSH_TIMEOUT would be ignored then

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah so it is actually handled. That was my earlier question about this. So then I would not catch it. Unless there is good reason to change this old behavior about TimeoutExpired.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes... I hadn't noticed that it would be catched later in the call stack. See #196.

logging.warning("Timeout expired for command: %s" % " ".join(system_command))
Icemole marked this conversation as resolved.
Show resolved Hide resolved

def fix_output(o):
"""
Expand All @@ -109,9 +113,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_ = []
Expand Down
16 changes: 10 additions & 6 deletions sisyphus/son_of_grid_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,16 +87,20 @@ 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))

def fix_output(o):
"""
Expand All @@ -110,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_ = []
Expand Down
Loading