EngineClient is a class that handles the communication between the client and the server.
| 63 | |
| 64 | |
| 65 | class EngineClient: |
| 66 | """ |
| 67 | EngineClient is a class that handles the communication between the client and the server. |
| 68 | """ |
| 69 | |
| 70 | def __init__(self, pid: int | str, port: int | str, fd_config: FDConfig, workers: int = 1, max_logprobs: int = 20): |
| 71 | self.fd_config = fd_config |
| 72 | self.tensor_parallel_size = self.fd_config.parallel_config.tensor_parallel_size |
| 73 | self.enable_mm = self.fd_config.model_config.enable_mm |
| 74 | self.max_logprobs = max_logprobs |
| 75 | input_processor = InputPreprocessor( |
| 76 | self.fd_config.model_config, |
| 77 | self.fd_config.structured_outputs_config.reasoning_parser, |
| 78 | self.fd_config.limit_mm_per_prompt, |
| 79 | self.fd_config.mm_processor_kwargs, |
| 80 | self.fd_config.tool_parser, |
| 81 | self.enable_mm and self.fd_config.cache_config.max_processor_cache > 0, |
| 82 | ) |
| 83 | self.enable_logprob = self.fd_config.model_config.enable_logprob |
| 84 | self.data_processor = input_processor.create_processor() |
| 85 | self.ori_vocab_size = ( |
| 86 | len(self.data_processor.tokenizer.sp_model) |
| 87 | if hasattr(self.data_processor.tokenizer, "sp_model") |
| 88 | else len(self.data_processor.tokenizer.vocab) |
| 89 | ) |
| 90 | self.max_model_len = self.fd_config.model_config.max_model_len |
| 91 | self.enable_prefix_caching = self.fd_config.cache_config.enable_prefix_caching |
| 92 | self.enable_cache_transfer = ( |
| 93 | self.fd_config.cache_config.swap_space or self.fd_config.cache_config.kvcache_storage_backend |
| 94 | ) |
| 95 | self.enable_splitwise = self.fd_config.scheduler_config.splitwise_role != "mixed" |
| 96 | self.max_chips_per_node = 16 if current_platform.is_iluvatar() else 8 |
| 97 | self.num_dp_per_node = self.max_chips_per_node // self.fd_config.parallel_config.tensor_parallel_size |
| 98 | self.data_parallel_rank = ( |
| 99 | self.fd_config.node_rank * self.num_dp_per_node + self.fd_config.parallel_config.local_data_parallel_id |
| 100 | ) |
| 101 | self.data_parallel_info = { |
| 102 | "dp_rank": self.data_parallel_rank, |
| 103 | "local_dp_rank": self.fd_config.parallel_config.local_data_parallel_id, |
| 104 | } |
| 105 | |
| 106 | if self.tensor_parallel_size <= self.max_chips_per_node: |
| 107 | self.is_master = True |
| 108 | else: |
| 109 | self.is_master = False |
| 110 | |
| 111 | if self.fd_config.eplb_config.enable_eplb: |
| 112 | self.init_eplb_signals(ipc_signal_suffix=port) |
| 113 | |
| 114 | array_size = min(self.max_chips_per_node, self.tensor_parallel_size) |
| 115 | self.worker_healthy_live_recorded_time_array = np.zeros(shape=[array_size], dtype=np.int32) |
| 116 | self.worker_healthy_live_signal = IPCSignal( |
| 117 | name="worker_healthy_live_signal", |
| 118 | array=self.worker_healthy_live_recorded_time_array, |
| 119 | dtype=np.int32, |
| 120 | suffix=port, |
| 121 | create=False, |
| 122 | ) |
no outgoing calls