(
self,
args,
)
| 605 | """Configuration for the distributed execution.""" |
| 606 | |
| 607 | def __init__( |
| 608 | self, |
| 609 | args, |
| 610 | ): |
| 611 | self.sequence_parallel = False # Whether to enable sequence parallelism. |
| 612 | self.use_ep = False # Whether to enable Expert Parallelism |
| 613 | self.msg_queue_id = 1 # message queue id |
| 614 | |
| 615 | self.tensor_parallel_rank = 0 # TP rank ID |
| 616 | self.tensor_parallel_size = 1 # TP degree |
| 617 | self.expert_parallel_rank = 0 # EP rank ID |
| 618 | self.expert_parallel_size = 1 # EP degree |
| 619 | self.data_parallel_rank = 0 # DP rank ID |
| 620 | self.data_parallel_size = 1 # DP degree |
| 621 | self.enable_expert_parallel = False |
| 622 | self.enable_chunked_moe = False |
| 623 | self.chunked_moe_size = 256 |
| 624 | |
| 625 | self.local_data_parallel_id = 0 |
| 626 | # Engine worker queue port |
| 627 | self.engine_worker_queue_port: Union[int, str, list] = None |
| 628 | self.local_engine_worker_queue_port: Optional[int] = None |
| 629 | # cuda visible devices |
| 630 | self.device_ids: str = "0" |
| 631 | # First token id |
| 632 | self.first_token_id: int = 1 |
| 633 | # Process ID of engine |
| 634 | self.engine_pid: Optional[int] = None |
| 635 | # Do profile or not |
| 636 | self.do_profile: bool = False |
| 637 | # Use internode_ll_two_stage or not |
| 638 | self.use_internode_ll_two_stage: bool = False |
| 639 | # disable sequence parallel moe |
| 640 | self.disable_sequence_parallel_moe: bool = False |
| 641 | # shutdown comm group if worker idle |
| 642 | self.shutdown_comm_group_if_worker_idle: bool = None |
| 643 | |
| 644 | self.pod_ip: str = None |
| 645 | # enable the custom all-reduce kernel and fall back to NCCL(dist.all_reduce). |
| 646 | self.disable_custom_all_reduce: bool = False |
| 647 | for key, value in args.items(): |
| 648 | if hasattr(self, key): |
| 649 | setattr(self, key, value) |
| 650 | |
| 651 | self.engine_worker_queue_port = parse_ports(self.engine_worker_queue_port) |
| 652 | |
| 653 | # currently, the expert parallel size is equal data parallel size |
| 654 | if self.enable_expert_parallel: |
| 655 | self.expert_parallel_size = self.data_parallel_size * self.tensor_parallel_size |
| 656 | else: |
| 657 | self.expert_parallel_size = 1 |
| 658 | self.use_ep = self.expert_parallel_size > 1 |
| 659 | |
| 660 | if self.shutdown_comm_group_if_worker_idle is None: |
| 661 | self.shutdown_comm_group_if_worker_idle = not self.use_ep |
| 662 | |
| 663 | if self.shutdown_comm_group_if_worker_idle and envs.FD_ENABLE_V1_UPDATE_WEIGHTS: |
| 664 | raise RuntimeError("shutdown_comm_group_if_worker_idle cannot be True when FD_ENABLE_V1_UPDATE_WEIGHTS=1") |
nothing calls this directly
no test coverage detected