Internal method to trigger server shutdown. Args: is_remote_call: Whether the shutdown is called by a remote call. This should be True when client.server_shutdown() is called.
(self, is_remote_call: bool = False)
| 115 | logger.info(f"RPCServer is bound to {self._address}") |
| 116 | |
| 117 | def shutdown(self, is_remote_call: bool = False) -> None: |
| 118 | """Internal method to trigger server shutdown. |
| 119 | |
| 120 | Args: |
| 121 | is_remote_call: Whether the shutdown is called by a remote call. |
| 122 | This should be True when client.server_shutdown() is called. |
| 123 | """ |
| 124 | # NOTE: shutdown is also a remote method, so it could be executed by |
| 125 | # a thread in a worker executor thread |
| 126 | |
| 127 | if self._stop_event.is_set(): |
| 128 | return |
| 129 | |
| 130 | logger_debug( |
| 131 | "[server] RPCServer is shutting down. Terminating server immediately..." |
| 132 | ) |
| 133 | |
| 134 | # Set the stop event to True, this will trigger immediate shutdown |
| 135 | self._stop_event.set() |
| 136 | |
| 137 | # Log pending requests that will be cancelled |
| 138 | logger_debug( |
| 139 | f"[server] RPCServer is shutting down: {self._num_pending_requests} pending requests will be cancelled" |
| 140 | ) |
| 141 | |
| 142 | # Signal asyncio shutdown event if available |
| 143 | if self._shutdown_event and self._loop: |
| 144 | self._loop.call_soon_threadsafe(self._shutdown_event.set) |
| 145 | |
| 146 | if not is_remote_call: |
| 147 | # Block the thread until shutdown is finished |
| 148 | |
| 149 | # 1. Cancel the main task gracefully which will trigger proper cleanup |
| 150 | if self._main_task and not self._main_task.done(): |
| 151 | self._loop.call_soon_threadsafe(self._main_task.cancel) |
| 152 | |
| 153 | # 2. Wait for the server thread to exit (this will wait for proper cleanup) |
| 154 | if self._server_thread and self._server_thread.is_alive(): |
| 155 | logger_debug( |
| 156 | "[server] RPCServer is waiting for server thread to exit") |
| 157 | self._server_thread.join() |
| 158 | self._server_thread = None |
| 159 | logger_debug("[server] RPCServer thread joined") |
| 160 | |
| 161 | # 3. Shutdown the executor immediately without waiting for tasks |
| 162 | if self._executor: |
| 163 | self._executor.shutdown(wait=False) |
| 164 | self._executor = None |
| 165 | |
| 166 | # 4. Close the client socket |
| 167 | if self._client_socket: |
| 168 | self._client_socket.close() |
| 169 | else: |
| 170 | # if the shutdown is called by a remote call, this method itself will |
| 171 | # be executed in a executor thread, so we cannot join the server thread |
| 172 | logger_debug( |
| 173 | f"[server] RPC Server shutdown initiated: {self._num_pending_requests} pending requests will be cancelled" |
| 174 | ) |