o پi @sddlZddlZddlZddlmZddlmZddlmZddl Z ddl m Z ddl m Z e eZddd Zd ed edefd d Z dde jdededee jfddZejGdddZdS)N)deque)Sequence)Any)TCPStore) init_loggerreturncCs ||dksJd||dS)z6Ensure that numerator is divisible by the denominator.rz{} is not divisible by {}N)format numerator denominatorr c/home/ubuntu/.local/lib/python3.10/site-packages/sglang/multimodal_gen/runtime/distributed/utils.pyensure_divisibilitys rr r cCst||||S)zXEnsure that numerator is divisible by the denominator and return the division value.)rr r r r divide s rFtensornum_partitionscontiguous_split_chunkscCsL|d}t|||}tj|||d}|r"tdd|DSt|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).0chunkr r r >sz.split_tensor_along_last_dim..)rrsizetorchsplittuple)rrrlast_dim last_dim_size tensor_listr r r split_tensor_along_last_dim's r c@sTeZdZUdZeed<eed<ejjj ed<dZ eed<e j e dZe eefed<e j e dZe eefed <d Zeed <e j e dZe eefed <e j edZeeeefed <ddZdedefddZd%ddZdedefddZdedBdedefddZdedeefddZdd Ze d&d!ed"ededededdf d#d$Z!dS)'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|]}|dqSrr rir r r [z7StatelessProcessGroup.__post_init__..cSr-r.r r/r r r r1\r2cSr-r.r r/r r r r1]r2)r"r#ranger(r)r+)selfr r r __post_init__Ysz#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_datar(r$setpickledumpsr,appendtime perf_counter)r4r6r7keyr r r send_obj_s zStatelessProcessGroup.send_objrNcCsP|jr&|jd\}}t||jkr|j||jndS|jsdSdS)zAExpire data that is older than `data_expiration_seconds` seconds.rN)r,r?r@r&r$ delete_keypopleft)r4rA timestampr r r r:gs  z!StatelessProcessGroup.expire_datasrcc Cs<t|jd|jd|j|}|j|d7<|S)z%Receive an object from a source rank.r8r9r)r<loadsr$getr"r))r4rFr6r r r recv_objrs 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/r9r)r"r:r*r$r;r<r=r,r>r?r@r+rGrH)r4r6rFrArIr r r broadcast_objzs z#StatelessProcessGroup.broadcast_objcCsVg}t|jD]!}||jkr|||j||jdq|jd|d}||q|S)z$All gather an object from all ranks.rFN)r3r#r"r>rJ)r4r6 gathered_objsr0rIr r r all_gather_objs   z$StatelessProcessGroup.all_gather_objcCs>t|jD]}||jkr|jd|jdq|jd|dqdS)z#A barrier to synchronize all ranks.NrK)r3r#r"rJ)r4r0r r r barriers  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_namerPr# is_master)r"r#r$r&)rr!)rOrPr"r#r&r$r r r createszStatelessProcessGroup.createrN)r%)"__name__ __module__ __qualname____doc__int__annotations__r_C_distributed_c10dStorer& dataclassesfielddictr(r)r*r+rr,rstrfloatr5rrBr:rIrJlistrMrN staticmethodrSr r r r r!CsB       r!rT)F)r^r<r? collectionsrcollections.abcrtypingrrtorch.distributedr1sglang.multimodal_gen.runtime.utils.logging_utilsrrUloggerrrYrTensorboolr dataclassr!r r r r s0