㚾jcddlZddlZddlZddlmZddlmZddlmZddl m Z eje Z GddZGd d ZGd d ZGd dZy)N)deque)asyn) sync_teardown) _fast_slicecleZdZdZd dZdefdZedefdZede fdZ edefdZ d Z y ) RunningAverageTrackerzTracks a running average of values over a sliding window. This is used to monitor read sizes and adaptively scale the prefetching strategy based on recent user behavior. c`tjd|t||_d|_y)zInitializes the tracker with a specific window size. Args: maxlen (int): The maximum number of historical values to keep. z2Initializing RunningAverageTracker with maxlen: %dmaxlenrN)loggerdebugr_history_sum)selfr s [/home/spazio/kingdom_calls/work5/transenv/lib/python3.12/site-packages/fsspec/prefetcher.py__init__zRunningAverageTracker.__init__s&  I6RV,  valuecb|dkr tdt|j|jjk(r"|xj|jdzc_|jj ||xj|z c_t jd||jy)zAdds a new value to the sliding window and updates the rolling sum. Args: value (int): The integer value to add to the history. rzDInternal error, RunningAverageTracker tried inserting negative valuez2RunningAverageTracker added value: %d, new sum: %dN) ValueErrorlenrr rappendr r)rrs raddzRunningAverageTracker.adds A:V  t}} !5!5 5 IIq) )I U# U  @% rreturncVt|j}|dk(ry|j|zS)zCalculates and returns the current running average. Returns: int: The integer average of the current history. ri)rrrrcounts raveragezRunningAverageTracker.average1s+DMM" A:yyE!!rczt|j}|dkrytt|jdkDS)z8Determines if the history contains distinct chunk sizes.Fr)rrsetrs r is_variablez!RunningAverageTracker.is_variable=s5DMM" 193t}}%&**rcN|js td|jdS)z-Returns the most recent entry in the history.zNo entry found in history)r RuntimeErrorrs r last_valuez RunningAverageTracker.last_valueFs&}}:; ;}}R  rcptjd|jjd|_y)z.Clears the history and resets the sum to zero.z'Clearing RunningAverageTracker history.rN)r rrclearrr's rr*zRunningAverageTracker.clearNs& >?  rN) ) __name__ __module__ __qualname____doc__rintrpropertyrboolr#r(r*rrr r sn   & " " "+T++!C!!rr ceZdZdZdZdZdZdZ ddeded e jd e jd d d e ddfdZ edefdZdZdZdefdZdZdeeeeffdZdZy)PrefetchProducera Background worker that fetches sequential blocks of data. This class handles the network requests. It spawns asynchronous tasks to fetch data ahead of the user's current reading position and places those task promises into a queue for the consumer. iPiiNsize concurrencyqueue wakeup_eventconsumerPrefetchConsumertracker orchestratorBackgroundPrefetcherc  tjd||| ||_||_||_||_||_||_||_tj||_ | |_ d|_ d|_t|_d|_y)a*Initializes the background producer. Args: fetcher (Callable): A coroutine function to fetch bytes from a remote source. size (int): Total size of the file being fetched. concurrency (int): Maximum number of concurrent fetch tasks. queue (asyncio.Queue): The shared queue to push download tasks into. wakeup_event (asyncio.Event): Event used to wake the producer from an idle state. consumer (PrefetchConsumer): The consumer reading the prefetched chunks. tracker (RunningAverageTracker): Tracker for history of read sizes. orchestrator (BackgroundPrefetcher): The parent object managing the operation. user_max_prefetch_size (int, optional): A hard limit for prefetch size overrides. zQInitializing PrefetchProducer: size=%d, concurrency=%d, user_max_prefetch_size=%srFN)r rfetcherr7r8r9r:r;r=weakrefproxyr>_user_max_prefetch_sizecurrent_offset is_stoppedr" _active_tasks_producer_task) rrAr7r8r9r:r;r=r>user_max_prefetch_sizes rrzPrefetchProducer.__init__ws2  _   "    & (   #MM,7'=$ U"rrc|jAt|jtd|jjz|j Std|jjz|j S)zCalculates the maximum prefetch size based on user intent or io size. Returns: int: The maximum number of bytes to prefetch ahead. r!)rDminmaxr=rMIN_PREFETCH_SIZEr's rmax_prefetch_sizez"PrefetchProducer.max_prefetch_sizesi  ' ' 3,,A ,,,d.D.DE 1t||+++T-C-CDDrctjdd|_|jj t j |j|_y)z|Starts the background producer loop. This clears any previous wakeup events and spawns the main loop task. zStarting PrefetchProducer loop.FN) r rrFr:r*asyncio create_task_looprHr's rstartzPrefetchProducer.startsC  67 !%11$**,?rcBKtjdt|jd|_|j j g}|jrO|jjs5|jj|j|j|jdt|jD|jjd}|jjs |jj!}t#|t$j&r0|jr |j)s|j+|dz }|jjs|dkDrtjd||rz(PrefetchProducer.stop..s TYY[D srrz5Cleared %d leftover items from the queue during stop.z8Waiting for %d cancelled tasks to finish their teardown.return_exceptionsN)r rrrGrFr:r"rHrWcancelrextendlistr*r9empty get_nowait isinstancerPTask cancelled exception QueueEmptygather)r tasks_to_wait cleared_itemsitems rstopzPrefetchProducer.stops  ?TEWEWAX     t':':'?'?'A    & & (  !4!4 5 !$"4"45    " **""$ zz,,.tW\\2  NN,NN$" **""$ 1  LLG   LLJM" ..-H4H H H !%%   Is>D H A)H5HAH$H%HHHHH new_offsetcKtjd||jd{||_|j y7w)zStops current tasks and restarts the background loop at a new byte offset. Args: new_offset (int): The new byte position to start prefetching from. z-Restarting PrefetchProducer at new offset: %dN)r rrjrErSrrks rrestartzPrefetchProducer.restarts<  DjQiik(  s*A A A cjKtjd |jso|jj d{|jj |jry|j d{|jsnyy7R7#tj$rtjdYyt$re}tjdd|_|jj||jj|d{7Yd}~yd}~wwxYww)zFThe main background loop that delegates calculations and spawns tasks.z.PrefetchProducer internal loop is now running.Nz$PrefetchProducer loop was cancelled.z6PrefetchProducer loop encountered an unexpected error.T)r rrFr:waitr*_process_prefetch_cyclerPCancelledError Exceptionrdr> set_errorr9putres rrRzPrefetchProducer._loops EF $oo'',,...!!'')??22444oo. 5%% A LL? @ $   U V"DO    ' ' ***..# # #  $swD3)BB*B-D3.BBBD3BB(D0D3D0 AD+ D#!D+&D3+D00D3c|jj}|jj}|jj}|jj }|j duxr||j kD}|r||jkDs|rtjdd}n$||jkrd}n||jz dz}|jjs|dk(r|}n|}t||z|j}|jj|z|jj kr-|jj |jjz }|r|} n||z|z} | dk(r|} ||| fS)z Evaluates current trackers and state to determine sizes. Returns: tuple: (prefetch_size, io_size, effective_prefetch_size) NzTLarge IO detected (variable > 64MB or > user max). Disabling background prefetching.rr)r=rr;sequential_streakr#r(rDVARIABLE_IO_THRESHOLDr rMIN_STREAKS_FOR_PREFETCHINGr9r_rKrNoffset target_offset) r avg_io_sizestreakr#last_read_sizeexceeds_user_maxprefetch_multiplierio_size prefetch_sizeeffective_prefetch_sizes r_calculate_prefetch_paramsz+PrefetchProducer._calculate_prefetch_paramsslll** 00ll.. 00  ( ( 4 ;d:::  K$*D*DD  LLf #$  d66 6"# "(4+K+K"Ka"O  ::   !4!9$G!G/'94;Q;QR ==  - /$--2M2M M MM77$--:N:NNM &3 #'4'?7&J #&!+*7'g'>>>rcK|j\}}}tjd|j|jj ||j sn|j|jj z |krG|j|jkr-|jj }|j|jz }||j|z z }||jk\r2||jk\r!tt|j||}nnt||}||kr |jjs||k(r|}nnz|jj}||jkr |j} n0t|jtd||jz|z} tjd|j|| t!j"|j%|j|| } |j&j)| | j+|j&j,|j.j1| d{|xj|z c_|j sA|j|jj z |kr|j|jkr-|j|jk\rtjdd|_yy7w)z1Executes a single cycle of enqueuing fetch tasks.zFProducer awake. Current offset: %d, User offset: %d, Prefetch size: %drz;Spawning fetch task. Offset: %d, Size: %d, Split Factor: %d) split_factorNz.Producer reached EOF. Exiting background loop.T)rr rrEr;r|rFr7MIN_CHUNK_SIZErKrLr=r#ryr{r8rPrQrArGradd_done_callbackdiscardr9ru) rrrr user_offsetspace_remainingprefetch_space_available actual_sizersfactor download_tasks rrqz(PrefetchProducer._process_prefetch_cycleDs  + + - 8 w 7  T    MM   $$t}}';';;}L##dii/--..K"ii$*=*==O'4##k1( $ 3 33+t/B/BB"%D//9?#K!'?; '+5LL,,/=@":K]]44F888**$$#d&6&66:QQ LLM##  $// T00+G TM    " "= 1  + +D,>,>,F,F G**../ / /   ; . o$$t}}';';;}L##dii/n   $)) + LLI J"DO , 0sIL!K?"A$L8LrV)r,r-r.r/rrMr{rzr0rPQueueEventr rr1rNrSrjrnrRtuplerrqr3rrr5r5Us%N*#$- $-#-# -# }} -# mm -#%-#'-#--#^ E3 E E@4"l  $*.?E#sC-,@.?`I#rr5ceZdZdZdej dej deddfdZde fd Z d Z d e d e d e efdZd e d efdZd e d dfdZy)r<zConsumes prefetched chunks from the queue and manages byte slicing. This class pulls data out of the shared queue and slices it into the exact byte sizes requested by the user. It also manages the local block buffer. r9r:r=r>r?ctjd||_||_||_t j ||_d|_d|_ d|_ d|_ d|_ y)aInitializes the consumer. Args: queue (asyncio.Queue): The shared queue containing fetch tasks. wakeup_event (asyncio.Event): Event used to wake the producer when more data is needed. tracker (RunningAverageTracker): Tracker for history of read sizes. orchestrator (BackgroundPrefetcher): The parent object managing the operation. zInitializing PrefetchConsumer.rrN) r rr9r:r=rBrCr>ryr|r}_current_block_current_block_idx)rr9r:r=r>s rrzPrefetchConsumer.__init__s_  56 ( #MM,7!" !"#rrkcvtjd|||_||_d|_d|_d|_y)zClears the buffer and resets the internal offset for a hard seek. Args: new_offset (int): The byte position the consumer is jumping to. zDConsumer executing hard seek to offset %d. Clearing internal buffer.rrN)r rr|r}ryrrrms rseekzPrefetchConsumer.seeks?  R  ! '!"!"#rcJtjdd|_d|_y)zADiscards the local byte buffer. Useful during shutdown or resets.z$Consumer local block buffer cleared.rrN)r rrrr's r clear_bufferzPrefetchConsumer.clear_buffers ;<!"#rr7 save_datarcK|dkrgSg}d}|j|z|_||kr6t|j|jz }d}|s|j j duxs |j j j}|r2|jjrtjd |S|jjr/tjd|jj|jjd{}t|t r3tj#d||j j%|| |d{} |xj&dz c_|j&t(j*k\r|j j,duxr-|j.j0|j j,kD} |j.j2xr'|j.j0t(j4kD} | s| sd}ntjd | |_d|_t|j}|js |S||z } t=| |}|r}|t|jk(r|jdk(r |j}n8t7j>t@|j|j|d{}|jC||xj|z c_||z }|xj|z c_|r|jj||kr6|S7=7#t6j8$rt $r7} tj;d |j j%| | d} ~ wwxYw7ŭw) zInternal method to advance the offset and optionally extract data. Handles queue exhaustion, producer wakeups, and streak tracking. rFNzConsumer reached EOF.z#Queue is empty. Waking up producer.z#Consumer retrieved an exception: %srTzfSuppressing proactive producer wakeup due to massive variable workload or exceeding user max prefetch.zConsumer caught an error.)"r|r}rrrr>producerrFr9r_r rr:r"getrarserrorrtryr5r{rNr=rr#rzrPrrrdrK to_threadrr)rr7rchunks processed availabletrigger_wakeupis_producer_stoppedrYblockris_massive_variablerwneededtakechunks r_advancezPrefetchConsumer._advancesc 19I ![[4/$D//043J3JJI"N%%..$6=((11<<$'4::+;+;+=LL!89T Q::##%LL!FG%%))+!ZZ^^--dI.LL!FM%%//5J%"&JE**a/*..+GGH!--??tKB $ 4 4"//AA!B) !LL44E $ 4 4.DD!E,!47G-1N"LL!L +0D'./D+ #D$7$7 8I&&* 'I%Fvy)D3t22338O8OST8T //E#*"3"3#T%8%8$:Q:QSW#E e$  # #t + #  I KK4 K!!%%'g$j I.'>-- $$%@A%%//2G s^D!N>#M $AN>,M&1M#2C;M&-B N>6N<7A%N>N>#M&&N92N44N99N>cK|dkry|j|dd{}|syt|dk(r|dStjdj|d{S7B7w)aPulls exactly 'size' bytes from the local block or the task queue. If the local block is exhausted, this will wait on the queue for the next available chunk of data. Args: size (int): The exact number of bytes to retrieve. Returns: bytes: The requested bytes. This may be shorter than 'size' if EOF is reached. Raises: Exception: Re-raises any exceptions encountered by the producer fetch tasks. rrTrNr)rrrPrjoin)rr7rs rconsumezPrefetchConsumer.consume*sh 19}}TT}:: v;! !9 &&sxx888;9s!A&A"=A&A$A&$A&NcFK|j|dd{y7w)z7Advances the consumer offset without allocating memory.FrN)r)rr7s rskipzPrefetchConsumer.skipFsmmDEm222s !!)r,r-r.r/rPrrr rr0rrr2r^bytesrrrr3rrr<r<s $}}$mm$' $ - $4$s$ $ a3a4aDKaF9#9%983s3t3rr<ceZdZdZdZ ddedefdZdZdZdZ d Z d e fd Z d efd Z dZdZdedzdedzdefdZdedzdedzdefdZdZddedzfdZy)r?a&Orchestrator that manages reading behavior and coordinates background work. This acts as the main public interface for the file reader. It tracks the user's reading history, routes seek operations, and links the producer's network tasks with the consumer's data slicing logic. Nr7r8ctjd|||_|__&dkr!tj dt d|_d_d_ d_ td_ d_ d_d_d_d_fd  t%j&}|jurjnNj&fd }t+j,j|n||_n t)d tjd y#t($rd}YwxYw) aInitializes the background prefetcher. Args: fetcher (Callable): A coroutine of the form `f(start, end)` which gets bytes from the remote. size (int): Total byte size of the file being read. concurrency (int): Number of concurrent network requests to use for large chunks. max_prefetch_size (int, optional): Maximum bytes to prefetch ahead of the current user offset. loop (asyncio.AbstractEventLoop, optional): The event loop to attach the prefetcher to. If executing synchronously, this should be the fsspec background loop. If executing asynchronously (asynchronous=True), this should be None so it can automatically inherit the user's currently running event loop. Raises: ValueError: If max_prefetch_size is provided but is not a positive integer. zJStarting BackgroundPrefetcher. Size: %d, Concurrency: %d, Max Prefetch: %sNrz&Invalid max_prefetch_size provided: %szKmax_prefetch_size should be a positive integer to use adaptive prefetching!Fr+r c tj_tj_tj _tjjj_ tjjjjjj _ jjy)N)r9r:r=r>) rAr7r8r9r:r;r=r>rI)rPrr9rr:Lock _async_lockr< read_trackerr;r5r7r8rrS)rArNrsr_startz-BackgroundPrefetcher.__init__.._starts DJ ' D &||~D ,jj!..))! DM-YY ,,jj!..))!'8 DM MM   !rcKywrVr3)rsr_start_wrapperz5BackgroundPrefetcher.__init__.._start_wrappers s zNo event loop foundz-BackgroundPrefetcher initialization complete.)r rr7r8rNrrloop_errorrFrr rr9r:rr;rrPget_running_loopr& fsspec_asynsync) rrAr7r8rNr current_looprrs `` ` @rrzBackgroundPrefetcher.__init__UsQ$  X      &!2  (->!-C LLACT U]   1<     "4 "335L 499 $)> H YY "    TYY 7  %$DI H45 5 DE) L s2E EEc|S)zContext manager entry point.r3r's r __enter__zBackgroundPrefetcher.__enter__s rc$|jy)zEContext manager exit point. Ensures the prefetcher is cleanly closed.N)closerexc_typeexc_valexc_tbs r__exit__zBackgroundPrefetcher.__exit__s  rcK|Sw)z"Async context manager entry point.r3r's r __aenter__zBackgroundPrefetcher.__aenter__s  sc@K|jd{y7w)zKAsync context manager exit point. Ensures the prefetcher is cleanly closed.N)aclosers r __aexit__zBackgroundPrefetcher.__aexit__skkms rwc>tjd|||_y)Nz2Global error state set in BackgroundPrefetcher: %s)r rrrvs rrtzBackgroundPrefetcher.set_errors I1M rrkcKtjd|d|_|jj |d{|j j ||jjy7:w)Nz8Handling seek request. Restarting producer at offset: %d) r rrrrnr;rrr*rms r_restart_producerz&BackgroundPrefetcher._restart_producersa F   mm##J/// :& ! 0sRecovering from error state. Restarting producer at offset: %dz1Soft seek detected. Skipping ahead from %d to %d.z\Local seek performed. User offset moved from %d to %d. Adjusting buffer index from %d to %d.z5Hard seek detected. Moving user offset from %d to %d.z*Completed _async_fetch. Returned %d bytes.z+Exception raised during asynchronous fetch.)rrFr&r rrinforrr;r|rrrErr}rrrrrPrrrsrdrj)rrSend block_offset skip_amountrequested_sizerrws r _async_fetchz!BackgroundPrefetcher._async_fetchs##B B A ??&'KLL H%QTU;;KKX(-D$00777d... ,,t}}/O/OO!''%O4==3O3OO O ,,! ',d.>.>&> "mm00===+0(%E1E1EE D ,,! MM<<!L0 ,< DkB t))     !NO ==)A)A--,,...  {B B B sM2J=M2MA+K KB=K KC"K %K&AK (K)AK + M27K 8M2K K K K  M2 MK'' M3AMMMMMM/#M& $M/+M2cK|j4d{tjd|jr8|jjs"|jj d{|j jtjddddd{y77F7 #1d{7swYyxYww)aAsynchronous teardown logic protected by the async lock. Held for the duration of teardown so a concurrent in-flight ``_async_fetch`` cannot observe a stopped producer/empty queue as EOF and silently return a truncated read. Nz4Signaling prefetcher stop and tearing down producer.z)BackgroundPrefetcher closed successfully.)rr rrrFrjr;rr's r _async_closez!BackgroundPrefetcher._async_closes## F F LLO P}}T]]%=%=mm((*** MM & & ( LLD E F F F+  F F F FsWC B1C AB7,B3-3B7 C +B5,C 3B75C 7C =C >C C rSrrcBK|d}| |j}t||j}tjd||||jk\s||k\ry|jr tj dt d|j||d{S7w)z)Asynchronous API counterpart to `_fetch`.Nrz7Asynchronous afetch called for bounds start=%s, end=%s.rz=Cannot fetch data: BackgroundPrefetcher is stopped or closed.zThe file instance has been closed. This can occur if a close operation is executed concurrently while a read operation is still in progress.)r7rKr rrFrr&rrrSrs rafetchzBackgroundPrefetcher.afetch(s =E ;))C#tyy! Euc  DII # ?? LLO X  &&uc2222sBBBBcZtj|j|j||S)z/Synchronous API wrapper delegating to `afetch`.)rrrrrs rfetchzBackgroundPrefetcher.fetchBs# 4;;sCCrcNKd|_|jd{y7w)z>Safely shuts down the prefetcher from an asynchronous context.TN)rFrr's rrzBackgroundPrefetcher.acloseGs!!!s %#%timeoutcd|_ t|j|j|dy#tj $rt jd|Yyt$rt jddYywxYw)aSafely shuts down the prefetcher from a synchronous context. Schedules teardown on the IO loop. If the wait times out, teardown continues in the background to let in-flight reads finish cleanly. Args: timeout: Maximum seconds to wait for teardown to finish. Defaults to 10.0 seconds. If 0 or negative, teardown runs fire-and-forget. If None, waits indefinitely. TzBackgroundPrefetcher teardown)r descriptionzbBackgroundPrefetcher teardown did not complete within %ss; it will keep running in the background.z,Exception while closing BackgroundPrefetcher)exc_infoN) rFrrrrFSTimeoutErrorr warningrs)rrs rrzBackgroundPrefetcher.closeLsy   !!;   ))  NN:    NN>   s#-)A:A:9A:)NN)g$@)r,r-r.r/rr0rrrrrrsrtrrrrrrrfloatrr3rrr?r?KsHRV^F ^F/2^F@9"#"DLF 3#*33:3%34D3:DC$JD5D " UT\rr?)rPloggingrB collectionsrrrrutilsr getLoggerr,r r r5r<r?r3rrrs_!   8 $EEPx#x#v x3x3v^^r