.`i"ddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZe eZdefdZded ed ed ed eded ejfdZdejded efdZdejd ed ejfdZdejd efdZdedededed eded eeejdzeffdZ ddedededededzdedzdejdzd ed eeejdzeffdZdS)N)ParallelConfig) get_dp_group) init_logger)check_ubatch_thresholdsis_last_ubatch_emptyparallel_configctj}tj}|jr/tdd}tj}||fS)Nz=Using CPU all reduce to synchronize DP padding between ranks.cpu)rdevice device_group#disable_nccl_for_dp_synchronizationlogger info_once cpu_group)rr groups k/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/vllm/v1/worker/dp_utils.py_get_device_and_grouprs_ ^^ "F NN 'E :) K   ( 5= should_ubatch should_dp_padorig_num_tokens_per_ubatchpadded_num_tokens_per_ubatchcudagraph_modereturnc4|j}|j}t|\}} tjd||tj} || d|<|| d|<|rdnd| d|<|rdnd| d|<|| d|<t j| | | S) Nr dtyper)r)data_parallel_sizedata_parallel_rankrtorchzerosint32dist all_reduce) rrrrrrdp_sizedp_rankr rtensors r_run_arr-&s0G0G)/::MFE [GF%+ F F FF3F1Ig5F1Ig+2F1Ig+2F1Ig'F1IgOF%(((( Mrr, num_ubatchesc|dddf}|dddf}ttj|ddk}|sdSt |}t |}t|||rt d||d}|S)Nrrr FzAborting ubatching %s %s) boolr%allitemintminmaxrrdebug)r,r.orig_num_tokens_tensorpadded_num_tokens_tensorrorig_min_num_tokenspadded_max_num_tokenss r_post_process_ubatchr;;s#AqqqD\%ad|uya88==??@@M u488::??AABB 8 < < > > C C E EFF/1F UU &(;=R    rc|dddf}|ret|}tj|gt |zdtjS|S)Nrr r)r3r5r2r%r,lenr'r )r,rnum_tokens_across_dpmax_num_tokenss r_post_process_dp_paddingr@Os!!QQQ$< *15577<<>>??|  s#788 8+    $'')))rc|t|dddfS)z Synchronize cudagraph_mode across DP ranks by taking the minimum. If any rank has NONE (0), all ranks use NONE. This ensures all ranks send consistent values (all padded or all unpadded). r"N)r3r4r2)r,s r_post_process_cudagraph_moderB^s6 vad|!!&&(( ) ))rnum_tokens_unpaddednum_tokens_paddedshould_attempt_ubatchingshould_attempt_dp_paddingc|||ksJt||||||}ttj|ddk}||ksJt ||j}|r |stddd}t||} t|} || | fS)a 1. Decides if each DP rank is going to microbatch. Either all ranks run with microbatching or none of them do. 2. Determines the total number of tokens that each rank will run. When running microbatched or if should_attempt_dp_padding is True, all ranks will be padded out so that the run with the same number of tokens 3. Synchronizes cudagraph_mode across ranks by taking the minimum. Returns: tuple[ should_ubatch: Are all DP ranks going to microbatch num_tokens_after_padding: A tensor containing the total number of tokens per-microbatch for each DP rank including any DP padding. synced_cudagraph_mode: The synchronized cudagraph mode (min across ranks) ] )rrrrrrr!rzzMicrobatching has been triggered and requires DP padding. Enabling DP padding even though it has been explicitly disabled.global)scopeT) r-r0r%r1r2r;r.r debug_oncer@rB) rCrDrErFrrr,rrnum_tokens_after_paddingsynced_cudagraph_modes r_synchronize_dp_ranksrMgs4  3 3 3 3 3 ./#6%6%' F6!9>227799::M % 5 5 5 5)1MNNM]       8   9@@ 24I IIrallow_microbatchingallow_dp_paddinguniform_decode num_scheduled_tokens_per_requestc|jdkrdd|fSd}|r|Jt|||}||}t||||||\} } } | | | fS)a Coordinates amongst all DP ranks to determine if and how the full batch should be split into microbatches. Args: num_tokens_unpadded: Number of tokens without accounting for padding allow_microbatching: If microbatching should be attempted allow_dp_padding: If all DP ranks should be padded up to the same value parallel_config: The parallel config num_tokens_padded: Number of tokens including any non-DP padding (CUDA graphs, TP, etc) uniform_decode: Only used if allow_microbatching is True. True if the batch only contains single token decodes num_scheduled_tokens_per_request: Only used if allow_microbatching is True. The number of tokens per request. cudagraph_mode: The cudagraph mode for this rank (0=NONE, 1=PIECEWISE, 2=FULL) Returns: tuple[ ubatch_slices: if this is set then all DP ranks have agreed to microbatch num_tokens_after_padding: A tensor containing the total number of tokens per-microbatch for each DP rank including padding. Will be padded up to the max value across all DP ranks when allow_dp_padding is True. synced_cudagraph_mode: The synchronized cudagraph mode (min across ranks) ] rFN)rP)r#rrM) rCrNrOrrDrPrQrrErrKrLs rcoordinate_batch_across_dprSsL)Q..dN** % )))#:  )$ $ $   /    $      E],.C 35J KKr)NNNr)numpynpr%torch.distributed distributedr( vllm.configrvllm.distributed.parallel_stater vllm.loggerrvllm.v1.worker.ubatch_utilsrr__name__rrr0r3Tensorr-r;r@rBtuplerMndarrayrSrrras  &&&&&&888888###### X  >$!$#&    $  \*ST( *U\ *$ *5< * * * ***#****CJCJCJ#CJ $ CJ  CJ $ CJ 4$c )*CJCJCJCJV%)"&:>CLCLCLCLCL$ CL Tz CL 4K CL')j4&7CLCL 4$c )*CLCLCLCLCLCLr