improved scheduler task sync

Author: LanderOtto

streamflow/scheduling/scheduler.py

@@ -40,6 +40,7 @@ def __init__(
         super().__init__(context)
         self.binding_filter_map: MutableMapping[str, BindingFilter] = {}
         self.pending_jobs: MutableMapping[str, MutableSequence[JobContext]] = {}
+        self.pending_jobs_conditional = asyncio.Condition()
         self.pending_job_event = asyncio.Event()
         self.policy_map: MutableMapping[str, Policy] = {}
         self.retry_interval: int | None = retry_delay if retry_delay != 0 else None
@@ -181,7 +182,12 @@ async def _get_available_locations(
     async def _scheduling_task(self):
         try:
             while True:
+                # Event that awake the scheduling task:
+                #   - there is a new job to schedule
+                #   - some resources are released and there are some pending jobs
                 await self.pending_job_event.wait()
+                async with self.pending_jobs_conditional:
+                    self.pending_job_event.clear()
                 for deployment_name, job_contexts in self.pending_jobs.items():
                     logger.info("Start scheduling")
 
@@ -230,14 +236,9 @@ async def _scheduling_task(self):
                             job_context.targets[0],
                         )
                         job_context.event.set()
-
-                        # todo: awake scheduling:
-                        #   - there is a new job to schedule
-                        #   - some resources are released and there are some pending jobs
-                        # self.pending_job_event.clear()
-                logger.info("Sleep")
-                await asyncio.sleep(0.2)
+                        # todo: fix job with multi-targets case
         except Exception as e:
+            # todo: propagate error to context (?)
             logger.exception(f"Scheduler failed: {e}")
             raise
 
@@ -260,8 +261,9 @@ async def notify_status(self, job_name: str, status: Status) -> None:
                 if logger.isEnabledFor(logging.DEBUG):
                     logger.debug(f"Job {job_name} changed status to {status.name}")
 
-                # Notify scheduling loop: there are free resources
-                self.pending_job_event.set()
+                # Notify scheduling task: there are free resources
+                async with self.pending_jobs_conditional:
+                    self.pending_job_event.set()
 
     async def schedule(
         self,
@@ -278,8 +280,9 @@ async def schedule(
             deployment = target.deployment
             self.pending_jobs.setdefault(deployment.name, []).append(job_context)
 
-        # Notify scheduling loop: there is a job to schedule
-        self.pending_job_event.set()
+        # Notify scheduling task: there is a job to schedule
+        async with self.pending_jobs_conditional:
+            self.pending_job_event.set()
 
         # Wait the job is scheduled
         await job_context.event.wait()