diff --git a/src/exo/worker/runner/runner.py b/src/exo/worker/runner/runner.py index bf272aaa..61205f38 100644 --- a/src/exo/worker/runner/runner.py +++ b/src/exo/worker/runner/runner.py @@ -37,6 +37,7 @@ from exo.shared.types.tasks import ( Shutdown, StartWarmup, Task, + TaskId, TaskStatus, ) from exo.shared.types.worker.instances import BoundInstance @@ -111,8 +112,12 @@ def main( event_sender.send( RunnerStatusUpdated(runner_id=runner_id, runner_status=current_status) ) + seen = set[TaskId]() with task_receiver as tasks: for task in tasks: + if task.task_id in seen: + logger.warning("repeat task - potential error") + seen.add(task.task_id) event_sender.send( TaskStatusUpdated(task_id=task.task_id, task_status=TaskStatus.Running) ) diff --git a/src/exo/worker/runner/runner_supervisor.py b/src/exo/worker/runner/runner_supervisor.py index fc17cddc..a1951d80 100644 --- a/src/exo/worker/runner/runner_supervisor.py +++ b/src/exo/worker/runner/runner_supervisor.py @@ -127,20 +127,25 @@ class RunnerSupervisor: self._tg.cancel_scope.cancel() async def start_task(self, task: Task): + if task.task_id in self.pending: + logger.warning( + f"Skipping invalid task {task} as it has already been submitted" + ) + return if task.task_id in self.completed: - logger.info( + logger.warning( f"Skipping invalid task {task} as it has already been completed" ) + return logger.info(f"Starting task {task}") event = anyio.Event() self.pending[task.task_id] = event try: - self._task_sender.send(task) + await self._task_sender.send_async(task) except ClosedResourceError: logger.warning(f"Task {task} dropped, runner closed communication.") return await event.wait() - logger.info(f"Finished task {task}") async def _forward_events(self): with self._ev_recv as events: