j SSKrSSKrSSKrSSKrSSKJr SSKJrJr SSK J r J r SSK J r SSKJr SSKrSSKJr SSKJr SS KJr S S KJr S S KJrJr S S KJr S SKJrJr S SKJ r SSK!J"r" SSK#J$r$ SSK%J&r& SSK'J(r( SSK)J*r*J+r+ SSK,J-r-J.r. SSK/J0r0 SSK1J2r2 SSK3J4r4J5r5J6r6J7r7 SSK8J9r9J:r:J;r; SSKr> "SS\R~5r@"SS 5rA"S!S"5rB"S#S$5rC"S%S&5rD"S'S(5rEg))N)abstractmethod)Callable Generator)contextmanager nullcontext) perf_counter)Any)nn)tqdm)logging_redirect_tqdm)PretrainedConfig)ContinuousBatchingConfigGenerationConfig)is_flash_attention_requested)is_flash_attn_2_availableis_flash_attn_3_available)logging)LogitsProcessorList)PagedAttentionCache)%ContinuousBatchingLogitsProcessorList)DistributedHelper)"resolve_continuous_batching_config%update_cb_config_after_cache_creation)ContinuousBatchingAsyncIOsContinuousBatchingIOs) ModelRunner)OffloadingManager)GenerationOutput RequestState RequestStatuslogger)SCHEDULER_MAPPING FIFOScheduler Scheduler) WorkloadHints drain_queuec\rSrSr%\\S'\R\S'\R\S'\ S\ SS4Sj5r \ S \ S\ 4S j5rS rg) ProtoPretrainedModelFconfigdtypedeviceattn_implementationreturnNcgN)selfr0s C:\Users\johnn\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.13_qbz5n2kfra8p0\LocalCache\local-packages\Python313\site-packages\transformers/generation/continuous_batching/continuous_api.pyset_attn_implementation,ProtoPretrainedModel.set_attn_implementationK generation_configcgr3r4)r5r;s r6_get_logits_processor*ProtoPretrainedModel._get_logits_processorOr9r:r4)__name__ __module__ __qualname____firstlineno__r__annotations__torchr.r/rstrr7rrr=__static_attributes__r4r:r6r+r+Fs`  ;; LL 3 4   7G L_  r:r+cb\rSrSrSrSSjrS\SS4SjrS\\SS4S jr S \ S \ SS4S jr S r g) OutputRouterTaDedicated object for routing generation outputs to the right destination. When an async handler is registered for a request, the output is forwarded to that handler via ``call_soon_threadsafe``. Otherwise the output is placed on the shared ``output_queue``. r1Ncz[R"5Ul0Ul[R "5Ulgr3)queueQueue output_queueresult_handlers threadingLock_lockr5s r6__init__OutputRouter.__init__\s&!KKMVX^^% r:outputcUR URRUR5nSSS5 WbUup4UR X15 gUR R U5 g!,(df  NC=f)zDRoute a single output to its registered handler or the output_queue.N)rQrNget request_idcall_soon_threadsaferMput)r5rUentrycallbackloops r6deliverOutputRouter.deliverasa ZZ((,,V->->?E  "NH  % %h 7    ! !& ) Zs &A00 A>outputscf/nSnUR UH^nURRUR5nUbUupcUR Xd45 MCUR R U5 M` SSS5 U(aUbU4SjnURU5 ggg!,(df  N2=f)zRoute a batch of outputs, using a single ``call_soon_threadsafe`` to minimize cross-thread overhead. Outputs without a registered handler fall back to the shared ``output_queue``. Nc*UH upU"U5 M gr3r4)batchcbouts r6 _run_batch.OutputRouter.deliver_batch.._run_batch|s$GBsG %r:)rQrNrWrXappendrMrZrY)r5r` callbacksr]rUr[r\rfs r6 deliver_batchOutputRouter.deliver_batchks >@  ZZ!,,001B1BC$%*NH$$h%78%%))&1 " )!*   % %j 1 *9Zs A%B"" B0stateerrorc[RUl[U5UlUR UR 55 g)zFail the request and deliver the output to the output router. This is made from the OutputRouter so that it can be called even when a batch processor could not be created.N)r#FAILEDstatusrErmr^to_generation_output)r5rlrms r6fail_and_deliverOutputRouter.fail_and_delivers1%++ %j  U//12r:)rQrMrNr1N)r?r@rArB__doc__rSr!r^listrjr" ExceptionrrrFr4r:r6rHrHTsS& *.*4*2T*:%;22.3l3933r:rHc\rSrSrSrSrSrSrSrSS jr SS jr S \ S \ SS4S jr S\ SS4SjrSSjrS\ SS4SjrS\S-4Sjr\S\ 4Sj5r\S\ 4Sj5rSrg)BackgroundThreadStatuszTracks the status of the background thread locally and in its TP group. The status is an int that can only increase, representing how soon the thread should stop.rrrr r1Nc[R"5UlURUlURUlSUlgr3)rOrP_local_status_lock DONT_STOP _local_status _tp_status fatal_errorrRs r6rSBackgroundThreadStatus.__init__s0"+.."2!^^..r:cURUlUR URUlSSS5 SUlg!,(df  N=f)zClear the local and TP statuses. This method should ONLY be called by the main thread itself BEFORE starting the background thread.N)r}rr|r~rrRs r6clearBackgroundThreadStatus.clears:..  $ $!%D %% $s A Arp global_rankcHXRUR4;a[SUSU35eUR [ XR UR 5UlSSS5 [R"SUSU<SUR <35 g!,(df  N:=f)zwRequest the background thread to stop. This does not take effect immediately, only after the TP group has communicated.zInvalid stop status z from rank NzRank z3 requested background thread to stop with status = z. Now self._local_status = ) FLUSH_AND_STOP HARD_STOP ValueErrorr|maxr~rr$info)r5rprs r6 request_stop#BackgroundThreadStatus.request_stops --t~~> >3F8;{mTU U  $ $!$V-?-?!QD % K= T6+Uq\`\n\n[r s % $s &B B!rmc^URb[R"SU35 gXlg)z3Record a fatal error if none has been recorded yet.Nz:A fatal error was already recorded, ignoring later error: )rr$rm)r5rms r6record_fatal_error)BackgroundThreadStatus.record_fatal_errors*    ' LLUV[U\] ^$ r:crUR URUlSSS5 g!,(df  g=f)zrMark the background thread as stopped. This should be called by the main thread when the generation loop finishes.N)r|STOPPEDr~rRs r6mark_as_stopped&BackgroundThreadStatus.mark_as_stoppeds% $ $!%D % $ $s( 6 tp_statuscXR:a[SU<SUR<35eXlUR [URU5UlSSS5 g!,(df  g=f)z8Update the local and TP statuses with the new TP status.z1TP communicated a lower stop status: tp_status = z, self._tp_status = N)rrr|rr~)r5rs r6update_with_tp_status,BackgroundThreadStatus.update_with_tp_statuss]  &QYNRgUYUdUdThij j#  $ $!$T%7%7!CD % $ $s A$$ A2cVURbgURUR:wagg)zdIf the background thread cannot accept new requests, return the reason why. Otherwise, returns None.Nz.The background thread died with a fatal error.z"The background thread is stopping.)r local_statusr}rRs r6can_accept_new_requests.BackgroundThreadStatus.can_accept_new_requestss)    'C    .7r:cUR$)zVThe locally requested status, possibly ahead of the value agreed upon by the TP group.)r~rRs r6r#BackgroundThreadStatus.local_statuss!!!r:cUR$)zKThe status last agreed upon by the TP group through a MAX-reduce operation.)rrRs r6r BackgroundThreadStatus.tp_statussr:)r~r|rrrt)r?r@rArBrur}rrrrSrintrrwrrrrErpropertyrrrFr4r:r6ryrys?INIG    3  S  T  % %d%. DsDtDt"c""3r:ryc\rSrSr%\\-\S'\\S'S\S\ S\ S\ S\ S \ RS -S \ RS -S \S \S\R$S\R&S\S\SS 4SjrS\4SjrS#SjrS#SjrS\4SjrS\S\SS 4SjrS\4SjrS#Sjr S\4Sjr!Sr"S\SS 4Sjr#\RH"5S\%RLSS 4S j5r'\RH"5S\%RLSS 4S!j5r(S"r)g )$ContinuousBatchProcessorinputs_and_outputs schedulercacher-r;continuous_batching_configlogit_processor input_queueN cancel_queue output_routerbackground_thread_status model_device model_dtypedistributed_helperr1c rXlX lX@lXPlX`lXplXlXlXlXl Xl Xl [USS5Ul URUlURRUR U 5 [USS5cSO UR"UlUR$UlUUUU U URS.nURR&UlUR&(a[)S 0UD6UlO[-S 0UD6Ul[/UU UR0UR2UR*R4URS9UlRURURUR*URURS9Ulg) aInitialize the continuous batch processor. Args: cache: A [`PagedAttentionCache`] object config: The model configuration generation_config: The generation configuration continuous_batching_config: The continuous batching configuration logit_processor: The [`ContinuousBatchingLogitsProcessorList`] object used to process the logits. input_queue: Queue for incoming requests. Is None if this process is not a TP driver. cancel_queue: Queue for cancellation request_ids. Is None if this process is not a TP driver. output_router: An [`OutputRouter`] object that routes outputs to handlers or the output queue. background_thread_status: A [`BackgroundThreadStatus`] object to track the background thread status. model_device: Device for model inputs/outputs model_dtype: Data type for model inputs/outputs scheduler: The [`Scheduler`] to use distributed_helper: The [`DistributedHelper`] to use do_sampleTsliding_windowNr)rr-rr/rr)rrcpu_offload_space_gibsafety_thresholdcompute_streamr)r cb_configrrrreturn_logprobsr4)rr-rrrrrrrrrrgetattrrr set_tp_seedseedrmax_batch_tokensuse_async_batchingrrrr cpu_offload_space"cpu_offload_space_safety_thresholdroffloading_managerr model_runner)r5rr-r;rrrrrrrrrr io_kwargss r6rS!ContinuousBatchProcessor.__init__sB  3.&(*(@%(&""4!!2KF9II ++,F,K,K\Z$+63CT#J#RaX^XmXm % 6 6*D"&#33  #'.."C"C  " "&@&M9&MD #&;&Hi&HD ##4"<"N"N7ZZ22AA#66 # ( 00nn**#66nn 00  r:cSURSURRSURRS3URR 5R 5-$)Nz%ContinuousBatchProcessor(input_queue=z, active_requests=z, waiting_requests=))rractive_requestswaiting_requestsrget_model_kwargs__repr__rRs r6r!ContinuousBatchProcessor.__repr__8sj3D4D4D3EF#~~==>>QRVR`R`RqRqQrrs u%%668AAC D r:cSUl[R"5 [RR 5(a[RR 5 ggr3)rgccollectrDcuda is_available empty_cacherRs r6__del__ ContinuousBatchProcessor.__del__?s;"& :: " " $ $ JJ " " $ %r:cURR5 URR5 URR5 URR 5 g)z4Reset the batch processor for a new generation loop.N)rresetrrrfree_all_requestsrRs r6rContinuousBatchProcessor.resetEsF %%'  %%' $$&r:cdURb8URb+[UR5[UR54nO//4n[US5[US5-nURR nUR RX#5up$URRU5 URR[R:XagUS:XagUR RU5upVUHDnURRUR5 UR R#U5 MF UHn UR R-U 5 M g![$a4n[&R("SU3SS9 UR+X5 SnAMSnAff=f)zCommunicates with the TP group to get A. the new requests and cancellations from the TP driver, and B. an eventual stop signal from any process in the TP group. Returns True if the TP group is hard-stopping, False otherwiseNrrTFzError processing new request: exc_info)rrr)lenrrrtp_all_reduce_staterrryrtp_broadcast_object_from_rank_0r check_kwargslogit_processor_kwargsradd_waiting_requestrwr$rm_handle_request_errorset_request_cancellation) r5payload payload_sizelocal_requested_statusr new_states cancellationsrlerXs r6_update_tp_group_state/ContinuousBatchProcessor._update_tp_group_stateLs{    'D,=,=,I"4#3#34k$BSBS6TUG2hG71:WQZ8 "&!>!>!K!K"&"9"9"M"Ml"s  %%;;IF  ( ( 2 26L6V6V V 1 $($;$;$[$[\c$d!  E 5$$11%2N2NO2259 (J NN 3 3J ?(  5 =aSADQ**144 5sAE11 F/;)F**F/rmrlc[UR[5(a+URR UR5UlO/UlUR RX!5 g)z(Handle general request processing error.N) isinstancerXrEr!get_active_request_static_outputsgenerated_tokensrrr)r5rmrls r6r.ContinuousBatchProcessor._handle_request_errorssQ e&& , ,%)^^%U%UV[VfVf%gE "%'E " ++E9r:cUR5nU(agURR5nUHnURR U5 M URR 5(dgURR URURR5upEpgUckURR5S:Xa [S5eURR URURR5upEpgUcMkU(dgURR(aURR5 URRU5 [R"[ R"5(a[R$"S['U5S['URR(5S['URR*5SUSUS URR-53 5 UR.R1XgU5upgUR2R5UUR6UUUUR.R8S 9 g ) z}Prepare tensors and metadata for the next model forward pass. Returns True if there are requests to process, False otherwise.Frz>No requests can be scheduled and no requests can be offloaded.z Scheduled: z , Waiting: z , Active: z . cum Q: z . cum KV: z, free blocks: )requests_in_batchlogits_processorsuse_decode_fast_path num_q_tokens max_kv_read use_paddingT)rrclear_cancelled_requestsrfree_request_cpu_cachehas_pending_requestsschedule_batchrr num_pagesoffload_requests RuntimeErrorstarved_requestsrestore_scheduled_requestsr$ isEnabledForrDEBUGdebugrrrget_num_free_blocksrmaybe_pad_inputsrprepare_batch_tensorsr pad_inputs)r5 hard_stoppingcancelled_statesrlrrrrs r6prepare_next_batch+ContinuousBatchProcessor.prepare_next_batch}s# 335 >>BBD%E  # # : :5 A&~~2244NR^^MjMj  ! !4::#7#7N J '&&779Q>"#cddQUQ_Q_QnQn%%tzz';';R N \  '! >> * *  # # 4 4 6 ::;LM   w}} - - LLc"345[T^^EdEdAe@fgt~~==>?yW&-tzz7U7U7W6XZ %)$5$5$F$F|bv$w!  55/"22!5%#))44 6 r:cURR5upnSn/nUGHnURnUR[R [R 4;aYUR(aUR(aUS- nMg[SURRSURS35eUR(GaUR5S:Xa[RUlX$nUbX4OSn US- nURX5n URR!XvR"5 U (a6UR$R'UR5 SUR$lUR*(dUR[R :Xa"UR-UR/55 GMGMUR[R0:XdGMURR!XvR"5 GM U(aUR2R5U5 //pUR$R6(GaUR$R6R95n U R:nSU l[=U5Vs/sHoRSU3PM nn[?XRRAU R55nUU- n[=U5HEnUR95nU RC5nUUl UR$REU5 MG US:XaGM UH+nU RGU5UR$RHU'M- URRKU RU5unnU RMU5 U RMU5 UR$R6(aGMU (aiURRNnUb[PRRRUU5O [W5nU URRYX5 SSS5 ggs snf!,(df  g=f) z0Update request states based on generated tokens.rrzTried to update z request z in sync mode.NFz__child#)-rprepare_batch_updaterlrpr#FINISHEDPENDINGr has_new_tokenrnamerX generated_lenDECODINGupdate_and_check_completionr!mark_shareable_blocks_as_completecomplete_blocksrfinish_requestblock_new_requests streamingrhrq PREFILLINGrrj_requests_to_forkpop num_childrenrangemincompute_max_num_forks!create_equivalent_initial_requestrforkr fork_requestextendrrDrstreamr copy_cache)r5r new_tokenslogprobscurrent_logits_indexpending_outputs future_staterltokenlogprob is_finished copy_sourcecopy_destination state_to_forkrinew_request_ids num_to_forknum_to_schedule_new_request_id child_statecopy_srccopy_dstr maybe_streams r6 update_batch%ContinuousBatchProcessor.update_batchs262I2I2^2^2`/x -L &&E|| 6 6 8M8MNN**#11,1,"%5ell6G6G5H RWRbRbQccq#rss)))&&(A-#0#9#9EL"8#2"&!8!89Q9QSb!c Hh   x (  # #H -/nn...4 "44CCN@N@Z5::,,^<`k`mL %%kD +f4sQ0Q Q)c6URR5$)z2Check if there are any active or waiting requests.)rrrRs r6r-ContinuousBatchProcessor.has_pending_requests s~~2244r:cURR5SnUHMnURXR5 URR URR 5 MO g)z&Handle errors during batch processing.rN)rr rrlrrrX)r5rmfailed_future_statesr*s r6handle_batch_error+ContinuousBatchProcessor.handle_batch_errorsX#66KKMaP0L  & &u.@.@ A NN ) ),*<*<*G*G H1r:c,[URRR55nUH9nUR X5 URR UR 5 M; URR5 [URRR55H9nURRRU5nUR X5 M; URRR5 g)z.Fail all active requests with the given error.N)rvrrvaluesrrrXrfree_all_waiting_cpu_cachesrkeysrwaiting_requests_orderr)r5rmrequestsrlreq_ids r6fail_all_requests*ContinuousBatchProcessor.fail_all_requestss66==?@E  & &u 4 NN ) )%*:*: ; ;;=4>>::??ABFNN3377?E  & &u 4C --335r:modelcURRURRS9nURR X5 URR 5 g)z!Perform a single generation step.)rN)rrrr compute_batchretrieve_device_outputs)r5rL batch_datas r6_generation_step)ContinuousBatchProcessor._generation_step&sS ,,==$J[J[JfJf=g  '': 779r:c:URRU5 g)a7Pre-capture CUDA graphs (or trigger compile warmup) for varlen and decode paths. In async mode, both IO pairs are warmed up since each has its own graph buffer and static tensors. The varlen path is warmed up at the largest possible `(q, kv)` sizes so subsequent captures fit inside it without growing the pool.N)rwarmup)r5rLs r6rTContinuousBatchProcessor.warmup5s   'r:)rrrrr-rrrrrrrrrrrrrrrrt)*r?r@rArBrrrCr'rrrrrrKrLrHryrDr/r.rrSrErrrboolrrwr"rr r;rrArJno_gradr ModulerQrTrFr4r:r6rrs-0JJJ\ "\ !\ , \ %= \ ? \ [[4'\ kkD(\ $\ #9\ ll\ [[\ \ .\  \ | # % '%%N:9:\:d:>D>@KEZ5d5I6y6T6" ]]_ :bii :D : : ]]_(BII($((r:rc\rSrSrSrS0S\S\S\S\S-SS4 S jjr S\SS4S jr S1S jr S\ 4S jr S1S jrS2S\ S\S-S\ S\ SS4 SjjrS0S\S\S-SS4SjjrS1SjrS3S\\S\S-S\S-S\ S\ S\\\-S-S\S\S-4SjjrS4S\\\S\S-S\ S\ S\S\\4 SjjrS\SS4S jrS5S\S-S\S-S\S-4S!jjrS"rS\S\\4S#jrS\S$\SS4S%jr S&\!S'\ S\ 4S(jr"\#RH"5S1S)j5r%S1S*jr&S\!4S+jr'S,\(S&\!S-SS4S-jr)S,\(S&\!S-SS4S.jr*S/r+g)6ContinuousBatchingManageri>aManager for handling continuous batching of generation requests. It provides a user interface for submitting generation requests, retrieving results, and managing the background generation thread. This class should not be created directly, but through one of the following entry points (all methods of the `ContinuousMixin` mixin): - `init_continuous_batching` - `continuous_batching_context_manager` - `generate_batch` NrLr;rworkload_hintsr1c[R"URS9Ul[R"5UlSUl[ R"5Ul[ R"5Ul [5Ul [5UlSUlSUlSUlSUlUR'U5 UR)5UlX l[/USS5nUbUOSUl[3[/UR*SS5UR4S9UlUR6R8UlUR:(aUR6R=5 [?UR*RAU5URBURDS 9Ul#[IUR*RJRM5UUURFRNS 9Ul(URPRRUl*g) aIInitialize the continuous batching manager. Args: model: The language model for generation generation_config: Configuration for generation parameters continuous_batching_config: Configuration for continuous batching parameters workload_hints: Workload hints for the continuous batching initialization (optional) )maxsizerNFnum_return_sequencesr _device_mesh) device_meshcpu_group_timeout)logits_processorper_request_processorsdrop_unsupported_processors)r-rr[has_logit_processors)+rKrLmax_queue_sizerr_request_counterrOrP _request_lockEvent_has_new_requestsryrrHrbatch_processor_generation_thread warmed_up_original_attn_implswitch_to_cb_friendly_attnevalrLr;rr^rrar is_tp_driverdisable_nccl_graph_mixingmaybe_warn_nccl_graph_mixingrr=rcrdrrr-get_text_config do_processingrallow_block_sharing_use_prefix_sharing)r5rLr;rr[r^s r6rS"ContinuousBatchingManager.__init__Gs !;;/I/X/XY.3kkm !&^^-!*!2)?(@%)^@D"&$(  ''.ZZ\ "3&'8:PRVWOP#=#T#T(B(^(^ +M::$$4460)!%!5!5!C!C + '$(#B#B#V#V r:cURRnUn[US9nSU;nU(dU(dxUR(ag[ SS9(aSnO[ SS9(aSnOSnSnUb%S U3n[ R"US US US 35 O[ R"US 35 SU;aURSU35 X l gg)zSwitch the attn implementation to one that is CB friendly: try to find a flash implementation if flash is requested and, in any cases, switch to a paged implementation.)"requested_attention_implementationzpaged|T)kernels_fallback_okr rNz>Continuous batching is much better when using flash attention.flash_attention_z Switching from z to z_. If you need to use eager or sdpa, use paged|eager or paged|sdpa as the `attn_implementation`.zE Consider using a flash `attn_implementation` when loading the model.) r-_attn_implementationr_supports_flash_attnrrr$warningrr7rn)r5rLoriginal_attn_impl target_implemis_flashis_pagedversionmsgs r6ro4ContinuousBatchingManager.switch_to_cb_friendly_attns#\\>>* 0S`a},U-G-G(TB*tDRC""27) < e+,>+?tM?Stt  se#hij = (  ) )F=/*B C'9 $ )r:cURcUR5UlURRUR5 SUlg)zPre-capture CUDA graphs for varlen and decode paths by running dummy batches. Initializes the batch processor if not already done.NT)rk_create_batch_processorrTrLrmrRs r6rT ContinuousBatchingManager.warmups@    '#'#?#?#AD  ##DJJ/r:c`URSL=(a URR5$)zUReturns True if the background generation thread has been started and is still alive.N)rlis_aliverRs r6 is_running$ContinuousBatchingManager.is_runnings'&&d2Yt7N7N7W7W7YYr:c UR5(a[R"S5 gURR 5 [ R "URS9UlURR5 g)z'Start the background generation thread.z"Manager thread is already running.N)target) rr$rrrrOThread_run_generation_looprlstartrRs r6rContinuousBatchingManager.starts\ ??   NN? @ %%++-"+"2"2$:S:S"T %%'r:blocktimeoutkeep_for_next_session hard_stopc(URc[R"S5 URc%SnU(aUS- n[R"U5 g[ 5nU(a[ R O[ RnURRXpRR5 U(aURXb5 U(dSUlO&[R"S5 XRlUR b,URR#UR 5 SUl[$R&"5 [(R*R-5(a[(R*R/5 gg)aStop the background generation thread. If the `block` flag is set to True, then this method waits for the thread to stop for a maximum time of `timeout` seconds (None means no timeout). If the `keep_for_next_session` flag is set to True, then the manager is cached on the model for future use. If the `hard_stop` flag is set, the background generation thread will be stopped immediately and pending requests will be failed.Nz% Batch processor was not initialized.zManager not started.z? Hence the unstarted manager will not be kept for next session.z:Continuous batching manager will be kept for next session.)rkr$rrlrryrrrrrrjoinrrL#_cached_continuous_batching_managerrnr7rrrDrrr)r5rrrrrstop_trigger_time stop_statuss r6stopContinuousBatchingManager.stops%    ' NNC D  " " *(C$XX NN3 )N:C,66I_InIn  %%22;@W@W@c@cd  II' 1%#'D  KKT U=AJJ :  # # / JJ . .t/G/G H'+D $ :: " " $ $ JJ " " $ %r:rc"URcgURRUS9 URR5(a[R"SUS35 g[ 5n[R "SX1- SS35 SUlg)zeWait for the background thread to finish. Wait can be capped using the timeout argument (in seconds).Nrz3Generation thread did not exit after join timeout (z).z+Background generation thread stopped after .2fs.)rlrrr$rrr)r5rrends r6rContinuousBatchingManager.joins  " " * $$W$5  " " + + - - NNPQXPYY[\ ].C KKEcF]^aEbbde f&*D #r:cUR5(aURSSS9 URR5 g)zTerminate the manager and release distributed resources. Safe to call multiple times. After calling this, the manager cannot be restarted.TF)rrN)rrrdestroy_cpu_comm_grouprRs r6destroy!ContinuousBatchingManager.destroys3 ??   IIDI > 668r: input_idsrXmax_new_tokensrrecord_timestamps eos_token_idrc UR(dgURR5nUb3USSSSS-USSSS-n [R"USU S35 gUc9UR  S UR 3nU=R S- slSSS5 UcURROUnUcURROUn[U[U5URS- UUUUUS 9n URRU S S S 9 URR!5 U$!,(df  N=f)aoAdd a new generation request to the queue. If the process is not a TP driver, this is a no-op. Args: input_ids: Input token IDs to use as prompt request_id: Optional custom request ID (auto-generated if None) max_new_tokens: Maximum number of new tokens to generate streaming: Whether to stream tokens as they're generated record_timestamps: Whether to record timestamps for each generated token eos_token_id: End-of-sequence token ID(s) logit_processor_kwargs: Keyword arguments for the logits processor. Returns: str | None: The request ID if the process is a TP driver, None otherwise. Nr z, ..., rz. Request with ids z will be dropped.req_)rXinitial_tokensrrrrrrT rr)rqrrr$rrhrgr;rrr"rvr^rrZrjset) r5rrXrrrrr denial_msgpreviewrls r6 add_request%ContinuousBatchingManager.add_requestsW2  22JJL  !"2A"- 9y~>NPQPR>^l>J>Rt--::Xd ! ?22Q6/)%#9   U$; ""$+$#s 2%D22 Einputsc 8[U5nUR [URURU-5Vs/sHnSU3PM nnU=RU- slSSS5 [ [ WU55n UR (a [U SSS9n URRn U c URRROU n U cSOU n U HupUR"SU U UUUU S.UD6 M U$s snf!,(df  N=f) zgUtility function to batch `add_request` and return their IDs. Check its documentation for more details.rNc US$)Nrr4)xs r68ContinuousBatchingManager.add_requests..Ys!A$r:T)keyreverser)rrXrrrrr4) rrhrrgrvziprwsortedr;rrLr-r) r5rrrrr num_requestsr1 request_idsids_and_inputsrrXrs r6 add_requests&ContinuousBatchingManager.add_requestsHs6{   /4T5J5JDLaLadpLp/qr/q!T!:/qKr  ! !\ 1 ! c+v67  # ##NPTUN--:: 9E9Mtzz((55S_ )1r| %3 !J    #%-#"3)  ) &4+s s&D D D D  DcUR(a6URRU5 URR 5 gg)zCancel a request by its ID. If this called from a process that is not a TP driver, it's a no-op: only TP driver processes interact with the manager.N)rqrrZrjr)r5rXs r6cancel_request(ContinuousBatchingManager.cancel_requestks8       ! !* -  " " & & ( r:cURRR5(a%URbURR bgURRR SUS9nUb6URU:wa&URRRU5 gU$![Ra gf=f)zRetrieve one result from the output queue. If an ID is provided, returns the first matching request. If a timeout is provided, returns None after the timeout (in seconds).NTr) rrMemptyrlrrrWrXrZrKEmpty)r5rXrresults r6 get_result$ContinuousBatchingManager.get_resultts    * * 0 0 2 2&&.$2O2O2[2[2g ''4488tW8UF%&*;*;z*I""//33F;M{{  sAB/-B//CCc## URbfURR5(aFURSS9nUbUv URb"URR5(aMDgggg7f)z.Iterate over results as they become available.N皙?r)rlrr)r5rs r6__iter__"ContinuousBatchingManager.__iter__sk%%1d6M6M6V6V6X6X__S_1F! %%1d6M6M6V6V6X6X16X1s A/A73A7c# # URb}URR5(a]URUSS9nUbUv UR5(agURb"URR5(aM[gggg7f)zIterate over results matching a specific request id (blocking). Uses the shared output queue with requeue. For high-concurrency serving, use :meth:`register_result_handler` instead. Nr)rXr)rlrrr-)r5rXrs r6request_id_iter)ContinuousBatchingManager.request_id_iters %%1d6M6M6V6V6X6X__ C_HF! %%'' %%1d6M6M6V6V6X6X16X1s BB Br\c^^^[R"5nUUU4SjnTRR XC4TRRT'SSS5 g!,(df  g=f)aRegister a callback for result delivery (streaming or non-streaming). The callback is invoked on the event loop via ``call_soon_threadsafe`` each time a result is produced for this request. For streaming requests, this happens on every token; for non-streaming, only on completion. The handler is automatically cleaned up when the request finishes. Args: request_id (`str`): The request ID to receive outputs for. callback (`callable`): Called with a ``GenerationOutput`` for each result. c>T"U5 UR5(aFTRR TRRR TS5 SSS5 gg!,(df  g=fr3)r-rrQrNr)rr\rXr5s r6 _auto_cleanupHContinuousBatchingManager.register_result_handler.._auto_cleanupsY V !!##''--&&66:::tL.-$--s 'A&& A4N)asyncioget_running_looprrQrN)r5rXr\r]rs``` r6register_result_handler1ContinuousBatchingManager.register_result_handlersO'') M    % %>K=RD   . .z :& % %s A A)rk bootstrappingcUR5nU(a>UR5 U=RS- slU(agUR5 gURR [ R:XagURR [ R:XaUR5(dgURRSS9 URR5 g)aBody of the generation loop. Returns True if the loop should continue, False otherwise. Behaves differently if this is for bootstrapping an async run: in that case, there is no need to update the batch, and the first step should exit the bootstrapping loop.rFTrr) r rQ current_batchr;rrryrrrrjwaitr)r5rkrrequests_availables r6_generation_loop_body/ContinuousBatchingManager._generation_loop_bodys -??A   ! ! #   ! # ,,.  * * 4 48N8X8X X  ) ) 3 37M7\7\ \#88::  " " ' ' ' 4  " " ( ( *r:cjSnURRRS:XaTURRRb3[R R URR5 UR5nXlSUl UR(a-URUSS9(aURUSS9(aMURUSS9(aURUSS9(aM[UR[5(a8SURR- URlUR!5 [#SUR$R&<35nUR)X!5 UR$R35 [,R4"S 5 g![*a3n[,R."S U3SS 9 UR1X15 SnANiSnAff=f!UR$R35 [,R4"S 5 f=f) z6Main processing loop running in the background thread.NrrT)rFrzdGeneration loop finished before this request completed w/ self.background_thread_status.tp_status = zError in generation loop: rzCGeneration loop finished and background thread exited successfully.)rLr/typeindexrDr set_devicerrkrrrrrr current_pairr;rrr_fail_all_remaining_requestsrwr$rm_handle_critical_errorrr)r5rkrmrs r6r.ContinuousBatchingManager._run_generation_loops% _zz  %%/DJJ4E4E4K4K4W %%djj&7&78#::XYYBCoFhFhFuFuBu22?,,.!wTMjMjMtMtLxyE  - -e E  ) ) 9 9 ; KK] ^  < LL5aS9D I  ' ' ; ; <  ) ) 9 9 ; KK] ^s7CG+GB G G= )G83H8G==H2H2c~URc [S5eURRUR5 g)z=Perform a single generation step. This is mostly cuda graphedNzNTried to perform a generation step before the batch processor was initialized.)rkrrQrLrRs r6rQ*ContinuousBatchingManager._generation_steps2    'op p --djj9r:c[USS5n[U[5(aUR5 U$[ UR R R5URUR RUR[UR S05UR RS9nURUl [URURUR URS9 UR"S:aSURlURR&n[(R*"US5nUc [,R."SUS35 [0nU"UURR2URR4S 9n[UUR R R5UR6URUR8UR:(a UR<OSUR:(a UR>OSUR@URBUR RUR RUURS 9 nU$) ziCreate a new batch processor. If an already initialized batch processor exists, it is reset and returned.rkNtp_plan)r-rr/rrr.)r num_blocksruse_prefix_sharingrz Scheduler 'z ' not found. Defaulting to FIFO.)r safety_marginmax_requests_per_batch) rr-r;rrrrrrrrrr)"rrrrrrLr-rtrr/rr.rrwrrrnum_sliding_attention_groupsmax_blocks_per_requestscheduler_typer%rWr$rr&rrr;rrqrrrr)r5rkpaged_attention_cacher scheduler_clsrs r6r1ContinuousBatchingManager._create_batch_processors"$(94@ o'? @ @  ! ! #" "!4::$$446'+'F'F::$$#66DJJ 26**"" ! $9#K#K -55,772CC#77   ! = = AEFD + + B88GG)--ndC  NN[(88XY Z)M!'99GG#'#B#B#Y#Y 3'::$$446"44'+'F'F 00,0,=,=((4.2.?.?**T,,%)%B%B** ((#66 r:rmc(URR[RURR S9 URR S[R5 URX5 URRU5 g)z:Handle critical errors that terminate the generation loop.)rprrN) rrryrrrrrr)r5rmrks r6r0ContinuousBatchingManager._handle_critical_error?su %%22)33AXAXAdAd 3 33A7M7W7WX ))%A %%88?r:cURR5nUbURX5 OURR X15 ML![ R a Of=fUbURU5 gg)zCFail all remaining requests in the input queue and active requests.N)r get_nowaitrrrrrKrrJ)r5rmrkreq_datas r6r6ContinuousBatchingManager._fail_all_remaining_requestsMs~ ++668".#99%J&&77H  {{     &  - -e 4 'sA AA&%A&)rlrjrnrgrhrwrrkrrrrr;rrqrrLr^rrmr3rt)TNFF)NNFFN)NFF)NN),r?r@rArBrur+rrr(rSrorTrVrrfloatrrrrvrrEr rrrr!rrrrrrrrrDrWrrQrrwrrrFr4r:r6rZrZ>s04 BW#BW,BW%= BW &, BW  BWH :0D : :DZDZ( $&+ 0%0%0% $ 0%  0%  0%d +e +edl +d +9"&%)"'/3898$J8d 8  8  8DIo,8#&8 t8z&*"' !T#Y!d ! !  ! #& ! c!F)))S4ZYilpYp" # )E3I#2NbQ R  "M LL< = LLuM_L `%!& LL< = LLuM_L `%!&sBDC?DADDrr progress_barc U(d0$[R"5[R::a[R"S5 SnUc UR OUn U R b U R OSn [U5U -n URSS5n U c U ROU n [[SU55U bU OSU S9n U(a[R"S 5 URUUS S UUU S 9n[[/5n[U U(+S U S3SS9n0nSn/nUnU UnUR!XUS9nUU :aUR#SS9nU(a=UR$nUR'5(aUUU'US- nUR)S5 O7UR+5(d"[R,"S5 [/S5 OUU :aMSSS5 SSS5 SSS5 0n//nnUHOnUR3U5nUb'UUU'UR,bUR5U5 M<M>UR5U5 MQ U(a[R,"SUS35 U(a[R,"SUS35 U$![0a"n[R,"SU3S S9 SnANSnAff=f!,(df  N=f!,(df  N=f!,(df  N=f)aGenerate sequences for a batch of prompts using continuous batching. Args: inputs: List of input token sequences (prompts) generation_config: Optional generation configuration continuous_batching_config: Optional continuous batching configuration record_timestamps: If set to true, the requests will have a timestamp for each token generated progress_bar: If set to true, a progress bar will be displayed persistent_manager: whether to persist the manager after the generation is finished. Default is False. warmup: whether to pre-capture CUDA graphs before processing requests. Default is True. Returns: `dict[str, GenerationOutput]`: a dictionary of request ids to GenerationOutput objects z=Progress bar is disabled when logger level is less than DEBUGFNrrc38# UHn[U5v M g7fr3)r).0rs r6 1ContinuousMixin.generate_batch..s!I&Y#i..&sr)max_prompt_lengthmax_generated_lengthra,Since you passed `persistent_manager=True`, the manager will be kept alive after the generation is finished. However, it was sized specifically for the requests passed in `generate_batch`. If you plan to reuse the manager for a very different workload, you might want to create a new manager instead.T)r;rrrrrTr[zSolving z requestsrequest)totaldisabledescunit)rrrrz*Generation thread terminated unexpectedly.zCReturning results of generate_batch despite unexpected termination.zError during batch generation: rz Requests z not found in results.z failed during generation.)r$getEffectiveLevelrrrr;r^rrrr(rrrr r rrrXr-updaterrmprintrwrWrh)r5rr;rrrrrTkwargsgen_cfgr^rrr[ manager_cm logging_cmpbar_cmresultsfinished_countrrpbarrrIrreordered_results missing_keys failed_keyss r6generate_batchContinuousMixin.generate_batchs4I  # # % 6 KKW X L->,E$((K\?F?[?[?gw;;mn6{%99  $4d;3A3I//~'!!I&!II3A3MST%   NNv ==/'A1)> +F84 %%L>3   7J4 S%22!Te3 %|3$///:F!'!2!2!--//.4GFO*a/N KKN$//11 %QRcd%|3 18JZ.$&k !F[[(F!,2!&)<<+&&v.,##F+"  LL9\N2HI J  LL9[M1KL M  ' S >qcBTR S'18JJZZsm%K(K +J;-BJ  J; J K K J8 J3 .J;3J8 8J;; K K  K K K+r4)NNNrt)NTNNFTN)NNFFFT)r?r@rArBrurrCrDrWrr(rZr rrrVrrrrvrdictrEr!r2rFr4r:r6rr^s(' ]]_6:FJ/3 8 +d28 %=t$C8 &, 8 # 8 8 tA ]]_6: $FJ#(/3#"+d2#"#" #" %=t$C #" ! #"#"&,#" , -#"#"L ]]_6:FJ"'"#(s!T#Ys!,d2s!%=t$C s!  s!  s!!s!s! c## $s!s!r:r)FrrrKrOabcrcollections.abcrr contextlibrrtimertypingr rDr r tqdm.contrib.loggingr configuration_utilsrgeneration.configuration_utilsrr utils.genericrutils.import_utilsrr utils.loggingrlogits_processrrrcb_logits_processorsr distributedrinitializationrr input_outputsrrrrrr rHr!r"r#r$rr%r&r'utilsr(r)rXr+rHryrrZrr4r:r6rFs /2 63X9V$0&G*eL%1KKBB-,  299  3333lHHXd(d(P ]5]5@j!j!r: