Reclaim one managed instance and return its GPU back to idle pool.
(self, instance_type: str, instance_address: str | None = None)
| 1508 | return instance_address |
| 1509 | |
| 1510 | def reclaim_instance(self, instance_type: str, instance_address: str | None = None) -> str: |
| 1511 | """Reclaim one managed instance and return its GPU back to idle pool.""" |
| 1512 | if instance_type not in {"encoder", "transformer", "decoder"}: |
| 1513 | raise ValueError("instance_type must be one of: encoder, transformer, decoder") |
| 1514 | |
| 1515 | with self._instance_lock: |
| 1516 | target_address = instance_address |
| 1517 | if target_address is None: |
| 1518 | candidates = [addr for addr, meta in self._managed_instances.items() if meta.get("instance_type") == instance_type] |
| 1519 | if not candidates: |
| 1520 | raise RuntimeError(f"no managed {instance_type} instance to reclaim") |
| 1521 | target_address = candidates[-1] |
| 1522 | |
| 1523 | meta = self._managed_instances.get(target_address) |
| 1524 | if meta is None: |
| 1525 | if (instance_type, target_address) in self.started_instances: |
| 1526 | self.started_instances.remove((instance_type, target_address)) |
| 1527 | self.logger.warning( |
| 1528 | "Skip reclaim for already-removed %s instance address=%s", |
| 1529 | instance_type, |
| 1530 | target_address, |
| 1531 | ) |
| 1532 | return target_address |
| 1533 | if meta.get("instance_type") != instance_type: |
| 1534 | raise RuntimeError(f"instance type mismatch for {target_address}: expected={instance_type} got={meta.get('instance_type')}") |
| 1535 | |
| 1536 | process = meta.get("process") |
| 1537 | process_meta = meta.get("process_meta") if isinstance(meta.get("process_meta"), dict) else None |
| 1538 | gpu_id = int(meta.get("gpu_id")) |
| 1539 | sidecar_meta = meta.get("sidecar") if isinstance(meta.get("sidecar"), dict) else None |
| 1540 | host = str(meta.get("host", self._bootstrap_addr)) |
| 1541 | launch_mode = str(meta.get("launch_mode", "local")) |
| 1542 | static_slot = meta.get("static_slot") if isinstance(meta.get("static_slot"), dict) else None |
| 1543 | slot_id_raw = meta.get("slot_id") |
| 1544 | slot_id = int(slot_id_raw) if slot_id_raw is not None else None |
| 1545 | |
| 1546 | self.remove_instance(instance_type, target_address) |
| 1547 | monitor_node = self._monitor_node_from_instance_address(target_address) |
| 1548 | |
| 1549 | if launch_mode == "remote": |
| 1550 | if static_slot is None: |
| 1551 | raise RuntimeError(f"remote instance metadata missing static slot for {target_address}") |
| 1552 | |
| 1553 | remote_service_pid = None |
| 1554 | if process_meta is not None and isinstance(process_meta.get("pid"), int): |
| 1555 | remote_service_pid = int(process_meta["pid"]) |
| 1556 | if remote_service_pid is not None and remote_service_pid > 0: |
| 1557 | self._stop_remote_pid(static_slot, remote_service_pid, self._graceful_reclaim_timeout_seconds) |
| 1558 | |
| 1559 | if sidecar_meta is not None and isinstance(sidecar_meta.get("pid"), int): |
| 1560 | self._stop_remote_pid(static_slot, int(sidecar_meta["pid"]), self._force_kill_wait_seconds) |
| 1561 | else: |
| 1562 | if process is not None and process.poll() is None: |
| 1563 | try: |
| 1564 | os.killpg(process.pid, signal.SIGTERM) |
| 1565 | except Exception: |
| 1566 | process.terminate() |
| 1567 | try: |
no test coverage detected