Paddle Distributed wrapper for fastdeploy.worker.Worker, for handling single-node multi-GPU tensor parallel. The wrapper internally executes an event loop that continuously executes requests in the task queue. Control flow is transmitted by IPC.
| 140 | |
| 141 | |
| 142 | class PaddleDisWorkerProc: |
| 143 | """ |
| 144 | Paddle Distributed wrapper for fastdeploy.worker.Worker, |
| 145 | for handling single-node multi-GPU tensor parallel. |
| 146 | The wrapper internally executes an event loop that continuously executes requests |
| 147 | in the task queue. Control flow is transmitted by IPC. |
| 148 | """ |
| 149 | |
| 150 | def __init__(self, fd_config: FDConfig, ranks: int = 1, local_rank: int = 0) -> None: |
| 151 | """ |
| 152 | Initialize a distributed worker and task queue for single-node multi-GPU setup. |
| 153 | Args: |
| 154 | fd_config (FDConfig): Arguments related to inference, containing |
| 155 | attributes such as weight_dtype, act_dtype, mp_size, hidden_size, head_dim, |
| 156 | num_attention_heads, and ffn_hidden_size. |
| 157 | """ |
| 158 | self.ranks = ranks |
| 159 | self.local_rank = local_rank |
| 160 | self.fd_config = fd_config |
| 161 | self.parallel_config = fd_config.parallel_config |
| 162 | self.cache_config = fd_config.cache_config |
| 163 | self.scheduler_config = fd_config.scheduler_config |
| 164 | self.eplb_config = fd_config.eplb_config |
| 165 | |
| 166 | # TODO(gongshaotian): Use worker factory to get worker |
| 167 | self.worker = get_worker(fd_config=fd_config, local_rank=self.local_rank, rank=self.ranks) |
| 168 | |
| 169 | self.max_chips_per_node = 16 if current_platform.is_iluvatar() else 8 |
| 170 | self.speculative_decoding = fd_config.speculative_config.method is not None |
| 171 | self.enable_overlap_schedule = self.scheduler_config.enable_overlap_schedule and ( |
| 172 | not self.speculative_decoding |
| 173 | ) |
| 174 | self.cached_control_reqs = [] |
| 175 | |
| 176 | def init_control(self): |
| 177 | engine_worker_queue_port = self.parallel_config.local_engine_worker_queue_port |
| 178 | queue_name = f"ctrl_w2e_rank{self.local_rank}_{engine_worker_queue_port}" |
| 179 | logger.info(f"Init Control Output Queue: {queue_name}(producer)") |
| 180 | self._ctrl_output = FMQ().queue(queue_name, "producer") |
| 181 | |
| 182 | def init_health_status(self) -> None: |
| 183 | """ |
| 184 | Initialize the health status of the worker. |
| 185 | Worker Status: |
| 186 | worker_ready_signal: |
| 187 | worker_healthy_live_signal: |
| 188 | exist_task_signal: |
| 189 | exist_swapped_task_signal: |
| 190 | model_weights_status: |
| 191 | """ |
| 192 | if self.parallel_config.data_parallel_size > 1 and not envs.FD_ENABLE_MULTI_API_SERVER: |
| 193 | launched_expert_service_signal_data = np.zeros( |
| 194 | shape=[self.parallel_config.data_parallel_size // self.fd_config.nnode], dtype=np.int32 |
| 195 | ) |
| 196 | self.launched_expert_service_signal = IPCSignal( |
| 197 | name="launched_expert_service_signal", |
| 198 | array=launched_expert_service_signal_data, |
| 199 | dtype=np.int32, |