(self, async_task: asyncio.Task)
| 366 | del self.pending_tasks[batch_id] |
| 367 | |
| 368 | def task_done_callback(self, async_task: asyncio.Task): |
| 369 | running_state = self.running_task_state_map.pop(async_task) |
| 370 | task = running_state.task |
| 371 | runner_id = running_state.runner_id |
| 372 | if async_task.cancelled(): |
| 373 | if not running_state.restart_runner_on_cancel: |
| 374 | self.busy_runners.pop(runner_id, None) |
| 375 | self.idle_runners.add(runner_id) |
| 376 | elif async_task.exception(): |
| 377 | self.logger.error(f"Task {task.task.task_id} failed: {async_task.exception()}") |
| 378 | self._schedule_runner_restart(runner_id) |
| 379 | else: |
| 380 | status, exp_payload, runner_id, run_time = async_task.result() |
| 381 | if not task.task.is_eval: |
| 382 | self.total_running_time += run_time |
| 383 | self.total_completed_sub_tasks += 1 |
| 384 | self._accumulate_task_result(task, status, exp_payload) |
| 385 | self.busy_runners.pop(runner_id, None) |
| 386 | self.idle_runners.add(runner_id) |
| 387 | # If all sub runs in a task are completed |
| 388 | if task.finished_sub_task_num == task.sub_task_num: |
| 389 | if not task.task.is_eval: |
| 390 | self.total_completed_tasks += 1 |
| 391 | self._emit_task_result(task) |
| 392 | self.logger.debug(f"Task completed (batch_id {task.batch_id}).") |
| 393 | |
| 394 | if task.batch_id in self.running_tasks: |
| 395 | self.running_tasks[task.batch_id].remove(async_task) |
| 396 | if not self.running_tasks[task.batch_id]: |
| 397 | del self.running_tasks[task.batch_id] |
| 398 | |
| 399 | def _accumulate_task_result( |
| 400 | self, task: TaskWrapper, status: Status, experience_payload: bytes |
nothing calls this directly
no test coverage detected