Merge pull request #20522 from mvdbeek/improve_workflow_scheduling_iteration_loop

[25.0] Improve workflow monitor loop times

Author: mvdbeek

lib/galaxy/model/__init__.py

@@ -9000,6 +9000,13 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
     states = InvocationState
     non_terminal_states = [states.NEW, states.READY]
 
+    def get_last_workflow_invocation_step_update_time(self) -> Optional[datetime]:
+        session = required_object_session(self)
+        stmt = select(func.max(WorkflowInvocationStep.update_time)).where(
+            WorkflowInvocationStep.workflow_invocation_id == self.id
+        )
+        return session.execute(stmt).scalar_one_or_none()
+
     def create_subworkflow_invocation_for_step(self, step):
         assert step.type == "subworkflow"
         subworkflow_invocation = WorkflowInvocation()

lib/galaxy/workflow/scheduling_manager.py

@@ -1,6 +1,13 @@
 import os
+from datetime import (
+    datetime,
+    timedelta,
+)
 from functools import partial
-from typing import Optional
+from typing import (
+    Dict,
+    Optional,
+)
 
 from sqlalchemy.orm import Session
 
@@ -30,6 +37,7 @@
 
 DEFAULT_SCHEDULER_ID = "default"  # well actually this should be called DEFAULT_DEFAULT_SCHEDULER_ID...
 DEFAULT_SCHEDULER_PLUGIN_TYPE = "core"
+DEFAULT_SCHEDULER_BACKFILL_SECONDS = int(os.getenv("GALAXY_SCHEDULER_BACKFILL_SECONDS", 300))
 
 EXCEPTION_MESSAGE_SHUTDOWN = "Exception raised while attempting to shutdown workflow scheduler."
 EXCEPTION_MESSAGE_NO_SCHEDULERS = "Failed to defined workflow schedulers - no workflow schedulers defined."
@@ -296,6 +304,8 @@ def __init__(self, app: MinimalManagerApp, workflow_scheduling_manager):
             name="WorkflowRequestMonitor.monitor_thread", target=self.__monitor, config=app.config
         )
         self.invocation_grabber = None
+        self.update_time_tracking_dict: Dict[int, datetime] = {}
+        self.timedelta = timedelta(seconds=DEFAULT_SCHEDULER_BACKFILL_SECONDS)
         self_handler_tags = set(self.app.job_config.self_handler_tags)
         self_handler_tags.add(self.workflow_scheduling_manager.default_handler_id)
         handler_assignment_method = InvocationGrabber.get_grabbable_handler_assignment_method(
@@ -310,6 +320,29 @@ def __init__(self, app: MinimalManagerApp, workflow_scheduling_manager):
                 handler_tags=self_handler_tags,
             )
 
+    def ready_to_schedule_more(self, invocation: model.WorkflowInvocation):
+        # Improve reactivity of scheduling using the history update_time as a heuristic.
+        # If there wasn't a change in the history we're unlikely to be able to make more progress.
+        if invocation.id not in self.update_time_tracking_dict:
+            return True
+        else:
+            last_schedule_time = self.update_time_tracking_dict[invocation.id]
+            last_history_update_time = invocation.history.update_time
+            do_schedule = last_history_update_time > last_schedule_time
+            if not do_schedule and (
+                invocation_step_update_time := invocation.get_last_workflow_invocation_step_update_time()
+            ):
+                do_schedule = invocation_step_update_time > last_schedule_time
+            if not do_schedule and (datetime.now() - last_schedule_time) > self.timedelta:
+                # If we haven't scheduled in a while, schedule anyway.
+                log.debug(
+                    "Scheduling workflow invocation [%s] after %s seconds without scheduling.",
+                    invocation.id,
+                    (datetime.now() - last_schedule_time).total_seconds(),
+                )
+                do_schedule = True
+            return do_schedule
+
     def __monitor(self):
         to_monitor = self.workflow_scheduling_manager.active_workflow_schedulers
         while self.monitor_running:
@@ -393,9 +426,11 @@ def __attempt_schedule(self, invocation_id, workflow_scheduler):
                     workflow_invocation.cancel_invocation_steps()
                     workflow_invocation.mark_cancelled()
                     session.commit()
+                    self.update_time_tracking_dict.pop(invocation_id, None)
                     return False
 
                 if not workflow_invocation or not workflow_invocation.active:
+                    self.update_time_tracking_dict.pop(invocation_id, None)
                     return False
 
                 # This ensures we're only ever working on the 'first' active
@@ -405,9 +440,12 @@ def __attempt_schedule(self, invocation_id, workflow_scheduler):
                     for i in workflow_invocation.history.workflow_invocations:
                         if i.active and i.id < workflow_invocation.id:
                             return False
-                workflow_scheduler.schedule(workflow_invocation)
-                log.debug("Workflow invocation [%s] scheduled", workflow_invocation.id)
+                if self.ready_to_schedule_more(workflow_invocation):
+                    self.update_time_tracking_dict[invocation_id] = datetime.now()
+                    workflow_scheduler.schedule(workflow_invocation)
+                    log.debug("Workflow invocation [%s] scheduled", invocation_id)
             except Exception:
+                self.update_time_tracking_dict.pop(invocation_id, None)
                 # TODO: eventually fail this - or fail it right away?
                 log.exception("Exception raised while attempting to schedule workflow request.")
                 return False