Adds a new task to the task dictionary. Args: dependency_task_id (List[int]): List of task IDs that the new task depends on. extra (Any, optional): Extra information associated with the task. Defaults to None. Returns: int: The ID of the
(self, dependency_task_id: List[int], extra=None)
| 41 | return len(self.task_dict) == 0 |
| 42 | |
| 43 | def add_task(self, dependency_task_id: List[int], extra=None) -> int: |
| 44 | """ |
| 45 | Adds a new task to the task dictionary. |
| 46 | |
| 47 | Args: |
| 48 | dependency_task_id (List[int]): List of task IDs that the new task depends on. |
| 49 | extra (Any, optional): Extra information associated with the task. Defaults to None. |
| 50 | |
| 51 | Returns: |
| 52 | int: The ID of the newly added task. |
| 53 | """ |
| 54 | with self.task_lock: |
| 55 | depend_tasks = [self.task_dict[task_id] for task_id in dependency_task_id] |
| 56 | self.task_dict[self.now_id] = Task( |
| 57 | task_id=self.now_id, dependencies=depend_tasks, extra_info=extra |
| 58 | ) |
| 59 | self.now_id += 1 |
| 60 | return self.now_id - 1 |
| 61 | |
| 62 | def get_next_task(self, process_id: int): |
| 63 | """ |
no test coverage detected