(self)
| 744 | return True, "" |
| 745 | |
| 746 | def launch_components(self): |
| 747 | if self.cfg.scheduler_config.splitwise_role != "mixed": |
| 748 | self.splitwise_receive_thread = threading.Thread( |
| 749 | target=self.engine.split_connector.start_receiver, args=() |
| 750 | ) |
| 751 | self.splitwise_receive_thread.daemon = True |
| 752 | self.splitwise_receive_thread.start() |
| 753 | |
| 754 | role = self.cfg.scheduler_config.splitwise_role |
| 755 | host_ip = self.cfg.host_ip |
| 756 | request_queues_for_dp_ipc = None |
| 757 | result_queues_for_dp_ipc = None |
| 758 | if self.cfg.scheduler_config.name == "splitwise": |
| 759 | self.engine.scheduler.start(role, host_ip, self.cfg.register_info) |
| 760 | elif self.cfg.scheduler_config.name == "dp": |
| 761 | request_queues_for_dp_ipc = [] |
| 762 | result_queues_for_dp_ipc = [] |
| 763 | for i in range(self.cfg.parallel_config.data_parallel_size): |
| 764 | request_queues_for_dp_ipc.append(multiprocessing.Queue()) |
| 765 | result_queues_for_dp_ipc.append(multiprocessing.Queue()) |
| 766 | self.engine.scheduler.start( |
| 767 | self.cfg.node_rank * self.cfg.worker_num_per_node % self.cfg.worker_num_per_node, |
| 768 | request_queues_for_dp_ipc, |
| 769 | result_queues_for_dp_ipc, |
| 770 | ) |
| 771 | |
| 772 | if not envs.FD_ENABLE_MULTI_API_SERVER: |
| 773 | if self.cfg.parallel_config.data_parallel_size > 1: |
| 774 | self.launched_expert_service_signal.value[0] = 1 |
| 775 | self.dp_processed = [] |
| 776 | self.dp_engine_worker_queue_server = [] |
| 777 | for i in range( |
| 778 | 1, |
| 779 | self.cfg.parallel_config.data_parallel_size // self.cfg.nnode, |
| 780 | ): |
| 781 | if not envs.FD_ENGINE_TASK_QUEUE_WITH_SHM: |
| 782 | address = ( |
| 783 | self.cfg.master_ip, |
| 784 | int(self.cfg.parallel_config.engine_worker_queue_port[i]), |
| 785 | ) |
| 786 | else: |
| 787 | address = f"/dev/shm/fd_task_queue_{self.cfg.parallel_config.engine_worker_queue_port[i]}.sock" |
| 788 | |
| 789 | llm_logger.info(f"dp start queue service {address}") |
| 790 | self.dp_engine_worker_queue_server.append( |
| 791 | EngineWorkerQueue( |
| 792 | address=address, |
| 793 | is_server=True, |
| 794 | num_client=self.cfg.parallel_config.tensor_parallel_size, |
| 795 | local_data_parallel_size=self.cfg.parallel_config.data_parallel_size, |
| 796 | ) |
| 797 | ) |
| 798 | ctx = multiprocessing.get_context("fork") |
| 799 | cfg = copy.deepcopy(self.cfg) |
| 800 | self.dp_processed.append( |
| 801 | ctx.Process( |
| 802 | target=start_data_parallel_service, |
| 803 | args=( |
no test coverage detected