Create one service instance on an idle GPU and add it to scheduling pool.
(self, instance_type: str)
| 1319 | self._free_gpus = set(range(total_ranks)) |
| 1320 | |
| 1321 | def create_instance(self, instance_type: str) -> str: |
| 1322 | """Create one service instance on an idle GPU and add it to scheduling pool.""" |
| 1323 | if instance_type not in {"encoder", "transformer", "decoder"}: |
| 1324 | raise ValueError("instance_type must be one of: encoder, transformer, decoder") |
| 1325 | if self._runtime_config is None: |
| 1326 | raise RuntimeError("controller runtime config is not initialized") |
| 1327 | |
| 1328 | with self._instance_lock: |
| 1329 | use_static_slots = bool(self._static_instance_slots) |
| 1330 | selected_slot: dict[str, Any] | None = None |
| 1331 | |
| 1332 | if use_static_slots: |
| 1333 | if not self._free_slot_ids: |
| 1334 | raise RuntimeError("no idle static slot available") |
| 1335 | |
| 1336 | now = time.time() |
| 1337 | for slot_id in sorted(self._free_slot_ids): |
| 1338 | slot = self._static_instance_slots[slot_id] |
| 1339 | if slot.get("instance_type") != instance_type: |
| 1340 | continue |
| 1341 | if now < self._slot_reuse_block_until.get(slot_id, 0.0): |
| 1342 | continue |
| 1343 | |
| 1344 | host = str(slot["host"]) |
| 1345 | engine_rank = int(slot["engine_rank"]) |
| 1346 | monitor_port = MONITOR_POLLING_PORT + engine_rank |
| 1347 | if self._is_tcp_port_open(host, monitor_port): |
| 1348 | self.logger.warning( |
| 1349 | "Skip static slot=%s host=%s rank=%s for %s creation because monitor port %s is still in use", |
| 1350 | slot_id, |
| 1351 | host, |
| 1352 | engine_rank, |
| 1353 | instance_type, |
| 1354 | monitor_port, |
| 1355 | ) |
| 1356 | continue |
| 1357 | |
| 1358 | selected_slot = slot |
| 1359 | break |
| 1360 | |
| 1361 | if selected_slot is None: |
| 1362 | raise RuntimeError(f"no idle static slot available for {instance_type}: all candidates cooling down or port is in use") |
| 1363 | |
| 1364 | engine_rank = int(selected_slot["engine_rank"]) |
| 1365 | host = str(selected_slot["host"]) |
| 1366 | cuda_device = str(selected_slot["cuda_device"]) |
| 1367 | else: |
| 1368 | if not self._free_gpus: |
| 1369 | raise RuntimeError("no idle GPU available") |
| 1370 | |
| 1371 | now = time.time() |
| 1372 | engine_rank: int | None = None |
| 1373 | host = self._bootstrap_addr |
| 1374 | for candidate_gpu in sorted(self._free_gpus): |
| 1375 | if now < self._gpu_reuse_block_until.get(candidate_gpu, 0.0): |
| 1376 | continue |
| 1377 | |
| 1378 | monitor_port = MONITOR_POLLING_PORT + candidate_gpu |
no test coverage detected