| 86 | |
| 87 | |
| 88 | class ProcessContext: |
| 89 | def __init__(self, processes, error_queues): |
| 90 | self.error_queues = error_queues |
| 91 | self.processes = processes |
| 92 | self.sentinels = { |
| 93 | process.sentinel: index for index, process in enumerate(processes) |
| 94 | } |
| 95 | |
| 96 | def pids(self): |
| 97 | return [int(process.pid) for process in self.processes] |
| 98 | |
| 99 | def join(self, timeout=None): |
| 100 | r""" |
| 101 | Tries to join one or more processes in this spawn context. |
| 102 | If one of them exited with a non-zero exit status, this function |
| 103 | kills the remaining processes and raises an exception with the cause |
| 104 | of the first process exiting. |
| 105 | |
| 106 | Returns ``True`` if all processes have been joined successfully, |
| 107 | ``False`` if there are more processes that need to be joined. |
| 108 | |
| 109 | Args: |
| 110 | timeout (float): Wait this long before giving up on waiting. |
| 111 | """ |
| 112 | # Ensure this function can be called even when we're done. |
| 113 | if len(self.sentinels) == 0: |
| 114 | return True |
| 115 | |
| 116 | # Wait for any process to fail or all of them to succeed. |
| 117 | ready = multiprocessing.connection.wait(self.sentinels.keys(), timeout=timeout,) |
| 118 | |
| 119 | error_index = None |
| 120 | for sentinel in ready: |
| 121 | index = self.sentinels.pop(sentinel) |
| 122 | process = self.processes[index] |
| 123 | process.join() |
| 124 | if process.exitcode != 0: |
| 125 | error_index = index |
| 126 | break |
| 127 | |
| 128 | # Return if there was no error. |
| 129 | if error_index is None: |
| 130 | # Return whether or not all processes have been joined. |
| 131 | return len(self.sentinels) == 0 |
| 132 | |
| 133 | # Assume failure. Terminate processes that are still alive. |
| 134 | for process in self.processes: |
| 135 | if process.is_alive(): |
| 136 | process.terminate() |
| 137 | process.join() |
| 138 | |
| 139 | # There won't be an error on the queue if the process crashed. |
| 140 | failed_process = self.processes[error_index] |
| 141 | if self.error_queues[error_index].empty(): |
| 142 | exitcode = self.processes[error_index].exitcode |
| 143 | if exitcode < 0: |
| 144 | name = signal.Signals(-exitcode).name |
| 145 | raise ProcessExitedException( |