o پi" @sddlZddlZddlZddlZddlZddlmZddlmZm Z m Z m Z m Z m Z ddlZddlmZeeZddZddZ dd ejd ed ed e ejfddZdededed e eeffddZejGdddZdS)N)deque)AnyDequeDictOptionalSequenceTuple)TCPStorecCs ||dksJd||dS)z6Ensure that numerator is divisible by the denominator.rz{} is not divisible by {}N)format numerator denominatorrP/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/distributed/utils.pyensure_divisibilitys rcCst||||S)zXEnsure that numerator is divisible by the denominator and return the division value.)rr rrrdivides rFtensornum_partitionscontiguous_split_chunksreturncCsH|d}t|||}tj|||d}|r"tdd|DS|S)a5Split a tensor along its last dimension. Arguments: tensor: input tensor. num_partitions: number of partitions to split the tensor contiguous_split_chunks: If True, make each chunk contiguous in memory. Returns: A list of Tensors )dimcss|]}|VqdS)N) contiguous).0chunkrrr :sz.split_tensor_along_last_dim..)rrsizetorchsplittuple)rrrlast_dim last_dim_size tensor_listrrrsplit_tensor_along_last_dim#s r#num_hidden_layerspp_rankpp_sizec Cs&tdd}|durcz dd|dD}Wnty*}ztd||d}~wwt||kr>tdt|d|d t||krQtd t|d |d t|d|}|||}||fS||}||} ||| kr|| } ||d | }||d }||fS||}||}||fS) zTry to evenly distribute layers across partitions. If the number of layers is not divisible by the number of partitions, the last N partitions will have one extra layer, where N = remainder. SGLANG_PP_LAYER_PARTITIONNcSsg|]}t|qSr)int)rlayerrrr Jsz"get_pp_indices..,zInvalid partition string: {}zlen(partitions)=z does not match pp_size=.zsum(partitions)=z" does not match num_hidden_layers=r)osgetenvr ValueErrorr lensum) r$r%r&partition_list_str partitionserr start_layer end_layer base_layers remainderpartitions_without_extra_layerrrrget_pp_indices?s:      r:c@sReZdZUdZeed<eed<ejjj ed<dZ eed<e j e dZeeefed<e j e dZeeefed <d Zeed <e j e dZeeefed <e j edZeeeefed <ddZdedefddZddZdedefddZdeededefddZdede efddZ!ddZ"e# d%d ed!ededededdf d"d#Z$d$S)&StatelessProcessGroupzA dataclass to hold a metadata store, and the rank, world_size of the group. Only use it to communicate metadata between processes. For data-plane communication, create NCCL-related objects. rank world_sizestoredata_expiration_seconds)default_factorysend_dst_counterrecv_src_counterrbroadcast_send_counterbroadcast_recv_src_counterentriescCsV|j|jksJddt|jD|_ddt|jD|_ddt|jD|_dS)NcSi|]}|dqSrrrirrr ~z7StatelessProcessGroup.__post_init__..cSrGrHrrIrrrrKrLcSrGrHrrIrrrrKrL)r<r=rangerBrCrE)selfrrr __post_init__|sz#StatelessProcessGroup.__post_init__objdstcCs\|d|d|j|}|j|t||j|d7<|j|t fdS)z%Send an object to a destination rank.send_to//rN) expire_datarBr>setpickledumpsrFappendtime perf_counter)rNrPrQkeyrrrsend_objs zStatelessProcessGroup.send_objcCsP|jr&|jd\}}t||jkr|j||jndS|jsdSdS)zAExpire data that is older than `data_expiration_seconds` seconds.rN)rFrYrZr@r> delete_keypopleft)rNr[ timestamprrrrTs  z!StatelessProcessGroup.expire_datasrcrc Cs<t|jd|jd|j|}|j|d7<|S)z%Receive an object from a source rank.rRrSr)rVloadsr>getr<rC)rNr`rPrrrrecv_objs zStatelessProcessGroup.recv_objcCs|j|kr/|d|d|j}|j|t||jd7_|j|t f|Sd|d|j |}t |j |}|j |d7<|S)zBroadcast an object from a source rank to all other ranks. It does not clean up after all ranks have received the object. Use it for limited times, e.g., for initialization. zbroadcast_from/rSr)r<rTrDr>rUrVrWrFrXrYrZrErarb)rNrPr`r[rcrrr broadcast_objs z#StatelessProcessGroup.broadcast_objcCsVg}t|jD]!}||jkr|||j||jdq|jd|d}||q|S)z$All gather an object from all ranks.r`N)rMr=r<rXrd)rNrP gathered_objsrJrcrrrall_gather_objs   z$StatelessProcessGroup.all_gather_objcCs>t|jD]}||jkr|jd|jdq|jd|dqdS)z#A barrier to synchronize all ranks.Nre)rMr=r<rd)rNrJrrrbarriers  zStatelessProcessGroup.barrierhostportcCs$t||||dkd}t||||dS)aA replacement for `torch.distributed.init_process_group` that does not pollute the global state. If we have process A and process B called `torch.distributed.init_process_group` to form a group, and then we want to form another group with process A, B, C, D, it is not possible in PyTorch, because process A and process B have already formed a group, and process C and process D cannot join that group. This function is a workaround for this issue. `torch.distributed.init_process_group` is a global call, while this function is a stateless call. It will return a `StatelessProcessGroup` object that can be used for exchanging metadata. With this function, process A and process B can call `StatelessProcessGroup.create` to form a group, and then process A, B, C, and D can call `StatelessProcessGroup.create` to form another group. r) host_namerjr= is_master)r<r=r>r@)r r;)rirjr<r=r@r>rrrcreateszStatelessProcessGroup.createN)r?)%__name__ __module__ __qualname____doc__r(__annotations__r_C_distributed_c10dStorer@ dataclassesfielddictrBrrCrDrErrFrrstrfloatrOrr\rTrcrrdlistrgrh staticmethodrmrrrrr;fsB      r;)F)rvloggingr-rVrY collectionsrtypingrrrrrrrtorch.distributedr getLoggerrnloggerrrTensorr(boolr#r: dataclassr;rrrrsB        '