Launch a single task. Args: task (BaseTask): Task to launch. tokens (SyncManager.Semaphore): Multiprocessing semaphore for every subprocess to follow. Returns: tuple[str, int]: Task name and exit code.
(task: BaseTask, tokens: SyncManager.Semaphore)
| 93 | |
| 94 | |
| 95 | def launch(task: BaseTask, tokens: SyncManager.Semaphore): |
| 96 | """Launch a single task. |
| 97 | |
| 98 | Args: |
| 99 | task (BaseTask): Task to launch. |
| 100 | tokens (SyncManager.Semaphore): Multiprocessing semaphore |
| 101 | for every subprocess to follow. |
| 102 | |
| 103 | Returns: |
| 104 | tuple[str, int]: Task name and exit code. |
| 105 | """ |
| 106 | |
| 107 | task_name = task.name |
| 108 | returncode = 0 |
| 109 | logger = get_logger() |
| 110 | |
| 111 | try: |
| 112 | # get log file and redirect stdout and stderr |
| 113 | out_path = task.get_log_path(file_extension='out') |
| 114 | mmengine.mkdir_or_exist(osp.split(out_path)[0]) |
| 115 | redirect_std_to_file(out_path) |
| 116 | |
| 117 | # start infer with monkey_run |
| 118 | start_time = time.time() |
| 119 | inferencer = OpenICLInferTask(task.cfg) |
| 120 | origin_run = inferencer.run |
| 121 | inferencer.run = monkey_run |
| 122 | inferencer.run(inferencer, tokens) |
| 123 | inferencer.run = origin_run |
| 124 | end_time = time.time() |
| 125 | logger.info(f'time elapsed: {end_time - start_time:.2f}s') |
| 126 | except Exception: |
| 127 | # print trace back in target file |
| 128 | traceback.print_exc() |
| 129 | # reset stdout and stderr |
| 130 | reset_std() |
| 131 | logger.error(f'task {task_name} fail, see\n{out_path}') |
| 132 | returncode = 1 |
| 133 | else: |
| 134 | # reset stdout and stderr |
| 135 | reset_std() |
| 136 | return task_name, returncode |
| 137 | |
| 138 | |
| 139 | def submit(task, type, tokens): |
no test coverage detected