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

Kill tasks during job prep. #5749

Closed
wants to merge 3 commits into from
Closed

Kill tasks during job prep. #5749

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8b6c5d0
Kill tasks during job prep.
hjoliver Sep 30, 2023
4be9c82
style fix
hjoliver Sep 30, 2023
118dd80
Log manual removal task proxy. [skip ci]
hjoliver Oct 1, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cylc/flow/scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -920,6 +920,7 @@ async def process_command_queue(self) -> None:
name, args, kwargs = command
except Empty:
break

args_string = ', '.join(str(a) for a in args)
kwargs_string = ', '.join(
f"{key}={value}" for key, value in kwargs.items()
Expand Down
16 changes: 15 additions & 1 deletion cylc/flow/task_job_mgr.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@
from cylc.flow.task_state import (
TASK_STATUS_PREPARING,
TASK_STATUS_SUBMITTED,
TASK_STATUS_SUBMIT_FAILED,
TASK_STATUS_RUNNING,
TASK_STATUS_WAITING,
TASK_STATUSES_ACTIVE
Expand Down Expand Up @@ -187,6 +188,9 @@ def kill_task_jobs(self, workflow, itasks):
itask.state_reset(is_held=True)
self.data_store_mgr.delta_task_held(itask)
to_kill_tasks.append(itask)
elif itask.state(TASK_STATUS_PREPARING):
itask.killed_in_job_prep = True
LOG.warning(f"[{itask}] killed in prep")
else:
LOG.warning(f"[{itask}] not killable")
self._run_job_cmd(
Expand Down Expand Up @@ -483,6 +487,16 @@ def submit_task_jobs(self, workflow, itasks, curve_auth,
)
continue

if itask.killed_in_job_prep:
Copy link
Member

Choose a reason for hiding this comment

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

itask is possibly unbound

itask.waiting_on_job_prep = False
itask.killed_in_job_prep = False
itask.state_reset(TASK_STATUS_SUBMIT_FAILED)
self.data_store_mgr.delta_task_state(itask)
itask.local_job_file_path = None # reset for retry
self._prep_submit_task_job_error(
workflow, itask, '(killed in job prep)', '')
Comment on lines +491 to +497
Copy link
Member

Choose a reason for hiding this comment

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

Another problem is this does not get run for tasks that are cylc removed in the preparing state, as they are removed from the pool and do not get to the submission stage

continue

# Build the "cylc jobs-submit" command
cmd = [self.JOBS_SUBMIT]
if LOG.isEnabledFor(DEBUG):
Expand Down Expand Up @@ -1087,7 +1101,7 @@ def _prep_submit_task_job(
Returns:
* itask - preparation complete.
* None - preparation in progress.
* False - perparation failed.
* False - preparation failed.

"""
if itask.local_job_file_path:
Expand Down
13 changes: 9 additions & 4 deletions cylc/flow/task_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -762,13 +762,18 @@ def spawn_to_rh_limit(self, tdef, point, flow_nums) -> None:
if ntask is not None:
self.add_to_pool(ntask)

def remove(self, itask, reason=""):
def remove(self, itask, reason="", manual=False):
"""Remove a task from the pool (e.g. after a reload)."""
self.tasks_removed = True
msg = "task proxy removed"
if reason:
msg += f" ({reason})"

if manual:
log = LOG.info
else:
log = LOG.debug

try:
del self.hidden_pool[itask.point][itask.identity]
except KeyError:
Expand All @@ -778,7 +783,7 @@ def remove(self, itask, reason=""):
self.hidden_pool_changed = True
if not self.hidden_pool[itask.point]:
del self.hidden_pool[itask.point]
LOG.debug(f"[{itask}] {msg}")
log(f"[{itask}] {msg}")
self.task_queue_mgr.remove_task(itask)
return

Expand All @@ -800,7 +805,7 @@ def remove(self, itask, reason=""):
# Event-driven final update of task_states table.
# TODO: same for datastore (still updated by scheduler loop)
self.workflow_db_mgr.put_update_task_state(itask)
LOG.debug(f"[{itask}] {msg}")
log(f"[{itask}] {msg}")
del itask

def get_all_tasks(self) -> List[TaskProxy]:
Expand Down Expand Up @@ -1663,7 +1668,7 @@ def remove_tasks(self, items):
"""Remove tasks from the pool."""
itasks, _, bad_items = self.filter_task_proxies(items)
for itask in itasks:
self.remove(itask, 'request')
self.remove(itask, 'request', manual=True)
if self.compute_runahead():
self.release_runahead_tasks()
return len(bad_items)
Expand Down
6 changes: 5 additions & 1 deletion cylc/flow/task_proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,11 +134,13 @@ class TaskProxy:
graph children: {msg: [(name, point), ...]}
.flow_nums:
flows I belong to
flow_wait:
.flow_wait:
wait for flow merge before spawning children
.waiting_on_job_prep:
True whilst task is awaiting job prep, reset to False once the
preparation has completed.
.killed_in_job_prep:
killed during job preparation; set to submit-failed once prepped

Args:
tdef: The definition object of this task.
Expand Down Expand Up @@ -180,6 +182,7 @@ class TaskProxy:
'tokens',
'try_timers',
'waiting_on_job_prep',
'killed_in_job_prep'
]

def __init__(
Expand Down Expand Up @@ -252,6 +255,7 @@ def __init__(
self.late_time: Optional[float] = None
self.is_late = is_late
self.waiting_on_job_prep = False
self.killed_in_job_prep = False

self.state = TaskState(tdef, self.point, status, is_held)

Expand Down