o ۾i0@sdZddlmZmZddlmZddlmZmZddl Z er$ddl m Z ddl m Z ddlmZdd lmZmZmZdd lmZmZmZeGd d d eZeGd ddeZGdddeeefZdS)z"NCCL-based weight transfer engine.)CallableIterator) dataclass) TYPE_CHECKINGAnyNPyNcclCommunicator)ParallelConfig)WeightTransferConfig)WeightTransferEngineWeightTransferInitInfoWeightTransferUpdateInfo) DEFAULT_PACKED_BUFFER_SIZE_BYTESDEFAULT_PACKED_NUM_BUFFERSpacked_broadcast_consumerc@s2eZdZUdZeed<eed<eed<eed<dS)NCCLWeightTransferInitInfoz5Initialization info for NCCL weight transfer backend.master_address master_port rank_offset world_sizeN)__name__ __module__ __qualname____doc__str__annotations__intrr`/home/ubuntu/.local/lib/python3.10/site-packages/vllm/distributed/weight_transfer/nccl_engine.pyrs  rc@sleZdZUdZeeed<eeed<eeeed<dZe ed< e Z eed< e Z eed< d d Zd S) NCCLWeightTransferUpdateInfoz-Update info for NCCL weight transfer backend.names dtype_namesshapesFpackedpacked_buffer_size_bytespacked_num_bufferscCsjt|j}t|j|krtdt|jdt|jt|j|kr3tdt|jdt|jdS)z-Validate that all lists have the same length.z9`dtype_names` should be of the same size as `names`: got z and z4`shapes` should be of the same size as `names`: got N)lenr r! ValueErrorr")self num_paramsrrr __post_init__8s$ z*NCCLWeightTransferUpdateInfo.__post_init__N)rrrrlistrrrr#boolrr$rr%r*rrrrr&s       rcseZdZdZeZeZdede ddffdd Z deddfd d Z d ed e e eeejfgdfddfd dZd"ddZeddddeefdeeeejfdedede eeejfgejfdBdedejjdBdededdfddZedeeBddfddZed d!Z Z!S)#NCCLWeightTransferEnginez Weight transfer engine using NCCL for communication between trainer and workers. This implementation uses NCCL broadcast operations to transfer weights from the trainer (rank 0) to all inference workers in a process group. configparallel_configreturnNcst||d|_dS)z Initialize the NCCL weight transfer engine. Args: config: The configuration for the weight transfer engine parallel_config: The configuration for the parallel setup N)super__init__model_update_group)r(r.r/ __class__rrr2Us z!NCCLWeightTransferEngine.__init__ init_infocCsR|jj}|jj}|jj}|||}||j}t|j|j||jt j |_ dS)z Initialize NCCL process group with the trainer. Args: init_info: NCCL initialization info containing master address, port, rank offset, and world size N) r/data_parallel_rankrrankrr-_stateless_init_process_grouprrtorchcudacurrent_devicer3)r(r6dp_rankworld_size_per_dprank_within_dp worker_rankr8rrrinit_transfer_enginebs   z-NCCLWeightTransferEngine.init_transfer_engine update_info load_weightsc s|jdur tdjr"fdd}t||jd|jjddStjjj D]&\}}}t t |}t j ||dd}|jj |dt jd |||fg~q+dS) aF Receive weights from trainer via NCCL broadcast and load them incrementally. If update_info.packed is True, uses packed tensor broadcasting for efficient transfer of multiple weights in batches. Otherwise, uses simple one-by-one broadcasting. Args: update_info: NCCL update info containing parameter names, dtypes, shapes, and packed flag load_weights: Callable that loads weights into the model. Called incrementally for each batch of weights to avoid OOM. NzHNCCL weight transfer not initialized. Call init_transfer_engine() first.c3s<tjjjD]\}}}tt|}|||ffVq dSN)zipr r!r"getattrr:)name dtype_nameshapedtyperBrrstate_dict_info_iterators  zJNCCLWeightTransferEngine.receive_weights..state_dict_info_iteratorr)iteratorgroupsrcpost_unpack_funcbuffer_size_bytes num_buffersr;)rJdevicerOstream)r3 RuntimeErrorr#rr$r%rEr r!r"rFr:empty broadcastr;current_stream) r(rBrCrLrGrHrIrJweightrrKrreceive_weightss2      z(NCCLWeightTransferEngine.receive_weightscCs|jdur d|_dSdSrD)r3)r(rrrshutdowns  z!NCCLWeightTransferEngine.shutdownrFrMrNrOpost_iter_funcr#rUr$r%c Csh|durdd}|rddlm}|||||||ddS|D]} || } |j| ||p.tjdqdS)aBroadcast weights from trainer to vLLM workers. Args: iterator: Iterator of model parameters. Returns (name, tensor) tuples group: Process group (PyNcclCommunicator) src: Source rank (default 0, trainer is typically rank 0) post_iter_func: Optional function to apply to each (name, tensor) pair before broadcasting. If None, extracts just the tensor. packed: Whether to use packed tensor broadcasting for efficiency. When True, multiple tensors are batched together before broadcasting to reduce NCCL communication overhead. stream: CUDA stream to use for broadcasting if packed is False. If packed is True, new streams will be created for each buffer. packed_buffer_size_bytes: Size in bytes for each packed tensor buffer. Must match the value used in NCCLWeightTransferUpdateInfo. packed_num_buffers: Number of buffers for double/triple buffering. Must match the value used in NCCLWeightTransferUpdateInfo. Example: >>> from vllm.distributed.weight_transfer.nccl_engine import ( ... NCCLWeightTransferEngine, ... ) >>> param_iter = ((n, p) for n, p in model.named_parameters()) >>> NCCLWeightTransferEngine.trainer_send_weights( ... param_iter, group, packed=True ... ) NcSs|dS)Nr)xrrrsz?NCCLWeightTransferEngine.trainer_send_weights..r)packed_broadcast_producer)rMrNrOr]rQrRrT).vllm.distributed.weight_transfer.packed_tensorrarXr:r;rY) rMrNrOr]r#rUr$r%raitemtensorrrrtrainer_send_weightss$'   z-NCCLWeightTransferEngine.trainer_send_weightsrcCsNt|tr|d}|d}|d}n |j}|j}|j}t||d|tj S)a Initialize NCCL process group for trainer-side weight transfer. The trainer is always rank 0 in the process group. Uses the current CUDA device (torch.cuda.current_device()). Args: init_info: Either an NCCLWeightTransferInitInfo object or a dict with keys: - master_address: str - master_port: int - world_size: int Returns: PyNcclCommunicator for weight transfer. Example: >>> from vllm.distributed.weight_transfer.nccl_engine import ( ... NCCLWeightTransferEngine, ... ) >>> group = NCCLWeightTransferEngine.trainer_init( ... dict( ... master_address=master_address, ... master_port=master_port, ... world_size=world_size, ... ), ... ) rrrr) isinstancedictrrrr-r9r:r;r<)r6rrrrrr trainer_inits  z%NCCLWeightTransferEngine.trainer_initc Cs:ddlm}ddlm}|j||||d}|||d}|S)aV vLLM provides `StatelessProcessGroup` to create a process group without considering the global process group in torch.distributed. It is recommended to create `StatelessProcessGroup`, and then initialize the data-plane communication (NCCL) between external (train processes) and vLLM workers. rr)StatelessProcessGroup)hostportr8r)rS),vllm.distributed.device_communicators.pyncclrvllm.distributed.utilsricreate) rrr8rrSrripgpyncclrrrr9)s  z6NCCLWeightTransferEngine._stateless_init_process_group)r0N)"rrrrr init_info_clsrupdate_info_clsr r r2rArr+tuplerr:Tensorr[r\ staticmethodrrrrrr,r;Streamrergrhr9 __classcell__rrr4rr-Gsp  6   @ -r-)rcollections.abcrr dataclassesrtypingrrr:rlrvllm.config.parallelr vllm.config.weight_transferr %vllm.distributed.weight_transfer.baser r r rbrrrrrr-rrrrs$