^j#UddlZddlmZddlmZmZmZddlZddlm Z ddl m Z e Z eed<esejjrddlmZneZedZGd d Zy) N) timedelta) TYPE_CHECKINGAnyTypeVar)loggerdist) DeviceMeshTcdeZdZdZdedzdedzddfdZededzddfdZededzdedzfdZ de fd Z dd Z d e jde jfd Zd ededeeeffdZdd e jde de jfdZdedefdZddZdedzde j.ddfdZy)DistributedHelperzlA helper class to handle distributed-related operations. Notably, it does not crash when distributed is off. device_meshNcpu_group_timeoutreturncZtjxrtj|_||_|j |j|j |j}||jstd|d|jrtjnd|_ |jrtjnd|_ ||j|_ |j|_tj!|jd|_|j%|_tj)|j}|dn t+|}tj-|d||_n#d|_ d|_d|_d|_d|_|j1|_|j|jz|_|j|jz|_t9j:ddgt8j<d |_y) Nz.Distributed is off but received device_mesh = .rr)secondsgloo)ranksbackendtimeoutcpudtypedevice) r is_availableis_initializeddist_onrcheck_device_mesh_for_cbextract_tp_mesh ValueErrorget_rank global_rankget_world_size world_sizesizetp_size get_grouptp_groupget_global_ranktp_root_global_rankget_local_rank tp_local_rankget_process_group_ranksr new_groupcpu_comm_groupinfer_if_tp_driver is_tp_driverdp_rankdp_sizetorchtensorint64 _cpu_int_acc)selfrrtp_meshtp_ranksrs /var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/transformers/generation/continuous_batching/distributed.py__init__zDistributedHelper.__init__)s((*Dt/B/B/D & %%d&6&67&&t'7'78  t||N ?OqQR R/3ll4==?37<<$--/Q  "<<>DL#--/DM'+';';DMM1'MD $!(!7!7!9D 33DMMBH/7dYO`=aG"&..xY`."aD DL DM'(D $!"D "&D !335''4<<7 $,,6 "LL!Qu{{5Qc|y|jyd|jvr&|djdkDrtd|dyy)z?Checks the validity of the device mesh for continuous batching.NfsdprzFFSDP is not compatible with continuous batching but got device_mesh = r)mesh_dim_namesr&r!rs r<rz*DistributedHelper.check_device_mesh_for_cbTsa    % % - [// /K4G4L4L4NQR4RfXcWgghij j5S /r>c|y|j|jdkDr|SdSd|jvr|djdkDr|dSy)zJExtracts the TP mesh from the device mesh if it exists and is non-trivial.Nrtp)rAr&rBs r<r z!DistributedHelper.extract_tp_meshase    % % -"-"2"2"4q"8; Bd B ;-- -+d2C2H2H2JQ2Nt$ $r>c |jdk(S)Nr)r-r9s r<r1z$DistributedHelper.infer_if_tp_driverns!!Q&&r>cj|j'tj|jd|_yy)zDestroys the CPU comm group.N)r0r destroy_process_grouprFs r<destroy_cpu_comm_groupz(DistributedHelper.destroy_cpu_comm_groupqs/    *  & &t':': ;"&D  +r>valuec~|jdkDr-tj||jd|j|S)zPInside each TP group, broadcasts the given value from rank 0 to all other ranks.rF)srcasync_opgroup)r'r broadcastr+r))r9rJs r<tp_broadcast_from_rank_0z*DistributedHelper.tp_broadcast_from_rank_0ws4 <&>VZVcVcN d r> payload_size stop_statusc(|jdkDr||jd<||jd<tj|jtjj d|j |jj\}}||fS)zBroadcasts two information: 1. the size of the payload held by the TP driver (all other rank broadcast 0) and 2. the requested stop status (all to all). These information are broadcasted through a MAX-reduce operation.rrF)oprMrN)r'r8r all_reduceReduceOpMAXr0tolist)r9rQrRs r<tp_all_reduce_statez%DistributedHelper.tp_all_reduce_state}s <on_cpuc|jdkDrJ|r |jn |j}tj |tj j ||S)zInside each TP group, all-reduces a tensor with the MIN op. No-op when TP is off. If the tensor is on CPU, it is all-reduced on the CPU comm group.r)rTrN)r'r0r)r rUrVMIN)r9rJrZrNs r<tp_all_reduce_minz#DistributedHelper.tp_all_reduce_minsD <objc|jdkr|S|jr|gndg}tj||j|j t jd|dS)aIInside each TP group, broadcasts an arbitrary picklable Python object from TP-rank 0 to all other ranks. Used to keep request ingress and cancellations consistent across TP workers without requiring all ranks to receive the same external request stream. Uses a dedicated CPU (gloo) `cpu_comm_group` for broadcast.rNr)rLrNrr)r'r2r broadcast_object_listr+r0r5r)r9r^holders r<tp_broadcast_object_from_rank_0z1DistributedHelper.tp_broadcast_object_from_rank_0si <<1 J++#$ "" 008K8KTYT`T`afTg # ayr>c|jdkD}tjjddk7}|r|rt j dyyy)aThrows a warning if TP is on and NCCL's graph mixing support was supposed to be disabled but isn't. That can happen if the distributed group is created before graph mixing is disabled. Typically, if the model is initialized before the ContinuousBatchingConfig is created.rNCCL_GRAPH_MIXING_SUPPORT0aNCCL_GRAPH_MIXING_SUPPORT was not set to '0' before init_process_group: performance will be harmed. Construct your `ContinuousBatchingConfig(...)` BEFORE calling `from_pretrained(tp_plan='auto')`, or set NCCL_GRAPH_MIXING_SUPPORT=0 in the launch environment.N)r'osenvirongetrwarning)r9tp_ongraph_mixing_not_disableds r<maybe_warn_nccl_graph_mixingz.DistributedHelper.maybe_warn_nccl_graph_mixingsK q $&JJNN3N$OSV$V! . NNM /5r>seed model_devicec|)tjdddtj|}n&tj|tj|}|j |}|j }|j dk(r|tjd|dtj||jzy)Nrl)rrzHFound no user-specified seed in the config. Setting the config seed to: r) r5randintr7r6rPitemr#rinfo manual_seedr3)r9rmrntp_seed_tensortp_seeds r< set_tp_seedzDistributedHelper.set_tp_seeds <"]]1iU[[YefN"\\$ekk,WN66~F %%'   q T\ KKbcjbkklm n 'DLL01r>)rN)F)__name__ __module__ __qualname____doc__r floatr= staticmethodrr boolr1rIr5TensorrPinttuplerYr]r rbrlrrvr>r<r r &s3v)RJ$5)R%RV,)R[_)RV kj4.? kD k k Z$%6 :;L  'D'' ellu|| ))#)%PSUXPX/)u||Tell 1    2d 2%,, 24 2r>r )rfdatetimertypingrrrr5torch.distributed distributed_distrequestsrr __annotations__rtorch.distributed.device_meshr objectr r rr>r<rs] .. ! cE%%2248J CLN2N2r>