| 2427 | return True, "" |
| 2428 | |
| 2429 | def launch_components(self): |
| 2430 | if self.cfg.scheduler_config.splitwise_role != "mixed": |
| 2431 | # 单机逻辑 |
| 2432 | self.splitwise_receive_thread = threading.Thread(target=self.split_connector.start_receiver, args=()) |
| 2433 | self.splitwise_receive_thread.daemon = True |
| 2434 | self.splitwise_receive_thread.start() |
| 2435 | |
| 2436 | role = self.cfg.scheduler_config.splitwise_role |
| 2437 | host_ip = self.cfg.host_ip |
| 2438 | request_queues_for_dp_ipc = None |
| 2439 | result_queue_for_dp_ipc = None |
| 2440 | if self.cfg.scheduler_config.name == "splitwise": |
| 2441 | self.scheduler.start(role, host_ip, self.cfg.register_info) |
| 2442 | elif self.cfg.scheduler_config.name == "dp": |
| 2443 | request_queues_for_dp_ipc = [] |
| 2444 | result_queue_for_dp_ipc = multiprocessing.Queue() |
| 2445 | for i in range(self.cfg.parallel_config.data_parallel_size): |
| 2446 | request_queues_for_dp_ipc.append(multiprocessing.Queue()) |
| 2447 | self.scheduler.start( |
| 2448 | self.cfg.node_rank * self.cfg.worker_num_per_node % self.cfg.worker_num_per_node, |
| 2449 | request_queues_for_dp_ipc, |
| 2450 | result_queue_for_dp_ipc, |
| 2451 | ) |
| 2452 | |
| 2453 | if not envs.FD_ENABLE_MULTI_API_SERVER: |
| 2454 | if self.cfg.parallel_config.enable_expert_parallel and self.cfg.parallel_config.data_parallel_size > 1: |
| 2455 | self.launched_expert_service_signal.value[0] = 1 |
| 2456 | self.dp_processed = [] |
| 2457 | self.dp_engine_worker_queue_server = [] |
| 2458 | for i in range( |
| 2459 | 1, |
| 2460 | self.cfg.parallel_config.data_parallel_size // self.cfg.nnode, |
| 2461 | ): |
| 2462 | if not envs.FD_ENGINE_TASK_QUEUE_WITH_SHM: |
| 2463 | address = ( |
| 2464 | self.cfg.master_ip, |
| 2465 | int(self.cfg.parallel_config.engine_worker_queue_port[i]), |
| 2466 | ) |
| 2467 | else: |
| 2468 | address = f"/dev/shm/fd_task_queue_{self.cfg.parallel_config.engine_worker_queue_port[i]}.sock" |
| 2469 | |
| 2470 | self.llm_logger.info(f"dp start queue service {address}") |
| 2471 | self.dp_engine_worker_queue_server.append( |
| 2472 | EngineWorkerQueue( |
| 2473 | address=address, |
| 2474 | is_server=True, |
| 2475 | num_client=self.cfg.parallel_config.tensor_parallel_size, |
| 2476 | local_data_parallel_size=self.cfg.parallel_config.data_parallel_size, |
| 2477 | ) |
| 2478 | ) |
| 2479 | from fastdeploy.engine.expert_service import ( |
| 2480 | start_data_parallel_service, |
| 2481 | ) |
| 2482 | |
| 2483 | self.dp_processed.append( |
| 2484 | multiprocessing.Process( |
| 2485 | target=start_data_parallel_service, |
| 2486 | args=( |