Safe Queue set exception to the future object linked to a job
| 170 | |
| 171 | |
| 172 | class _SafeQueue(Queue): |
| 173 | """Safe Queue set exception to the future object linked to a job""" |
| 174 | def __init__(self, max_size=0, *, ctx, pending_work_items, thread_wakeup): |
| 175 | self.pending_work_items = pending_work_items |
| 176 | self.thread_wakeup = thread_wakeup |
| 177 | super().__init__(max_size, ctx=ctx) |
| 178 | |
| 179 | def _on_queue_feeder_error(self, e, obj): |
| 180 | if isinstance(obj, _CallItem): |
| 181 | tb = format_exception(type(e), e, e.__traceback__) |
| 182 | e.__cause__ = _RemoteTraceback('\n"""\n{}"""'.format(''.join(tb))) |
| 183 | work_item = self.pending_work_items.pop(obj.work_id, None) |
| 184 | self.thread_wakeup.wakeup() |
| 185 | # work_item can be None if another process terminated. In this |
| 186 | # case, the executor_manager_thread fails all work_items |
| 187 | # with BrokenProcessPool |
| 188 | if work_item is not None: |
| 189 | work_item.future.set_exception(e) |
| 190 | else: |
| 191 | super()._on_queue_feeder_error(e, obj) |
| 192 | |
| 193 | |
| 194 | def _process_chunk(fn, chunk): |
no outgoing calls
no test coverage detected
searching dependent graphs…