Initializes the LLMEngine with the provided configuration. Args: cfg (Config): Config object containing all the configuration parameters.
(self, cfg, local_data_parallel_id, start_queue=True)
| 40 | """ |
| 41 | |
| 42 | def __init__(self, cfg, local_data_parallel_id, start_queue=True): |
| 43 | """ |
| 44 | Initializes the LLMEngine with the provided configuration. |
| 45 | |
| 46 | Args: |
| 47 | cfg (Config): Config object containing all the configuration parameters. |
| 48 | """ |
| 49 | |
| 50 | self.cfg = cfg |
| 51 | |
| 52 | if self.cfg.parallel_config.data_parallel_size > 1: |
| 53 | self.llm_logger = get_logger("fastdeploy", f"fastdeploy_dprank{local_data_parallel_id}.log") |
| 54 | else: |
| 55 | self.llm_logger = llm_logger |
| 56 | |
| 57 | if envs.FD_ENABLE_INTERNAL_ADAPTER: |
| 58 | assert ( |
| 59 | envs.FD_ZMQ_RECV_REQUEST_SERVER_PORTS is not None or envs.FD_ZMQ_RECV_REQUEST_SERVER_PORT is not None |
| 60 | ), "Please set FD_ZMQ_RECV_REQUEST_SERVER_PORTS or FD_ZMQ_RECV_REQUEST_SERVER_PORT when enabling internal adapter." |
| 61 | assert ( |
| 62 | envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORTS is not None or envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORT is not None |
| 63 | ), "Please set FD_ZMQ_SEND_RESPONSE_SERVER_PORTS or FD_ZMQ_SEND_RESPONSE_SERVER_PORT when enabling internal adapter." |
| 64 | if envs.FD_ZMQ_RECV_REQUEST_SERVER_PORTS is not None: |
| 65 | envs.FD_ZMQ_RECV_REQUEST_SERVER_PORT = envs.FD_ZMQ_RECV_REQUEST_SERVER_PORTS.split(",")[ |
| 66 | local_data_parallel_id |
| 67 | ] |
| 68 | if envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORTS is not None: |
| 69 | envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORT = envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORTS.split(",")[ |
| 70 | local_data_parallel_id |
| 71 | ] |
| 72 | self.llm_logger.info( |
| 73 | f"local_data_parallel_id: {local_data_parallel_id},envs.FD_ZMQ_RECV_REQUEST_SERVER_PORT:{envs.FD_ZMQ_RECV_REQUEST_SERVER_PORT},envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORT:{envs.FD_ZMQ_SEND_RESPONSE_SERVER_PORT}" |
| 74 | ) |
| 75 | |
| 76 | if self.cfg.cache_config.num_gpu_blocks_override is None: |
| 77 | self.do_profile = True |
| 78 | else: |
| 79 | self.do_profile = False |
| 80 | |
| 81 | # Update config for the current dp process |
| 82 | if not envs.FD_ENABLE_MULTI_API_SERVER: |
| 83 | self.cfg.parallel_config.local_data_parallel_id = local_data_parallel_id |
| 84 | self.cfg.postprocess_devices_and_ports() |
| 85 | self.llm_logger.info( |
| 86 | f"Update config for the current dp process: " |
| 87 | f"local_engine_worker_queue_port: {self.cfg.parallel_config.local_engine_worker_queue_port} " |
| 88 | f"local_cache_queue_port: {self.cfg.cache_config.local_cache_queue_port} " |
| 89 | f"local_pd_comm_port: {self.cfg.cache_config.local_pd_comm_port} " |
| 90 | f"local_rdma_comm_ports: {self.cfg.cache_config.local_rdma_comm_ports} " |
| 91 | ) |
| 92 | |
| 93 | self.engine = EngineService(self.cfg, start_queue) |
| 94 | if self.cfg.scheduler_config.name == "splitwise": |
| 95 | self.engine.scheduler.reset_nodeid(f"{self.engine.scheduler.infer.nodeid}_{local_data_parallel_id!s}") |
| 96 | |
| 97 | self._finalizer = weakref.finalize(self, self._exit_sub_services) |
| 98 | |
| 99 | def start( |
nothing calls this directly
no test coverage detected