Executes the tasks in the current process. Multiple tasks can be passed for batch processing. However, they must all use the same function and will share the execution entry.
(self, tasks: List[Task], log: BoundLogger)
| 108 | raise NotImplementedError |
| 109 | |
| 110 | def execute_tasks(self, tasks: List[Task], log: BoundLogger) -> bool: |
| 111 | """ |
| 112 | Executes the tasks in the current process. Multiple tasks can be passed |
| 113 | for batch processing. However, they must all use the same function and |
| 114 | will share the execution entry. |
| 115 | """ |
| 116 | success = False |
| 117 | |
| 118 | execution: Dict[str, Any] = {} |
| 119 | |
| 120 | assert len(tasks) |
| 121 | task_func = tasks[0].serialized_func |
| 122 | assert all([task_func == task.serialized_func for task in tasks[1:]]) |
| 123 | |
| 124 | execution["time_started"] = time.time() |
| 125 | |
| 126 | try: |
| 127 | func = tasks[0].func |
| 128 | |
| 129 | runner_class = get_runner_class(log, tasks) |
| 130 | runner = runner_class(self.tiger) |
| 131 | |
| 132 | is_batch_func = getattr(func, "_task_batch", False) |
| 133 | g["tiger"] = self.tiger |
| 134 | g["current_task_is_batch"] = is_batch_func |
| 135 | |
| 136 | hard_timeouts = self.worker.get_hard_timeouts(func, tasks) |
| 137 | |
| 138 | with WorkerContextManagerStack(self.config["CHILD_CONTEXT_MANAGERS"]): |
| 139 | if is_batch_func: |
| 140 | # Batch process if the task supports it. |
| 141 | g["current_tasks"] = tasks |
| 142 | runner.run_batch_tasks(tasks, hard_timeouts[0]) |
| 143 | else: |
| 144 | # Process sequentially. |
| 145 | for task, hard_timeout in zip(tasks, hard_timeouts): |
| 146 | g["current_tasks"] = [task] |
| 147 | runner.run_single_task(task, hard_timeout) |
| 148 | |
| 149 | except RetryException as exc: |
| 150 | execution["retry"] = True |
| 151 | if exc.method: |
| 152 | execution["retry_method"] = serialize_retry_method(exc.method) |
| 153 | execution["log_error"] = exc.log_error |
| 154 | execution["exception_name"] = serialize_func_name(exc.__class__) |
| 155 | exc_info = exc.exc_info or sys.exc_info() |
| 156 | except (JobTimeoutException, Exception) as exc: |
| 157 | execution["exception_name"] = serialize_func_name(exc.__class__) |
| 158 | exc_info = sys.exc_info() |
| 159 | else: |
| 160 | success = True |
| 161 | |
| 162 | if not success: |
| 163 | execution["time_failed"] = time.time() |
| 164 | if self.worker.store_tracebacks: |
| 165 | # Currently we only log failed task executions to Redis. |
| 166 | execution["traceback"] = "".join(traceback.format_exception(*exc_info)) |
| 167 | execution["success"] = success |
no test coverage detected