(self)
| 288 | request_id=request_id) |
| 289 | |
| 290 | def shutdown(self): |
| 291 | if hasattr(self, '_shutdown_event') and self._shutdown_event.is_set(): |
| 292 | return |
| 293 | if hasattr(self, '_shutdown_event'): |
| 294 | self._shutdown_event.set() |
| 295 | |
| 296 | logger_debug(f"Shutting down RayExecutor", color="yellow") |
| 297 | |
| 298 | if hasattr(self, 'main_loop') and self.main_loop and hasattr( |
| 299 | self, 'main_loop_task_obj') and self.main_loop_task_obj: |
| 300 | logger_debug("Cancelling main loop task.", color="yellow") |
| 301 | try: |
| 302 | self.main_loop.call_soon_threadsafe( |
| 303 | self.main_loop_task_obj.cancel) |
| 304 | except Exception as e: |
| 305 | logger_debug(f"Error cancelling main loop task: {e}", |
| 306 | color="yellow") |
| 307 | |
| 308 | if hasattr(self, 'main_loop_thread'): |
| 309 | self.main_loop_thread.join() |
| 310 | |
| 311 | # Then, shutdown the workers |
| 312 | if hasattr(self, 'workers') and self.workers is not None: |
| 313 | try: |
| 314 | shutdown_refs = [ |
| 315 | worker.shutdown.remote() for worker in self.workers |
| 316 | ] |
| 317 | # Add timeout to prevent indefinite hanging |
| 318 | ray.get(shutdown_refs, timeout=30.0) |
| 319 | except ray.exceptions.GetTimeoutError: |
| 320 | logger.warning( |
| 321 | "Timeout waiting for workers to shutdown after 30 seconds") |
| 322 | except Exception as e: |
| 323 | logger.warning(f"Error shutting down: {e}") |
| 324 | |
| 325 | if hasattr(self, 'rpc_client') and self.rpc_client is not None: |
| 326 | try: |
| 327 | self.rpc_client.close() |
| 328 | except Exception as e: |
| 329 | logger_debug(f"Suppressed error during RPC client close: {e}") |
| 330 | |
| 331 | self.workers = None |
| 332 | if hasattr(self, |
| 333 | "placement_group") and self.placement_group is not None: |
| 334 | # Only remove placement group if Ray is still initialized |
| 335 | # to avoid triggering auto_init_ray() during program exit |
| 336 | if ray.is_initialized(): |
| 337 | ray.util.remove_placement_group(self.placement_group) |
| 338 | self.placement_group = None |
| 339 | self.bundle_indices = None |
| 340 | |
| 341 | if self.has_start_local_cluser and ray.is_initialized(): |
| 342 | logger.debug("Shutting down Ray cluster") |
| 343 | ray.shutdown() |
| 344 | |
| 345 | def _get_worker_ready_futures(self): |
| 346 | return [worker.__ray_ready__.remote() for worker in self.workers] |
no test coverage detected