-`i8Q `ddlZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZddlZddlmZmZddlmZmZmZmZddlmZddlmZdd lmZdd lmZdd l m!Z!dd l"m#Z#ee$Z%ej&dd dkp#ej&dddkoej&ddkZ'dZ(dZ)dZ* d.dej+de,de-de ej+fdZ.de,de,de,de/e,e,ffdZ0ej1Gdd Z2d!ed"e,d#e,d$e def d%Z3d&e4d'e,d(e,d)e,d*e4def d+Z5d,eddfd-Z6dS)/N)deque)Sequence) timedelta)Any) ProcessGroupTCPStore)Backend PrefixStore_get_default_timeout_unregister_process_group) rendezvous) init_logger) get_tcp_uri)suppress_stdout)is_torch_equal_or_newer)r )r cftrtjdStjddS)Nr)USE_SCHED_YIELDos sched_yieldtimesleepj/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/vllm/distributed/utils.pyrr.s/  1 rcT||zdksJd||dS)z6Ensure that numerator is divisible by the denominator.rz{} is not divisible by {}N)format numerator denominators r ensure_divisibilityr&5s< { "a ' ' ')D)K)K;** ' ' ' ' 'rc,t||||zS)zXEnsure that numerator is divisible by the denominator and return the division value.)r&r#s r divider(<s ;///  ##rFtensornum_partitionscontiguous_split_chunksreturnc|dz }t|||}tj|||}|rt d|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 r)dimc3>K|]}|VdSN) contiguous).0chunks r z.split_tensor_along_last_dim..Zs.AAEU%%''AAAAAAr)r.r(sizetorchsplittuple)r)r*r+last_dim last_dim_size tensor_lists r split_tensor_along_last_dimr<Cst"zz||aH6;;==2NCCM+fmBBBKBAA[AAAAAA rnum_hidden_layerspp_rankpp_sizec tj}| d|dD}n5#t$r(}td||d}~wwxYwt ||kr#tdt |d|dt ||kr#tdt |d |dn||z fd t|D}||zx}r_td |d zD]}|| xxd z cc<t d d d|Dt |d|}|||z} || fS)aTry to evenly distribute layers across partitions. If the number of layers is not divisible by the number of partitions, the remaining layers are evenly distributed across all but the last partition. The last partition is excluded because it often contains an additional norm layer and we are attempting to balance compute. If `pp_size > 2` and the number of remaining layers is `0 < x <= pp_size - 2` then the remaining layers are evenly distributed across the middle partitions. The first and last partitions are excluded because they contain the input and output embeddings respectively and we are attempting to reduce maximum memory consumption across partitions. Nc,g|]}t|Sr)int)r2layers r z"get_pp_indices..rsPPP#e**PPPr,zInvalid partition string: {}zlen(partitions)=z does not match pp_size=.zsum(partitions)=z" does not match num_hidden_layers=cg|]}Srr)r2_layers_per_partitions r rDz"get_pp_indices..}sCCCq*CCCrrrzHidden layers were unevenly partitioned: [%s]. This can be manually overridden using the VLLM_PP_LAYER_PARTITION environment variablec34K|]}t|VdSr0)str)r2ps r r4z!get_pp_indices..s(44AQ444444r) envsVLLM_PP_LAYER_PARTITIONr7 ValueErrorr"lensumrangeloggerinfojoin) r=r>r?partition_list_str partitionserrremaining_layersi start_layer end_layerrIs @r get_pp_indicesr]_s 5% PP2D2J2J32O2OPPPJJ   .556HII   z??g % %MJMM'MMMNN N z??/ / /WJWWBSWWWXX X 0 1G;CCCCE'NNCCC 07: :  1.233 $ $A2!# KK?4444444    j'*++Kj11I  ##s1 A##AA#c,eZdZUdZeed<eed<ejjj ed<e j dzed<dZ eed<e j e Zeeefed <e j e Zeeefed <d Zeed <e j e Zeeefed<e j e Zeeeefed<dZdedefdZdZdedefdZdedzdedefdZdedeefdZd"defdZ e! d#dededededed eddfd!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_sizestoreNsocketdata_expiration_seconds)default_factorysend_dst_counterrecv_src_counterrbroadcast_send_counterbroadcast_recv_src_counterentriesc|j|jksJdt|jD|_dt|jD|_dt|jD|_dS)Nci|]}|dSrrr2rZs r z7StatelessProcessGroup.__post_init__.. F F F!A F F Frci|]}|dSrnrros r rpz7StatelessProcessGroup.__post_init__..rqrci|]}|dSrnrros r rpz7StatelessProcessGroup.__post_init__..s*P*P*PA1a*P*P*Pr)r`rarRrgrhrj)selfs r __post_init__z#StatelessProcessGroup.__post_init__s}y4?**** F FuT_/E/E F F F F FuT_/E/E F F F*P*Pt9O9O*P*P*P'''robjdstc2|d|d|j|}|j|t j||j|xxdz cc<|j|tjfdS)z%Send an object to a destination rank.send_to//rN) expire_datargrbsetpickledumpsrkappendr)rtrvrwkeys r send_objzStatelessProcessGroup.send_objs ;;;t4S9;; sFL--... c"""a'""" S$)++./////rc|jrn|jd\}}tj|z |jkr4|j||jndS|jldSdS)zAExpire data that is older than `data_expiration_seconds` seconds.rN)rkrrerb delete_keypopleft)rtr timestamps r r{z!StatelessProcessGroup.expire_datasl !\!_NCy{{Y&)EEE %%c*** $$&&&&l     rsrcr,c tj|jd|jd|j|}|j|xxdz cc<|S)z%Receive an object from a source rank.ryrzr)r}loadsrbgetr`rh)rtrrvs r recv_objzStatelessProcessGroup.recv_objsel JNNNdiNN$2G2LNN O O   c"""a'""" rc|j|kr|d|d|j}|j|t j||xjdz c_|j|tj f|Sd|d|j |}t j |j |}|j |xxdz cc<|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/rzr) r`r{rirbr|r}r~rkrrrjrr)rtrvrrrs r broadcast_objz#StatelessProcessGroup.broadcast_objs 9        GCGG$*EGGC JNN3 S 1 1 2 2 2  ' '1 , ' ' L  dikk 2 3 3 3JPCPP$*I#*NPPC|DJNN3$7$788H  +C 0 0 0A 5 0 0 0Orc g}t|jD]k}||jkr2|||||j?|d|}||l|S)z$All gather an object from all ranks.rN)rRrar`rr)rtrv gathered_objsrZrs r all_gather_objz$StatelessProcessGroup.all_gather_objs t'' / /ADI~~$$S)))""3DI"6666--d-::$$X....r>@timeoutc  |jdkr.dtj}||dn|dd}n"#t$r}t d|d}~wwxYwd|d|j} |j|dn"#t$r}t d |d}~wwxYwtj}t}t||j krtj}||z |krt d |d d t|j D]}||vrd|d|} |j | | |@#t$rYLt$r3} td | t#Yd} ~ d} ~ wwxYwt||j krt#t||j kd|d|j} |j| dn"#t$r}t d|d}~wwxYw|jdkrdStj}t} t| |j kr tj|z |krt d|d d t|j D]}|| vrd|d|} |j | | |@#t$rYLt$r3} td | t#Yd} ~ d} ~ wwxYwt| |j krt#t| |j k t|j D]} |jd|d|n1#t$r$tdd|d|YnwxYw |jd|d|w#t$r$tdd|d|YwxYwdS)aqA robust barrier to synchronize all ranks. Uses a multi-phase approach to ensure all processes reach the barrier before proceeding: 1. Each process signals it has reached the barrier 2. Each process signals that it has confirmed the arrival of all other ranks. 3. Rank 0 waits for all other ranks to signal their departure to ensure that all ranks have departed the barrier first. Args: timeout: Maximum time in seconds to wait for each phase (in seconds) Raises: RuntimeError: If coordination fails or times out rbarrier_rNzFailed to broadcast barrier_idarrival_rH1z Failed to signal barrier arrivalzBarrier timed out after z.2fz secondsz Error checking key existence: %s departure_z"Failed to signal barrier departurez"Barrier departure timed out after zError deleting key: %s)r`uuiduuid4r Exception RuntimeErrorrbr|rrPrarRraddKeyErrorrSdebugrr) rtr barrier_ide arrival_key start_timeprocesses_arrivedcur_timerZrcheck_e departure_keyprocesses_departeds r barrierzStatelessProcessGroup.barriersu. HyA~~6 66 "":1"5555!//!/<<  H H H?@@a G H:99di99  J JNN; - - - - J J JABB I JY[[ &)ee#$$t66y{{H*$w.."#Sg#S#S#S#STTT4?++ " ")))111a11 "JNN3'''%))!,,,,D """LL!CWMMMMMMMMMMM" $%%77 5#$$t66F>Z==$)==  L JNN=$ / / / / L L LCDD! K L 9>> FY[[ '*uu$%%77y{{Z''11"NNNNN 4?++ " "***3:3333 "JNN3'''&**1----D """LL!CWMMMMMMMMMMM" %&&88 7$%%77<t'' V VA T %%&A&A&Aa&A&ABBBB T T T 57R*7R7Rq7R7RSSSSS T V %%&C:&C&C&C&CDDDD V V V 57TJ7T7TQR7T7TUUUUU V V VsAA A2A--A2B B>)B99B>/E?? G G)GGH33 I=I  I./L M&* M&3)M!!M&? O  +P P P33+Q! Q!,hostport store_timeoutc |dk}|rtjtjtj}|tjtjd|||f||}nd}d}t||||t|d|} t||| ||S)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. rrN)secondsF) host_namerra is_masterr use_libuvmaster_listen_fd)r`rarbrcre) rcAF_INET SOCK_STREAM setsockopt SOL_SOCKET SO_REUSEADDRbindlistenfilenorrr_) rrr`rarer launch_server listen_socket listen_fdrbs r createzStatelessProcessGroup.createns0  "M&.&:LMMM  $ $V%68KQ O O O   d| , , ,  " " "%,,..II MI!#m444&   %! $;     r)r)rdr)#__name__ __module__ __qualname____doc__rB__annotations__r6_C_distributed_c10dStorercre dataclassesfielddictrgrhrirjrrkr8rKfloatrurrr{rrlistrr staticmethodrrrr r_r_s IIIOOO 8 % ++++ MD    #'S'''(9{'8'N'N'Nd38nNNN'8{'8'N'N'Nd38nNNN"#C###1B1BSW1X1X1XS#XXXX): (9%(P(P(PGU5e$ %PPPQQQ 0C0c0000   CCt##$ # $s)    DVDVuDVDVDVDVL (, 3 3 3 3 3  3 "% 3  3  !3 3 3 \3 3 3 rr_ prefix_store group_rank group_sizerct5tdrt|||}n'tjd}t||||}ddlm}|||||}tjj}tj d} tdr| || | | ||dddn #1swxYwY|S) ze Stateless init ProcessGroup with gloo backend compatible with different torch versions. z2.6gloo)backendr)ProcessGroupGloorcpuN) rrrOptions"torch.distributed.distributed_c10dr BackendTypeGLOOr6device_set_default_backend_set_sequence_number_for_group_register_backend) rrrrpgoptionsr backend_class backend_typers r init_gloo_process_grouprsv   BB "5 ) ) BB #*6:::G B HGGGGG(( *j'   $/4 e$$ "5 ) ) 2  # #L 1 1 144666 V\=AAA7BBBBBBBBBBBBBBB8 IsCC--C14C1rrr`rarct||}t|}t|}tt ||||\}}}|||}|} t ||} ddlm} | || || |S#t$rt| || |cYSwxYw)a A replacement for `torch.distributed.init_process_group` that does not pollute the global state. The created ProcessGroup object can be used for some operations such as `allreduce`, because it does not depend on the global rank. However, some operations such as `broadcast` cannot be used because it depends on the global rank. # TODO: ask for help from PyTorch team if we need the `broadcast` operation. This function is useful when we are not sure about the total number of processes in the process group. For example, we may have process 1, 2, ..., 8 who want to communicate, and process 9 might be the same process as process 1, or it might be a different process; process 10 might be the same process as process 5, or it might be a different process. In this case, how can we reliably form a communication channel within process 9 and 10, without affecting the communication channel within process 1, 2, ..., 8? One possible solution is to figure out if process 9 and 10 are the same as process 1 and 5 beforehand, and then form a communication channel based on the information, adjusting the ranks and world_size etc. However, figuring out the information is not always easy, and it will interfere with the main communication channel. Our solution is to always form a communication channel with process 1, 2, ..., 8, and then use this function to form another communication channel with process 9 and 10. This way, regardless of whether process 9 and 10 are the same as process 1 and 5, the main communication channel is always formed with process 1, 2, ..., 8, and the additional communication channel is formed with process 9 and 10. rr)current_platform)rrrrr)rrrr) rr r nextr set_timeoutr vllm.platformsr#stateless_init_device_torch_dist_pgNotImplementedErrorr) rrr`rar init_methodrrbrrrrs r .stateless_init_torch_distributed_process_grouprsDdD))KgG"7++G";j'BBBE4 gJJ{E22L 333333CC%!! D       '%!!        s=BB=<B=rctdr|nddlm}||t |jdS)zd Destroy ProcessGroup returned by stateless_init_torch_distributed_process_group(). z2.7r)_shutdown_backendN)rshutdownrrr group_name)rrs r 1stateless_destroy_torch_distributed_process_grouprs_ u%%  IHHHHH"bm,,,,,r)F)7rrr}rcsysrr collectionsrcollections.abcrdatetimertypingrr6torch.distributedrrrr r r r torch.distributed.rendezvousr vllm.envsrM vllm.loggerrvllm.utils.network_utilsrvllm.utils.system_utilsrvllm.utils.torch_utilsrrrS version_inforrr&r(TensorrBboolr<r8r] dataclassr_rrKrrrrr rse $$$$$$ 44444444 433333######000000333333:::::: X   #BQB':5RaRG#@(8(;q(@ $$$%* L"el 8-$-$%(-$36-$ 38_-$-$-$-$` R R R R R R R R j&&&& &  &&&&RC C C #C 14C ?BC C C C C L -, -4 - - - - - -r