[j(ddlZddlZddlZddlZddlmZddlmZmZddl m Z m Z ddl m Z ddlmZddlZddlmZddlmZdd lmZd d lmZd d lmZmZd d lmZd dlmZmZd dlm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*m+Z+ddl,m-Z-m.Z.ddl/m0Z0ddl1m2Z2ddl3m4Z4m5Z5m6Z6m7Z7ddl8m9Z9m:Z:m;Z;ddlZ> Gddej?Z@Gdd ZAGd!d"ZBGd#d$ZCGd%d&ZDGd'd(ZEdS))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_queueceZdZUeed<ejed<ejed<ede ddfdZ ede de fd Z dS) ProtoPretrainedModelconfigdtypedeviceattn_implementationreturnNcdSN)selfr/s hC:\Projects\DSCons\.venv\Lib\site-packages\transformers\generation\continuous_batching\continuous_api.pyset_attn_implementationz,ProtoPretrainedModel.set_attn_implementationK generation_configcdSr2r3)r4r9s r5_get_logits_processorz*ProtoPretrainedModel._get_logits_processorOr7r8)__name__ __module__ __qualname__r__annotations__torchr-r.rstrr6rrr;r3r8r5r+r+Fs  ; L 3 4   ^  7G L_   ^   r8r+cFeZdZdZd dZdeddfdZdeeddfdZdS) OutputRouteraDedicated 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``. r0Ncttj|_i|_t j|_dSr2)queueQueue output_queueresult_handlers threadingLock_lockr4s r5__init__zOutputRouter.__init__\s+!KMMVX^%% r8outputc|j5|j|j}dddn #1swxYwY||\}}|||dS|j|dS)zDRoute a single output to its registered handler or the output_queue.N)rKrHget request_idcall_soon_threadsaferGput)r4rNentrycallbackloops r5deliverzOutputRouter.deliveras Z @ @(,,V->??E @ @ @ @ @ @ @ @ @ @ @ @ @ @ @  "NHd  % %h 7 7 7 7 7   ! !& ) ) ) ) )s 488outputscLg}d}|j5|D]Z}|j|j}||\}}|||f@|j|[ dddn #1swxYwY|r||fd}||dSdSdS)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,|D]\}}||dSr2r3)batchcbouts r5 _run_batchz.OutputRouter.deliver_batch.._run_batch|s.$GBBsGGGGr8)rKrHrPrQappendrGrSrR)r4rX callbacksrVrNrTrUr^s r5 deliver_batchzOutputRouter.deliver_batchks0 >@  Z 2 2! 2 2,001BCC$%*NHd$$h%78888%))&1111  2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2  2)!*      % %j 1 1 1 1 1  2 2))sAA77A;>A;r0N) r<r=r>__doc__rMr!rWlistrar3r8r5rCrCTs}&&&& *.*4****2T*:%;2222222r8rCceZdZdZdZdZdZdZddZdd Z d e d e ddfd Z dd Z de ddfdZ ede fdZede fdZdS)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 r0Ncftj|_|j|_|j|_dSr2)rIrJ_local_status_lock DONT_STOP _local_status _tp_statusrLs r5rMzBackgroundThreadStatus.__init__s)"+."2"2!^.r8cv|j|_|j5|j|_ddddS#1swxYwYdS)zClear the local and TP statuses. This method should ONLY be called by the main thread itself BEFORE starting the background thread.N)rirkrhrjrLs r5clearzBackgroundThreadStatus.clears.  $ 0 0!%D  0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0s .22status global_rankc||j|jfvrtd|d||j5t ||j|j|_dddn #1swxYwYtjd|d|d|jdS)zRequest 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 ValueErrorrhmaxrjrkr$info)r4rnros r5 request_stopz#BackgroundThreadStatus.request_stops $-t~> > >TFTT{TTUU U  $ R R!$VT-?!Q!QD  R R R R R R R R R R R R R R R sK s s6 s s\`\n s s     s!AA!Ac^|j5|j|_ddddS#1swxYwYdS)zzMark the background thread as stopped. This should be called by the main thread when the generation loop finishes.N)rhSTOPPEDrjrLs r5mark_as_stoppedz&BackgroundThreadStatus.mark_as_stoppeds{ $ . .!%D  . . . . . . . . . . . . . . . . . .s "&& tp_statusc||jkrtd|d|j||_|j5t|j||_ddddS#1swxYwYdS)z8Update the local and TP statuses with the new TP status.z1TP communicated a lower stop status: tp_status = z, self._tp_status = N)rkrsrhrtrj)r4rzs r5update_with_tp_statusz,BackgroundThreadStatus.update_with_tp_statuss t & &iYiiUYUdiijj j#  $ D D!$T%7!C!CD  D D D D D D D D D D D D D D D D D DsAA #A c|jS)zVThe locally requested status, possibly ahead of the value agreed upon by the TP group.)rjrLs r5 local_statusz#BackgroundThreadStatus.local_statuss !!r8c|jS)zKThe status last agreed upon by the TP group through a MAX-reduce operation.)rkrLs r5rzz BackgroundThreadStatus.tp_statuss r8rb)r<r=r>rcrirqrrrxrMrmintrvryr|propertyr~rzr3r8r5rfrfs??INIG)))) 0000  3  S  T     .... DsDtDDDD"c"""X"3Xr8rfceZdZUeezed<eed<dedede de de de j d zd e j d zd ed ed ejdejdededd fdZdefdZd!dZd!dZdefdZdededd fdZdefdZd!dZdefdZ dZ!dedd fdZ"ej#de$j%dd fdZ&ej#de$j%dd fd Z'd S)"ContinuousBatchProcessorinputs_and_outputs schedulercacher,r9continuous_batching_configlogit_processor input_queueN cancel_queue output_routerbackground_thread_status model_device model_dtypedistributed_helperr0c||_||_||_||_||_||_||_| |_| |_| |_ | |_ | |_ t|dd|_ |j|_|j |j| t|dddn|j|_|j|_|||| | |jd}|jj|_|jrt)d i||_nt-d i||_t/|| |j|j|jj|j |_t9|j|j|j|j|j |j|_dS) a)Initialize 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_logprobsr3)rr,rrrrrrrrrrgetattrrr set_tp_seedseedrmax_batch_tokensuse_async_batchingrrrr cpu_offload_space"cpu_offload_space_safety_thresholdroffloading_managerr model_runner)r4rr,r9rrrrrrrrrr io_kwargss r5rMz!ContinuousBatchProcessor.__init__sB  3.&(*(@%(&""4!!2KFF9I ++,F,K\ZZZ$+63CT#J#J#RaaX^Xm % 6*D"&#3   #'."C  " I&@&M&M9&M&MD # #&;&H&Hi&H&HD ##4"<"N7Z2A#6 # # # ( 0n*#6n 0    r8cd|jd|jjd|jjd|jzS)Nz%ContinuousBatchProcessor(input_queue=z, active_requests=z, waiting_requests=))rractive_requestswaiting_requestsrget_model_kwargs__repr__rLs r5rz!ContinuousBatchProcessor.__repr__ sq uD4D u u#~= u uRVR`Rq u u u%6688AACC D r8cd|_tjtjr tjdSdSr2)rgccollectr@cuda is_available empty_cacherLs r5__del__z ContinuousBatchProcessor.__del__'sO"& : " " $ $ % J " " $ $ $ $ $ % %r8c|j|j|j|jdS)z4Reset the batch processor for a new generation loop.N)rresetrrrfree_all_requestsrLs r5rzContinuousBatchProcessor.reset-sZ %%'''  %%''' $$&&&&&r8c$|j0|j)t|jt|jf}nggf}t|dt|dz}|jj}|j||\}}|j||jj tj krdS|dkrdS|j |\}}|D]} |j |j|j|=#t$$r9}t'jd|d|||Yd}~{d}~wwxYw|D]} |j| dS)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)lenrr~rtp_all_reduce_stater|rzrfrrtp_broadcast_object_from_rank_0r check_kwargslogit_processor_kwargsradd_waiting_request Exceptionr$error_handle_request_errorset_request_cancellation) r4payload payload_sizelocal_requested_statusrz new_states cancellationsstateerQs r5_update_tp_group_statez/ContinuousBatchProcessor._update_tp_group_state4s   'D,=,I"4#344k$BS6T6TUGG2hG71:WQZ8 "&!>!K"&"9"M"Ml\r"s"s i %;;IFFF  ( 26L6V V V4 1  5$($;$[$[\c$d$d! M  5 5E 5$11%2NOOO2259999 5 5 5 AaAADQQQQ**1e44444444 5( @ @J N 3 3J ? ? ? ?us19D++ E.5/E))E.rrc4tj|_t||_t |jtr%|j|j|_ ng|_ |j | dS)z(Handle general request processing error.N) r#FAILEDrnrAr isinstancerQr!get_active_request_static_outputsgenerated_tokensrrWto_generation_output)r4rrs r5rz.ContinuousBatchProcessor._handle_request_error[s$+ %jj  e& , , (%)^%U%UV[Vf%g%gE " "%'E " ""5#=#=#?#?@@@@@r8cF|}|rdS|j}|D]}|j||jsdS|j|j|jj \}}}}|]|j dkrtd|j|j|jj \}}}}|]|sdS|jj r|j |j |tjt jrxtjdt'|dt'|jjdt'|jjd|d |d |j |j|||\}}|j||j||||jj d S) zPrepare tensors and metadata for the next model forward pass. Returns True if there are requests to process, False otherwise.FNrz>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)r4 hard_stoppingcancelled_statesrrrrrs r5prepare_next_batchz+ContinuousBatchProcessor.prepare_next_batchhs 3355  5>BBDD% B BE  # : :5 A A A A~2244 5NR^MjMj  !4:#7N N J/{ '&7799Q>>"#cdddQUQ_QnQn%tz';RR N 3\;  '! 5 > * 7  # 4 4 6 6 6 ::;LMMM  w} - -  LZc"344ZZT^EdAeAeZZt~=>>ZZIUZZ&ZZ7;z7U7U7W7WZZ   %)$5$F$F|U`bv$w$w! k 55/"2!5%#)4 6   tr8c^|j\}}}d}g}|D]}|j}|jtjtjfvr9|jr |jr|dz }=td|jj d|j d|jr| dkrtj |_||}|||nd} |dz }||| } |j||j| r+|j|j d|j_|js|jtjkr'||S|jtjkr |j||j|r|j|gg} } |jjr\|jjj} d_fdt=| D}t?| |j j }| |z }t=|D]K}|}!}||_ |j"|L|dkr|D]$}#||jj$|<%|j%j |\}}| &|| &||jj\| rs|jj'}|tPj)*|n tW}|5|j,| | ddddS#1swxYwYdSdS) z0Update request states based on generated tokens.rrzTried to update z request z in sync mode.NFc(g|]}jd|S)z__child#)rQ).0i state_to_forks r5 z9ContinuousBatchProcessor.update_batch..s*eeeA-":GGAGGeeer8)-rprepare_batch_updaterrnr#FINISHEDPENDINGr has_new_tokenrnamerQ generated_lenDECODINGupdate_and_check_completionr!mark_shareable_blocks_as_completecomplete_blocksrfinish_requestblock_new_requests streamingr_r PREFILLINGrra_requests_to_forkpop num_childrenrangemincompute_max_num_forks!create_equivalent_initial_requestrforkr fork_requestextendrr@rstreamr copy_cache)r4r new_tokenslogprobscurrent_logits_indexpending_outputs future_statertokenlogprob is_finished copy_sourcecopy_destinationrnew_request_ids num_to_forknum_to_schedule_new_request_id child_statecopy_srccopy_dstr maybe_streamrs @r5 update_batchz%ContinuousBatchProcessor.update_batchsH262I2^2^2`2`/:x - b bL &E| 6 8MNNN*#12,1,"#rel6G#r#rRWRb#r#r#rsss) b&&((A--#0#9EL"#78N11%2BCCC8=DN5?Ielm6L&L&L#**5+E+E+G+GHHH!999 <   , ,_ = = =)+B% n. . N<@@BBM(5L)*M &eeeeQVWcQdQdeeeOlDJ,L,L]Me,f,fggK*[8O?++ @ @!0!4!4!6!6+MMOO )7 &22;????a"1 d dANASASTbAcAc.~>>!%!8!89QSb!c!c Hh   x ( ( (  # #H - - -/n. .4  E"4CN@N@Z5:,,^<<<`k`m`mL E E %%k3CDDD E E E E E E E E E E E E E E E E E E  E Es7N  N$'N$c4|jS)z2Check if there are any active or waiting requests.)rrrLs r5rz-ContinuousBatchProcessor.has_pending_requestss~22444r8c|jd}|D]A}|||j|j|jjBdS)z&Handle errors during batch processing.rN)rrrrrrrQ)r4rfailed_future_statesr s r5handle_batch_errorz+ContinuousBatchProcessor.handle_batch_errorsq#6KKMMaP0 I IL  & &ul.@ A A A N ) ),*<*G H H H H I Ir8ct|jj}|D]7}||||j|j8|jt|jj D]7}|jj |}|||8|jj dS)z.Fail all active requests with the given error.N)rdrrvaluesrrrQrfree_all_waiting_cpu_cachesrkeysrwaiting_requests_orderrm)r4rrequestsrreq_ids r5fail_all_requestsz*ContinuousBatchProcessor.fail_all_requestss6==??@@ < :??AABB 5 5FN377??E  & &ue 4 4 4 4 -3355555r8modelc|j|jj}|j|||jdS)z!Perform a single generation step.)rN)rrrr compute_batchretrieve_device_outputs)r4r' batch_datas r5_generation_stepz)ContinuousBatchProcessor._generation_steps[ ,==$J[Jf=gg  ''z::: 7799999r8c:|j|dS)aGPre-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)r4r's r5r.zContinuousBatchProcessor.warmup s!   '''''r8rb)(r<r=r>rrr?r'rrrrrrErFrCrfr@r.r-rrMrArrrboolrrr"rrrrrr&no_gradr Moduler,r.r3r8r5rrs~-0JJJJJ\ "\ !\ , \ %= \ ? \ [4'\ kD(\ $\ #9\ l\ [\ \ .\  \ \ \ \ | #    %%%% ''''%%%%%N A9 A\ Ad A A A A>D>>>>@KEKEKEKEZ5d5555III6y6T6666"U]__ :bi :D : : :_ :U]__(BI($(((_(((r8rcXeZdZdZ d.dededededzddf dZdeddfd Z d/d Z de fd Z d/d Z d0de dedzde de ddf dZd.dededzddfdZd/dZ d1deededzdedzde de deeezdzdededzfdZ d2deeededzde de dedeef d Zdeddfd!Zd3dedzdedzdedzfd"Zd#Zdedeefd$Zded%eddfd&Ze j!d/d'Z"d/d(Z#de$fd)Z%d*e&d+e$dzddfd,Z'd*e&d+e$dzddfd-Z(dS)4ContinuousBatchingManageraManager 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` Nr'r9rworkload_hintsr0ctj|j|_tj|_d|_t j|_t j |_ t|_ t|_d|_d|_d|_d|_d|_||||_||_t1|dd}||nd|_t5t1|jdd|j|_|jj|_|jr|jtA|j!||j"|j# |_$tK|jj&|||j$j' |_(|j(j)|_*dS) ayInitialize 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,rr4has_logit_processors)+rErFmax_queue_sizerr_request_counterrIrJ _request_lockEvent_has_new_requestsrfrrCrbatch_processor_generation_thread fatal_error warmed_up_original_attn_implswitch_to_cb_friendly_attnevalr'r9rr7rr:r is_tp_driverdisable_nccl_graph_mixingmaybe_warn_nccl_graph_mixingrr;r<r=rrr, do_processingrallow_block_sharing_use_prefix_sharing)r4r'r9rr4r7s r5rMz"ContinuousBatchingManager.__init__2s !;/I/XYYY.3kmm !&^--!*!2!2)?(@(@%)^^@D"&.2$(  ''...ZZ\\ "3&'8:PRVWWOPP#=#T(B(^   +M:$0)!%!5!C + + + '$(#B#V   r8cv|jj}|}t|}d|v}|so|sm|jrft drd}nt drd}nd}d}|#d |}t j|d |d |d nt j|d d|vr!|d|||_ dSdS)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$warningr6rH)r4r'original_attn_impl target_implemis_flashis_pagedversionmsgs r5rIz4ContinuousBatchingManager.switch_to_cb_friendly_attnwsD#\>* 0S`aaa}, n nU-G n(TBBB *tDDD RC" <7 < < tt,>ttMttt #lllmmm = ( (  ) )*B=*B*B C C C'9D $ $ $ ) (r8c|j||_|j|jd|_dS)zPre-capture CUDA graphs for varlen and decode paths by running dummy batches. Initializes the batch processor if not already done.NT)rD_create_batch_processorr.r'rGrLs r5r.z ContinuousBatchingManager.warmupsE   '#'#?#?#A#AD  ##DJ///r8cF|jduo|jS)zUReturns True if the background generation thread has been started and is still alive.N)rEis_aliverLs r5 is_runningz$ContinuousBatchingManager.is_runnings$&d2Yt7N7W7W7Y7YYr8c |rtjddS|jd|_t j|j|_ |j dS)z'Start the background generation thread.z"Manager thread is already running.N)target) rbr$rWrrmrFrIThread_run_generation_looprEstartrLs r5rgzContinuousBatchingManager.starts{ ??    N? @ @ @4 %++---"+"2$:S"T"T"T %%'''''r8TFblocktimeoutkeep_for_next_session hard_stopc|jtjd|jd}|r|dz }tj|dSt }|r t jn t j}|j ||j j |r| |||sd|_n tj d||j_|j&|j|jd|_t%jt(jr t(jdSdS)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.)rDr$rWrErrfrrrqrrvrrojoinrur'#_cached_continuous_batching_managerrHr6rrr@rrr)r4rhrirjrkr]stop_trigger_time stop_statuss r5stopzContinuousBatchingManager.stops\   ' NC D D D  " *(C$ YXX N3   4)NN:Cn,66I_In  %22;@W@cddd  2 II' 1 1 1% B#'D  KT U U U=ADJ :  # / J . .t/G H H H'+D $ : " " $ $ % J " " $ $ $ $ $ % %r8roc|jdS|j||jrtjd|ddSt }tjd||z ddd|_dS)zeWait for the background thread to finish. Wait can be capped using the timeout argument (in seconds).Nriz3Generation thread did not exit after join timeout (z).z+Background generation thread stopped after .2fs.)rErmrar$rWrru)r4roriends r5rmzContinuousBatchingManager.joins  " *4 $$W$555  " + + - - + N\QX\\\ ] ] ] ] ]..C KecL]F]eeee f f f&*D # # #r8c|r|dd|jdS)zTerminate the manager and release distributed resources. Safe to call multiple times. After calling this, the manager cannot be restarted.TF)rhrjN)rbrqrdestroy_cpu_comm_grouprLs r5destroyz!ContinuousBatchingManager.destroysG ??   ? IIDI > > > 6688888r8 input_idsrQmax_new_tokensrrecord_timestamps eos_token_idrc N|jsdS|jjtjkrD|dddddz|ddddz}t jd|ddS|9|j5d |j}|xjdz c_dddn #1swxYwY| |j j n|}| |j j n|}t|t||jdz ||||| } |j| d d |j|S)aAdd 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, ..., rz0Background thread is stopping. Request with ids z will be dropped.req_)rQinitial_tokensrr|r{r}rrT rhri)rKrr~rfrqr$rWrAr@r9r{r}r"rdr7rrSrCset) r4rzrQr{rr|r}rpreviewrs r5 add_requestz%ContinuousBatchingManager.add_requests2  4  ( 59O9^ ^ ^"2A2("- 9y~>^l>J>Rt-::Xd ! ??2Q6/)%#9     U$;;; ""$$$s1BBBinputsc t|}|j5dt|j|j|zD}|xj|z c_dddn #1swxYwYt t ||}|jrt|dd}|jj } | |j j j n| } | dn| } |D]\} } |j d| | |||| d||S) zgUtility function to batch `add_request` and return their IDs. Check its documentation for more details.cg|]}d|S)rr3)rrs r5rz:ContinuousBatchingManager.add_requests..@srrr!:!::rrrr8Nc|dS)Nrr3)xs r5z8ContinuousBatchingManager.add_requests..Es !A$r8T)keyreverser)rzrQr{rr|r}r3) rrArr@rdziprPsortedr9r}r'r,r) r4rr{rr|r num_requests request_idsids_and_inputsr}rQrzs r5 add_requestsz&ContinuousBatchingManager.add_requests4su6{{   2 2rruT5JDLadpLp/q/qrrrK  ! !\ 1 ! ! 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2c+v6677  # V#NPTUUUN-: 9E9Mtz(55S_ )1rr| %3   !J D  #%-#"3)   )    s8AA"Ac~|jr5|j||jdSdS)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)rKrrSrCr)r4rQs r5cancel_requestz(ContinuousBatchingManager.cancel_requestWsK   )   ! !* - - -  " & & ( ( ( ( ( ) )r8c |j |jjrdS |jjd|}|,|j|kr!|jj|dS|S#tj$rYdSwxYw)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) rErrGemptyrPrQrSrEEmpty)r4rQriresults r5 get_resultz$ContinuousBatchingManager.get_result`s  " *t/A/N/T/T/V/V *4 '488tW8UUF%&*;z*I*I"/33F;;;tM{   44 sA A:8A::B  B c#K|jW|jr@|d}||V|j|j._auto_cleanups HV   !!## M'-MM&6:::tLLLMMMMMMMMMMMMMMMMMM M Ms!AA"AN)asyncioget_running_looprrKrH)r4rQrUrVrs``` r5register_result_handlerz1ContinuousBatchingManager.register_result_handlers')) M M M M M M M   % S S>KT=RD  .z : S S S S S S S S S S S S S S S S S SsA  A A cd} |}||_d|_|jrG|st d||xjdz c_ |}|r9|||xjdz c_n~|jj tj krnd|jj tj kr| sn5|jd|jt#|jt&r-d|jjz |j_|t d|jj }|||nF#t,$r9}t/jd |d |||Yd}~nd}~wwxYw|jt/jd dS#|jt/jd wxYw) z6Main processing loop running in the background thread.Nrz$Failed to bootstrap the first batch.rTrrszdGeneration 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.)r_rD current_batchrrrr,rrrzrfrrrqrrCwaitrmrrr current_pair_fail_all_remaining_requestsrr$r_handle_critical_errorryru)r4rDrequests_availablerrs r5rfz.ContinuousBatchingManager._run_generation_loops8 _"::<XYY /BCoFhFuBu2?,,...!yTMjMtyyE  - -e_ E E E E < < < L9a99D I I I I  ' '? ; ; ; ; ; ; ; ; <  ) 9 9 ; ; ; K] ^ ^ ^ ^ ^  ) 9 9 ; ; ; K] ^ ^ ^ ^s0F8F=<H2= H/G;6H2;HH22/I!cp|jtd|j|jdS)z=Perform a single generation step. This is mostly cuda graphedNzNTried to perform a generation step before the batch processor was initialized.)rDrr,r'rLs r5r,z*ContinuousBatchingManager._generation_steps9   'opp p --dj99999r8cdt|dd}t|tr||St |jj|j|jj|j t|jdi|jj }|j |_ t|j|j|j|j |jdkr d|j_|jj}t'j|d}|t+jd|dt.}|||jj|jj }t||jj|j|j|j|jr|jnd|jr|jnd|j|j |jj|jj ||j }|S) ziCreate a new batch processor. If an already initialized batch processor exists, it is reset and returned.rDNtp_plan)r,rr.rrr-)r num_blocksruse_prefix_sharingrz Scheduler 'z ' not found. Defaulting to FIFO.)r safety_marginmax_requests_per_batch) rr,r9rrrrrrrrrr)!rrrrrr'r,rr.rr-rrPrrrnum_sliding_attention_groupsmax_blocks_per_requestscheduler_typer%rPr$rWr&rrr9rrKrrrr)r4rDpaged_attention_cacher scheduler_clsrs r5r_z1ContinuousBatchingManager._create_batch_processors"$(94@@ o'? @ @ #  ! ! # # #" "!4:$'+'F:$#6DJ 266*" ! ! ! $9#K -5,72C#7      ! = A AEFD + B8G)-ndCC  NYYYY Z Z Z)M!M'9G#'#B#Y   3':$"4'+'F 0,0,=G((4.2.?I**T,%)%B* (#6   r8rrDc||_|jtj|jj|jdtj|||dS)z:Handle critical errors that terminate the generation loop.)rnrorN) rFrrvrfrrrrorr)r4rrDs r5rz0ContinuousBatchingManager._handle_critical_errorsu! %22)3AXAd 3   33A7M7WXXX ))%AAAAAr8c |j}||||2#tj$rYnwxYw|||dSdS)zCFail all remaining requests in the input queue and active requests.TN)r get_nowaitrrErr&)r4rrDreq_datas r5rz6ContinuousBatchingManager._fail_all_remaining_requests,s  K+6688".#99%JJJ K{    D   &  - -e 4 4 4 4 4 ' &s35AAr2rb)TNFF)NNFFN)NFF)NN))r<r=r>rcr+rrr(rMrIr.r/rbrgfloatrqrmryrdrrAr rrrr!rrrrrrr@r0rfr,rr_rrrr3r8r5r3r3)s&04 CWCW#CW,CW%= CW &, CW  CWCWCWCWJ :0D : : : : :DZDZZZZ(((( $&+ 0%0%0%0% $ 0%  0%  0%0%0%0%d + +e +edl +d + + + +9999"&%)"'/37797$J7d 7  7  7DIo,7#&7 t7777x&*"' !!T#Y!d ! !  ! #& ! c!!!!F))))))  S4Z  YilpYp     # )z1ContinuousMixin.generate_batch..s(!I!IY#i..!I!I!I!I!I!Ir8r)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)r9rrhrirr.r4zSolving z requestsrequest)totaldisabledescunit)rr{r|rsz*Generation thread terminated unexpectedly.zCReturning results of generate_batch despite unexpected termination.zError during batch generation: rz Requests z not found in results.)r$getEffectiveLevelrrrWr9r7rrr{r(rtrr r rrrQrupdaterbrprintrrPr_)r4rr9rr|rrr.kwargsgen_cfgr7rr{r4 manager_cm logging_cmpbar_cmresultsfinished_countrpbarrrr%rreordered_results missing_keyss r5generate_batchzContinuousMixin.generate_batchsj4 I  # % % 6 6 NZ [ [ [ L->,E$((K\?F?[?gw;;mn6{{%99  $4d;;3A3I//~'!!I!I&!I!I!III3A3MST%      Nv   ==/'A1)>  +F844 %%3L333      S7J S S S4 S%22!.Te3 %|33$///::F !'!2!--//+.4GFO*a/N KKNNN$//11 %QRRRcddd%|33 S S S BqBBTRRRRRRRRR S' S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S S. ! , ,F[[((F!,2!&))##F++++  K LI\III J J J  sH>H'"H$B/GH H G< 7H<H H H'H H'H H' H>'H+ +H>.H+ /H>>II)NNNrb)NTNNFTN)NNFTFT)r<r=r>rcrr?r@r0rr(r3rrrr/rrrrdrdictrAr!rr3r8r5rr;s#('''U]__6:FJ/3 8 8 +d28 %=t$C8 &, 8 # 8 8 8 _8 tAAAAU]__6: $FJ#(/3#"#"+d2#"#" #" %=t$C #" ! #"#"&,#" , -#"#"#"_^#"LU]__6:FJ"'!#(m!m!T#Ym!,d2m!%=t$C m!  m!  m!!m!m! c## $m!m!m!_m!m!m!r8r)FrrrErIabcrcollections.abcrr contextlibrrtimertypingr r@r r tqdm.contrib.loggingr configuration_utilsrgeneration.configuration_utilsrr utils.genericrutils.import_utilsrr utils.loggingrlogits_processrrrcb_logits_processorsr distributedrinitializationrr input_outputsrrrrrr r$r!r"r#r$rr%r&r'utilsr(r)r1r+rCrfrr3rr3r8r5rsP ////////22222222 666666333333XXXXXXXX999999VVVVVVVV$$$$$$000000&&&&&&GGGGGG******eeeeeeeeLLLLLLLL%%%%%%111111KKKKKKKKKKKKBBBBBBBBBB--------,      29    ,2,2,2,2,2,2,2,2^77777777vg(g(g(g(g(g(g(g(V O5O5O5O5O5O5O5O5dd!d!d!d!d!d!d!d!d!d!r8