o ۾iE, @sldZddlZddlZddlmZmZmZmZddlmZm Z m Z ddl m Z m Z ddlmZddlmZmZddlmZdd lmZed Zed ZGd d d ZdefddZ d&de eefde dBde ee effddZdede fddZdedefddZerdeedffddZd eedfdee e!efdffd!d"Z"d#eedfde#efd$d%Z$dS)'ze Contains helpers related to asynchronous code. This is similar in concept to the `asyncio` module. N)FIRST_COMPLETEDAbstractEventLoopFutureTask)AsyncGenerator AwaitableCallable)ExecutorThreadPoolExecutor)partial) TYPE_CHECKINGTypeVar) BatchEncoding) ParamSpecPTc @seZdZdZ  d dededdfdd Zdefd d Zde efd d Z de fddZ de jdede jee eefee eefBfddZde jdefddZde jfddZde dedefddZddZdS)!AsyncMicrobatchTokenizerzAsynchronous tokenizer with micro-batching. Pulls pending encode/decode requests from a queue and batches them up to reduce overhead. A single-thread ThreadPoolExecutor is used so the event loop stays responsive. Mb`?max_batch_sizebatch_wait_timeout_sreturnNcCs8||_||_||_t|_i|_g|_tdd|_ dS)N) max_workers) tokenizerrrasyncioget_running_loop_loop_queues_batcher_tasksr _executor)selfrrrr"J/home/ubuntu/.local/lib/python3.10/site-packages/vllm/utils/async_utils.py__init__ s z!AsyncMicrobatchTokenizer.__init__csF|j}|d|}||j|}||||fIdH|IdHS)Nencoder create_future _queue_key _get_queueput)r!promptkwargs result_futurekeyqueuer"r"r#__call__5s    z!AsyncMicrobatchTokenizer.__call__cs||fi|IdHjSN) input_ids)r!r+r,r"r"r#r%<szAsyncMicrobatchTokenizer.encodecsD|j}|d|}||j|}|||fIdH|IdHS)Ndecoder&)r! token_idsr,r-r.r/r"r"r#r3?s    zAsyncMicrobatchTokenizer.decodeloopr.cCs|j|}|durDt|j|<}|ddkr&|ddk}|||}n|ddks6Jd|dd||}|j|||S) zkGet the request queue for the given operation key, creating a new queue and batcher task if needed.Nrr%rotherr3zUnknown operation type: .) rgetrQueue_batch_encode_loop_batch_decode_looprappend create_task)r!r5r.r/ can_batchcoror"r"r#r)Gs     z#AsyncMicrobatchTokenizer._get_queuer/r>c s |IdH\}}}|g}|g}|g}jj} t|jkrc| j} | dkr0n3z!t|| IdH\}}}|||||sP||Wn tj y[Ynwt|jks$zf|rt|dkrt j |fi|} j j | IdH} t|D]\} | sfdd| D}| t|qn'||ffdd }j j |IdH} t|| D] \} }| s| |qWn ty}z|D] } | s| |qWYd}~nd}~wwq) z.Batch incoming encode requests for efficiency.TNrrcsi|] \}}||qSr"r").0kv)ir"r# {sz?AsyncMicrobatchTokenizer._batch_encode_loop..csfddt||DS)Ncs"g|] \}}j|fi|qSr")r)r@pkwr!r"r# ~szQAsyncMicrobatchTokenizer._batch_encode_loop....)zip)promptsr,rGr"r#~s z=AsyncMicrobatchTokenizer._batch_encode_loop..)r8rtimerlenrrwait_forr< TimeoutErrorr rrun_in_executorr enumeratedoneitems set_resultrrI Exception set_exception)r!r/r>r+r,r-rJ kwargs_listresult_futuresdeadlinetimeoutbatch_encode_fnresultsfutdata encode_fnreser")rCr!r#r:Xsj        z+AsyncMicrobatchTokenizer._batch_encode_loopc s8 |IdH\}}|g}|g}|j|j}t||jkrW||j}|dkr,n+zt||IdH\}}||||Wn tj yOYnwt||jks z#|j |j |j j |IdH}t||D] \} } | sx| | qkWn ty} z|D] } | s| | qWYd} ~ nd} ~ wwq)z.Batch incoming decode requests for efficiency.TNr)r8rrLrrMrrrNr<rOrPr r batch_decoderIrRrTrUrV) r!r/r4r-token_ids_listrXrYrZr\r]r`rar"r"r#r;sL       z+AsyncMicrobatchTokenizer._batch_decode_loopopr,cCst|dkrdS|dd}|dd}|d}|sd|dd fSt|jd d }|d us2|d ur8||kr8d|dd fSd S) a Return a normalized key describing operation + kwargs. - `add_special_tokens`: {True/False} - `truncation`: {True/False} - If `truncation` is False (`max_length` is None), returns a key for a can_batch queue. - If `truncation` is True and `max_length` is None or equals `tokenizer.model_max_length`, returns a key for a can_batch queue. - Otherwise, returns a key for a cannot_batch queue. Examples: - Decode: ("decode",) - Encode typical: ("encode", add_special_tokens, bool_truncation, max_length_label) - Fallback: ("encode", "other") r3)r3add_special_tokensT truncationF max_lengthr%Nmodel_max_length model_max)r%r6)r8getattrr)r!rdr,rerfrgrir"r"r#r(s     z#AsyncMicrobatchTokenizer._queue_keycsNt|ddr!t|dd}r#|s%fdd}||dSdSdSdS)NrrcsD]}|qdSr1)canceltasktasksr"r# cancel_taskss z6AsyncMicrobatchTokenizer.__del__..cancel_tasks)rj is_closedcall_soon_threadsafe)r!r5rpr"rnr#__del__s  z AsyncMicrobatchTokenizer.__del__)rr)__name__ __module__ __qualname____doc__intfloatr$rr0listr%strr3rrtupler9dictrr)boolr:r;r(rsr"r"r"r#rs2  " 5" #rrmcCs(|r|st||jdSdSdSr1)rR run_in_loopget_looprkrlr"r"r#cancel_task_threadsafes rfuncexecutorrcs(dtjdtjdttffdd }|S)z Take a blocking function, and run it on in an executor thread. This function prevents the blocking function from blocking the asyncio event loop. The code in this function needs to be thread safe. argsr,rcs,t}tg|Ri|}|j|dS)Nrr)rget_event_loopr rP)rr,r5p_funcrr"r#_async_wrappersz"make_async.._async_wrapper)rrr,rr)rrrr"rr# make_asyncs$ rr5functioncGs6t|r ||dS|s|j|g|RdSdSr1)in_looprqrr)r5rrr"r"r#rs  r event_loopcCs$zt|kWStyYdSw)NF)rr RuntimeError)rr"r"r#rs  ritcCs|Sr1) __anext__)rr"r"r#anext sr iteratorsc st|dkr|d2z 3dHW}d|fVq 6dStfddt|D}zl|rhtj|tdIdH\}}|D])}||}z|IdH}|\}}||t |<||fVWq<t yeYq.) return_when)rMrrrQwaitkeysrpopr=rStopAsyncIterationrS contextlibsuppress BaseExceptionrkaclose) ritemawaitsrR_dpairrCrfr"rr#merge_async_iteratorssH       riteratorcs(g}|2z 3dHW}||q6|S)z6Collect all items from an async generator into a list.N)r<)rrSrr"r"r#collect_from_async_generator4s  rr1)%rwrrrrrrcollections.abcrrrconcurrent.futuresr r functoolsr typingr r $transformers.tokenization_utils_basertyping_extensionsrrrrrrrr~rrr|rxrrzrr"r"r"r#sB   I    "&