(self)
| 86 | self.start_thread(self.await_response_thread) |
| 87 | |
| 88 | def shutdown(self): |
| 89 | |
| 90 | if self.doing_shutdown: |
| 91 | return |
| 92 | else: |
| 93 | self.doing_shutdown = True |
| 94 | |
| 95 | logger_debug(f'Worker {mpi_rank()} shutdown...\n', "yellow") |
| 96 | |
| 97 | if self.engine is not None: |
| 98 | if self.engine.can_enqueue_requests(): |
| 99 | if self.await_response_thread.is_alive(): |
| 100 | self.await_response_thread.stop() |
| 101 | self.await_response_thread.join() |
| 102 | |
| 103 | self.engine.shutdown() |
| 104 | self.engine = None |
| 105 | |
| 106 | if self.llm_args is not None: |
| 107 | assert self._executor_config is None, "An empty executor_config is expected in shutdown when LLM arguments are defined." |
| 108 | if (self.llm_args.backend == "pytorch" |
| 109 | and hasattr(self, "checkpoint_loader") |
| 110 | and self.checkpoint_loader is not None): |
| 111 | self.checkpoint_loader.cleanup() |
| 112 | self.checkpoint_loader = None |
| 113 | else: |
| 114 | if hasattr( |
| 115 | self._executor_config, "checkpoint_loader" |
| 116 | ) and self._executor_config.checkpoint_loader is not None: |
| 117 | self._executor_config.checkpoint_loader.cleanup() |
| 118 | self._executor_config.checkpoint_loader = None |
| 119 | |
| 120 | # Check if there are any errors from the threads before shutdown. |
| 121 | self._handle_background_error() |
| 122 | |
| 123 | logger_debug(f"Worker {mpi_rank()} shutdown done.\n", "yellow") |
| 124 | |
| 125 | def block_subordinates(self): |
| 126 | if self.rank != 0: |
no test coverage detected