o .j@sUddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z m Z mZmZmZmZddlmZddlmZejrdd lmZdd lmZdd lmZdd lmZdd l m!Z!ej"eee!fZ#de$d<ej"eefZ%de$d<ej&de%dZ'ne&dZ'e Gdddej(e'Z)d,ddZ*e Gdddej(e'Z+dZ,de$d<ej-dkrej.Z/d-d d!Z0nd-d"d!Z0d.d&d'Z1Gd(d)d)ej(e'Z2Gd*d+d+ej(e'Z3dS)/) annotationsN) dataclassfield) AtomicTrafficOverwhelmedTraffic TrafficStateUnavailableTraffictraffic_state_ofItemPlaceholder)timeout) GroupedDict)AsyncHTTPConnection)AsyncConnectionPool)AsyncHTTPResponse)ResponsePromise)PoolKeyztyping.TypeAliasMappableTrafficManageableTrafficT)boundc@s.eZdZUdZded<ded<dZded<dS) ActiveCursorz_Duplicated from sync part for typing reasons. 'T' in TrafficPolice sync is bound to sync types!intobj_idr conn_or_pooldepthN)__name__ __module__ __qualname____doc____annotations__rr#r#^/home/thesage/.local/lib/python3.10/site-packages/urllib3_future/util/_async/traffic_police.pyr)s rreturnasyncio.Task[typing.Any]cCst}|dur td|S)Nz>Attempt to call function outside of (async) running event loop)asyncio current_task RuntimeError)tr#r#r$_current_task_or_die2s r+c@steZdZUdZeedZded<eej dZ ded<dZ ded <dZ d ed <dZ ded <ejejejfZd ed<dS)AsyncPendingSignalzPA task may want to block on a specific signal to avoid consuming CPU needlessly.)default_factoryr& owner_taskz asyncio.EventeventNT | Nonetarget_conn_or_pool int | None target_obj_idrztuple[TrafficState, ...]states)rrr r!rr+r.r"r'Eventr/r1r3rrIDLEUSED SATURATEDr4r#r#r#r$r,;s    r,g-C6?float_COALESCE_DELAYwin32NonecCs(tt}t|kr t|ks dSdSN) _perf_counterr:)deadliner#r#r$_kernel_recv_coalesce^s  r@cCsttdSr=)timesleepr:r#r#r#r$r@hsr4rboolcGstdd|DS)uInfer whether the caller wants to WRITE (True) or READ (False) based on the traffic states being queried. Querying exclusively for states with remaining request capacity (IDLE, USED) implies a WRITE intent — looking for a connection to send a new request on. Querying for states with open streams that may be at capacity (SATURATED) implies a READ intent — waiting to consume responses. css|]}|tjkVqdSr=)rr8).0stater#r#r$ vsz#_intent_to_write..)all)r4r#r#r$_intent_to_writels rHc@sneZdZdZd#ddZd$dd Zd%d d Zd%d dZd&ddZ d'd(ddZ d)ddZ d%ddZ d%d d!Z d"S)* AsyncSignalszeThe signal mgr counterpart of our synchronous implementation but for monothreaded/goto label version.cursors/dict[asyncio.Task[typing.Any], ActiveCursor[T]]r%r<cCsLt|_t|_t|_|ji|_||_ t|_ d|_ d|_ dS)Ng) collectionsdeque_priority_signals_furthest_signalsr'r5_generic_waitersset_saturated_signals_cursors_writing_tasks_last_dispatch_end_ema_gap)selfrJr#r#r$__init__|s   zAsyncSignals.__init__cCs|jtdSr=)rTaddr+rWr#r#r$declare_writingszAsyncSignals.declare_writingrCcCs(t}||jvr|j||SdSNF)r+rTremove_refresh_generic_waitersrWr(r#r#r$terminate_writings   zAsyncSignals.terminate_writingcCsD|js|jr|jr|jdSdS|js |jdSdSNTF)rTrNrPis_setclearrQrZr#r#r$r^s    z%AsyncSignals._refresh_generic_waitersr1r0r4rAsyncPendingSignal[T]cGsnt||d}|durt||_|dur#t|}||jvr#t|j|<t|r/|j||S|j ||S)N)r4r1) r,idr3rRr'r5rHrN appendleftrOappend)rWr1r4signalrr#r#r$registers    zAsyncSignals.registerFrhunblockcCsRz||jvr |j|n|j|Wn tyYnw|r'|jdSdSr=)rNr]rO ValueErrorr/rQ)rWrhrjr#r#r$ unregisters   zAsyncSignals.unregisterrcCs`|jrdSt|tjurdSt|}||jvr t|j|<|j|r)dS|j r.dSdSNFT) rPrbr rr8rerRr'r5rN)rWr1rr#r#r$should_queue_read_operations  z(AsyncSignals.should_queue_read_operationcCst}||jvr t|j|}d}|j|jvr#|j|j}|d}g}|jr4|j}|jd}|j D]}|j |jkrI|j| |q7|D]}|j |qL|j|=|r]dS|Srm) r+rSrrrRrQrNpopleftr/rOr3rgr])rWr(cursorany_broadcast_push conditionawoken_signals awoken_signalrhr#r#r$kills2          zAsyncSignals.killcCst}||jvr t|j|}t|j}t|jddsDt}|jdkr?||j}|j d|d|_ t |j }|dkr?t t|_|t j u}|t ju}|t ju} d} |j|jvry|j|j} |t j urq| sp| d} n| ry| t|j} | r|r|j} |j| _|j|=||j| j<d|_||jvr|j||j| j| j|dS| r| r|s| Sd}|jD]}||jvrq|j |jkr|}nq|dur|j|_|j|=||j|j<d|_|j|j||dS| S)N_fast_recv_modeTrgffffff?g333333?Fr)!r+rSrr rgetattrrA perf_counterrUrVr:r@rr8r6rrRrbrQrcrCrNpopr.rrTr]rYr/r^rOr4r3)rWr(rp current_statenow natural_gap remainingcan_write_morecan_read_anythingcan_only_writerqr/anything_pending_writeearly_unpack_signallate_signal_electedrhr#r#r$next sx                          zAsyncSignals.nextN)rJrKr%r<r%r<r%rC)r1r0r4rr%rd)F)rhrdrjrCr%r<)r1rr%rC) rrr r!rXr[r`r^rirlrnrurr#r#r#r$rIys       %rIc@seZdZdZ   d_d`d d ZedaddZdbddZedcddZddddZ edcddZ edcddZ edcddZ ded d!Z dfd#d$Zdgdhd)d*Zdbd+d,Zdgdid-d.Zdjd0d1Zddd2dkd6d7Z dldmd:d;Z    dndod>d?Z dpdqdAdBZdrdCdDZej   dsdtdHdIZ  dudvdJdKZej    dwdxdMdNZdcdOdPZdbdQdRZdedSdTZdedUdVZ dcdWdXZ!dydZd[Z"dzd]d^Z#dS){AsyncTrafficPoliceaTask-safe extended-Queue implementation. This class is made to enforce the 'I will have order!' psychopath philosophy. Rational: Recent HTTP protocols handle concurrent streams, therefore it is not as flat as before, we need to answer the following problems: 1) we cannot just dispose of the oldest connection/pool, it may have a pending response in it. 2) we need a map, a dumb and simple GPS, to avoid wasting CPU resources searching for a response (promise resolution). - that would permit doing read(x) on one response and then read(x) on another without compromising the concerned connection. - multiplexed protocols can permit temporary locking of a single connection even if response isn't consumed (instead of locking through entire lifecycle single request). This program is (very) complex and need, for patches, at least both unit tests and integration tests passing. NFTmaxsizer2 concurrencyrCstrict_maxsizecCsZ||_||_||_||_i|_i|_ttd|_t|_ d|_ i|_ d|_ t |j |_dS)a :param maxsize: Maximum number of items that can be contained. :param concurrency: Whether to allow a single item to be used across multiple tasks. :param strict_maxsize: If True the scheduler does not increase maxsize for a temporary increase. )key_fnFN)rrr_original_maxsize _registry _containerrre_map _map_types _shutdownrSparentrI_signals)rWrrrr#r#r$rXs  zAsyncTrafficPolice.__init__r%ActiveCursor[T] | NonecCst}||jvr |j|SdSr=r+rSr_r#r#r$_cursors  zAsyncTrafficPolice._cursorr<cCs t}||jvr|j|=dSdSr=rr_r#r#r$ _unset_cursors  z AsyncTrafficPolice._unset_cursorcCs |jduSr=)rrZr#r#r$busys zAsyncTrafficPolice.busyrrcCs |j}|dur dS|jt|kSr\)rrre)rWr active_cursorr#r#r$is_heldszAsyncTrafficPolice.is_heldcCstdd|jDS)Ncs|] }t|tjuVqdSr=)r rr6rD_r#r#r$rFs z3AsyncTrafficPolice.bag_only_idle..)rGrvaluesrZr#r#r$ bag_only_idlesz AsyncTrafficPolice.bag_only_idlecCs"|jsdStdd|jDS)z@All manageable traffic is saturated. No more capacity available.Fcsrr=)r rr8rr#r#r$rFs  z8AsyncTrafficPolice.bag_only_saturated..)rrGrrZr#r#r$bag_only_saturateds z%AsyncTrafficPolice.bag_only_saturatedcCs2|jsdS|jD] }t|tjurdSq dSrm)rrr rr8)rWconnr#r#r$bag_not_saturatedsz$AsyncTrafficPolice.bag_not_saturatedrcC t|jSr=lenrrZr#r#r$__len__ zAsyncTrafficPolice.__len__valuecCsbt|}||jvr dSg}|jD]\}}t||kr!||q|D] }|j|=|j|=q$dSr=)rerritemsrgr)rWrr outdated_keyskeyvalr#r#r$ _map_clears    zAsyncTrafficPolice._map_clear traffic_typetypeblockr0c sd}|j|D]}|d7}|j|}t|}||jvr!|Sq |r&|s(dS|jdtjtj }|j IdH|j durAdS|j|j D]}|j||urRqH|j S| t|j|ddu} | sldS|j||dIdHS)zRFind the first available conn or pool that is linked to at least one traffic type.rrN)rr)rkeys_forrrerrrirr7r8r/waitrreleaser__iter___find_by) rWrreligible_object_counttkrrrhmkany_remaining_eligible_objectr#r#r$rs@     zAsyncTrafficPolice._find_bycs|j}|dur dS||j|j|j=|js&|jr%t dIdHn |j |j=| z |j IdHWdSt yDYdSw)zIn case there is no other way, a conn or pool may be unusable and should be destroyed. This make the scheduler forget about it.Nr)rrrrrrrrur'rBrrclose Exception)rWrr#r#r$ kill_cursor s"     zAsyncTrafficPolice.kill_cursorcs*d\}}|js dS|jD]\}}||jvr&t|tjur&||}}nq|dur]|dur]|||j|=|j|=|jtd|dIdHz | IdHWdSt y\YdSw|sht d|j d|j dtj}z |jIdHWntjy|j |w|j|dIdHdS)zWhen trying to fill the bag, arriving at the maxsize, we may want to remove an item. This method try its best to find the most appropriate idle item and removes it. NNNTimmediately_unavailablerzCannot select a disposable connection to ease the charge. This usually means that your pool sizing is insufficient, please increase your pool maxsize appropriately. Currently set at maxsize=zE. Usually you want as much as the number of active tasks/connections.r)rrrr rr6rputr rrrrrrir/rr'CancelledErrorrl_sacrifice_first_idle)rWreligible_obj_ideligible_conn_or_poolrrrhr#r#r$r8sV      z(AsyncTrafficPolice._sacrifice_first_idletyping.AsyncGenerator[T, None]c Cs|jrtdt}|jrjg}|jD]\}}t|tjur!q|||fq|D]@\}}||jvs;t|tjur   zAsyncTrafficPolice.iter_idle)rrtraffic_indicatorsrrc s|jr|IdH|jsd|_dS|jdur?t|j|jkr?t||jvr?|j|dIdH|jdur?|j|jj=| d}t }|j}zt|}||jv} | r||j vrgW|ret dIdHdSdS|dur|j|krttd|jd8_|jdkr|js|jrd}|jrd}n| ||j |<n | n||j|<|rt|||j|<t|t} | s|js|j|jdur| s||j |<d}|r|D]} || |qW|rt dIdHdSdS|rt dIdHww)NFrrz7You must release the previous connection prior to this.rT)rrrrrrerrrrr+rr'rBrrrrr`rrrS isinstancer r[memorize) rWrrrrshould_schedule_another_taskr(rrregistered_conn_or_poolis_placeholder indicatorr#r#r$rsv     +          zAsyncTrafficPolice.putnon_saturated_only not_idle_onlycs|jd||dIdHS)NF)rrr)get)rWrrr#r#r$ get_nowaits  zAsyncTrafficPolice.get_nowaitr float | Nonec sd}|jdur td|jr|jr-|jdur-|jt|jkr-|jtd|dIdHdSt }|jr|rYd\}}|j D]\}} t | t j urJq>|| }}|durX|j|=n-|sc|j\}}n#d\}}|j D]\}} t | t jurxql|| }}|dur|j|=|dur|durt|||j|<|jr||j|<|S|r|js|j|S|durT|rP|rt jt j f} n|rt jt jf} nt jt jt j f} |jjdg| R} zP|durz*t|d4IdH| jIdHWdIdHn 1IdHswYWnty} z td|d| d} ~ ww| jIdHWntjy8|j| w| jdurM|jtd|dIdHdS| jStd|S) Nz_One connection/pool active per task at a given time. Call release prior to calling this method.Trr)delayzNo connection available within z second(s)zNo connection available) rrrrrrrrr r+rr rr8popitemr6rrSrrr[r7ri ctx_expire_inr/r TimeoutErrorr r'rrlr) rWrr rrrr(rrrr4rher#r#r$rs          *    zAsyncTrafficPolice.gettraffic_indicatorcCs|j}|dur|durtd|dur!|dusJ|j|j}}nt||}}||jvr/dSt|trB||j|<t ||j |<dSt|}||j|<t ||j |<dS)Nz(No connection active on the current task) rrrrrerrtuplerrr)rWrrrrtraffic_indicator_idr#r#r$rys     zAsyncTrafficPolice.memorizecCsjt|tr|nt|}||jvrdS|j|=|j|=|jdur3z |j|WdSty2YdSwdSr=)rrrerrrforgetr )rWrrr#r#r$rs   zAsyncTrafficPolice.forgetMappableTraffic | Noneplaceholder_setGtyping.AsyncGenerator[typing.Callable[[T], typing.Awaitable[None]] | T]cs|durj||dIdH}|dur|VdSg|dur#||s6jtgRd|dIdHdd fd d }|VsOIdHdSdS) aWReserve a spot into the TrafficPolice instance while you construct your conn_or_pool. Creating a conn_or_pool may or may not take significant time, in order to avoid having many thread racing for TrafficPolice insert, we must have a way to instantly reserve a spot meanwhile we built what is required. N)rrTrFswappable_conn_or_poolrr%r<csJdj}|durj|j=j|gRdddIdHdS)NTFr)rrrrr)rrrW swap_maderr#r$ inner_swaps z5AsyncTrafficPolice.locate_or_hold..inner_swap)rrr%r<)locatergrr r)rWrrrrrr#rr$locate_or_holds4  z!AsyncTrafficPolice.locate_or_holdc s t|ts&t|tr|nt|}||jvrd\}}n|j|}t|}ntd|dur\|dur\t|ts\|jD]\}}t|dr[t|j t r[|j |IdHr[||}}nq||IdH}|r=t|}|j}|dur-|j|kr,tdn?t|||jt <|j s=|j |=n.|j |||dIdH}n"|jdura|j}|j d7_ |j|j}}n |j|||dIdH}|dur||durxtdtd|VWn tyd}wW| ot|d d} | r|IdHdS|rtd IdHdSdS| ot|d d} | r|IdHw|rtd IdHww) NTr)rr r)rr rzDNo connection matches the traffic indicator (promise, response, ...)zNo connection are availableF is_closedr)rrrrerrrrrSr+rrrrrrr rrwrrr'rB) rWrrr r clean_exitrrr withhold_connr#r#r$borrow<s|        zAsyncTrafficPolice.borrowcCsd|j}d}|dur0|jd8_|jdkr0|js,|j}|s*|j|j|j<||S||S)NFrr) rrrrrrrrr)rWrcursor_transferredr#r#r$rs   zAsyncTrafficPolice.releasec sg}|jD]}t|j|tjur||q|D]}|j|=qt|jt|kr-d|_|D] }|j|}z | IdHWn t yIYnw| |q/|j }|durb|j |vrd|dSdSdS)zShutdown traffic pool.TN)rr rr6rgrrrryrrrrrr)rWplanned_removalrrrr#r#r$rcs0        zAsyncTrafficPolice.clearcCrr=)rrrZr#r#r$qsizerzAsyncTrafficPolice.qsizecCrr=rrZr#r#r$rsizerzAsyncTrafficPolice.rsizecCs t|Sr=)rCrrZr#r#r$emptys zAsyncTrafficPolice.emptyMappableTraffic | typecsxt|tst|tr |nt|}||jvS|j}z||IdHduW|s.|jr/|SSS|s:|jr;|wwwr=)rrrrerrrr)rWrrpre_acquired_connr#r#r$rs     zAsyncTrafficPolice.beaconstrcCsH|j}| o|j}|rd}n|rd}nd}d|d|jd|dS)N SaturatedIdleUsedz)rrrr)rW is_saturatedis_idlestatusr#r#r$__repr__s zAsyncTrafficPolice.__repr__rm)rr2rrCrrC)r%rrr)rrr%rC)r%r)rrr%r<)T)rrrrCr%r0)rrCr%r<)r%r) rrrrrrCrrCr%r<)FF)rrCrrCr%r0)TNFF) rrCr rrrCrrCr%r0r=)rrrr0r%r<)rrr%r<ra)rrrrCrrCr%r)TN)rrrrCr rr%r0)NTNF) rrrrCr rrrCr%r)rrr%rC)r%r)$rrr r!rXpropertyrrrrrrrrrrrrrrrrrr contextlibasynccontextmanagerrrrrrcrrrrrr#r#r#r$ryst           .  :2Z   x A U L     r)r%r&r)r4rr%rC)4 __future__rr'rLrsysrAtyping dataclassesrrtraffic_policerrrr r r contrib.ssa._timeoutr r _collectionsr TYPE_CHECKING_async.connectionr_async.connectionpoolr_async.responserbackendr poolmanagerrUnionrr"rTypeVarrGenericrr+r,r:platformrxr>r@rHrIrr#r#r#r$sP