| 561 | self._split_and_submit_tasks(tasks, batch_id=batch_id) |
| 562 | |
| 563 | def _split_and_submit_tasks(self, tasks: List[Task], batch_id: Union[int, str]) -> None: |
| 564 | for i, task in enumerate(tasks): |
| 565 | assert task.repeat_times is not None, "Task repeat_times should not be None" |
| 566 | task_wrapper = TaskWrapper( |
| 567 | task=replace(task, batch_id=batch_id, task_id=i), |
| 568 | batch_id=batch_id, |
| 569 | total_runs=task.repeat_times, |
| 570 | ) |
| 571 | if self.max_repeat_times is None: |
| 572 | task_wrapper.sub_task_num = 1 |
| 573 | self.pending_tasks[batch_id].appendleft((task_wrapper, task.repeat_times, 0)) |
| 574 | continue |
| 575 | sub_tasks = [] |
| 576 | for run_id_base in range(0, task.repeat_times, self.max_repeat_times): |
| 577 | repeat_times = min(self.max_repeat_times, task.repeat_times - run_id_base) |
| 578 | sub_tasks.append((task_wrapper, repeat_times, run_id_base)) |
| 579 | task_wrapper.sub_task_num = len(sub_tasks) |
| 580 | self.pending_tasks[batch_id].extendleft(sub_tasks) |
| 581 | |
| 582 | def dynamic_timeout(self, timeout: Optional[float] = None) -> float: |
| 583 | """Calculate dynamic timeout based on historical data.""" |