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.

Sign up for GitHub

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

Jump to bottom

fix: efficient job status checking when using DRMAA API (this should yield much better parallelization and performance when using --drmaa) #1156

Merged
merged 3 commits into from Sep 3, 2021
Merged

fix: efficient job status checking when using DRMAA API (this should yield much better parallelization and performance when using --drmaa) #1156

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5f9240d
fix drmaa job status checking to not wait but return immediately. Han…
johanneskoester Aug 27, 2021
de85745
refactor
johanneskoester Aug 27, 2021
0b1bc6c
fix potential key error when removing suspended message that has neve…
johanneskoester Aug 31, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
42 changes: 31 additions & 11 deletions snakemake/executors/__init__.py
View file
Open in desktop
Expand Up @@ -1402,6 +1402,8 @@ def shutdown(self):
def _wait_for_jobs(self):
import drmaa

suspended_msg = set()

while True:
with self.lock:
if not self.wait:
Expand All @@ -1412,9 +1414,7 @@ def _wait_for_jobs(self):
for active_job in active_jobs:
with self.status_rate_limiter:
try:
retval = self.session.wait(
active_job.jobid, drmaa.Session.TIMEOUT_NO_WAIT
)
retval = self.session.jobStatus(active_job.jobid)
except drmaa.ExitTimeoutException as e:
# job still active
still_running.append(active_job)
Expand All @@ -1427,20 +1427,40 @@ def _wait_for_jobs(self):
os.remove(active_job.jobscript)
active_job.error_callback(active_job.job)
continue
# job exited
os.remove(active_job.jobscript)
if (
not retval.wasAborted
and retval.hasExited
and retval.exitStatus == 0
):
if retval == drmaa.JobState.DONE:
os.remove(active_job.jobscript)
active_job.callback(active_job.job)
else:
elif retval == drmaa.JobState.FAILED:
os.remove(active_job.jobscript)
self.print_job_error(active_job.job)
self.print_cluster_job_error(
active_job, self.dag.jobid(active_job.job)
)
active_job.error_callback(active_job.job)
else:
# still running
still_running.append(active_job)

def handle_suspended(by):
if active_job.job.jobid not in suspended_msg:
logger.warning(
"Job {} (DRMAA id: {}) was suspended by {}.".format(
active_job.job.jobid, active_job.jobid, by
)
)
suspended_msg.add(active_job.job.jobid)

if retval == drmaa.JobState.USER_SUSPENDED:
handle_suspended("user")
elif retval == drmaa.JobState.SYSTEM_SUSPENDED:
handle_suspended("system")
else:
try:
suspended_msg.remove(active_job.job.jobid)
except KeyError:
# there was nothing to remove
pass

with self.lock:
self.active_jobs.extend(still_running)
sleep()
Expand Down