a h @sUddlmZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z ddl mZddlmZddlmZmZmZmZmZmZmZerddlZddlmZed Zned Zed Zed Zee e ge!fZ"ee#e e fge!fZ$Gd d d Z%Gddde%Z&Gddde%Z'Gddde%Z(Gddde%Z)Gddde%Z*Gddde%Z+Gddde%Z,GdddeeefZ-Gddde%Z.de%iZ/d e0d!<d)d#d$d%d&d'd(Z1e%e&e*e'e)e(e+e,e.f D]Z2e1e2qdS)*) annotationsN) OrderedDict)FutureThreadPoolExecutor)groupby) itemgetter) TYPE_CHECKINGAnyCallableClassVarGeneric NamedTupleTypeVar) ParamSpecPTZfsspecc@sneZdZUdZdZded<dddddd d Zd d d d ddZddddZddddZ ddddZ dS) BaseCacheagPass-though cache: doesn't keep anything, calls every time Acts as base class for other cachers Parameters ---------- blocksize: int How far to read ahead in numbers of bytes fetcher: func Function of the form f(start, end) which gets bytes from remote as specified size: int How big this file is none ClassVar[str]nameintFetcherNone blocksizefetchersizereturncCs.||_d|_||_||_d|_d|_d|_dSNr)rnblocksrr hit_count miss_counttotal_requested_bytesselfrrrr%|jdkr|jdkrdSd|jd|jd|jd|jd S)z2Return a formatted string of the cache statistics.rz , z: z hits, z misses, z total requested bytes)r r!rr"r1r%r%r& _log_statsUszBaseCache._log_statscCs@d|jjd|jd|jd|jd|jd|jd|jdS) Nz ) __class____name__rrrr r!r"r1r%r%r&__repr__\szBaseCache.__repr__N) r7 __module__ __qualname____doc__r__annotations__r'r/r2r5r8r%r%r%r&r*s    rc sveZdZdZdZdddddddd d fd d Zd dddZddddddZddddZdd dddZ Z S) MMapCacheamemory-mapped sparse file cache Opens temporary file, which is filled blocks-wise when data is requested. Ensure there is enough disc space in the temporary location. This cache method might only work on posix Parameters ---------- blocksize: int How far to read ahead in numbers of bytes fetcher: Fetcher Function of the form f(start, end) which gets bytes from remote as specified size: int How big this file is location: str Where to create the temporary file. If None, a temporary file is created using tempfile.TemporaryFile(). blocks: set[int] Set of block numbers that have already been fetched. If None, an empty set is created. multi_fetcher: MultiFetcher Function of the form f([(start, end)]) which gets bytes from remote as specified. This function is used to fetch multiple blocks at once. If not specified, the fetcher function is used instead. mmapNrrz str | Nonezset[int] | NonezMultiFetcher | Noner)rrrlocationblocks multi_fetcherrcs>t||||durtn||_||_||_||_dSN)superr'setr@r?rA _makefilecache)r$rrrr?r@rAr6r%r&r's zMMapCache.__init__zmmap.mmap | bytearrayr0cCsddl}ddl}|jdkr tS|jdus8tj|js|jdurT|}t |_ n t |jd}| |jd| d|n t |jd}|||jS)Nrzwb+1zr+b)r>tempfiler bytearrayr?ospathexists TemporaryFilerDr@openseekwriteflushfileno)r$r>rJfdr%r%r&rEs       zMMapCache._makefiler(r)r+endrc s$td|d||dur"d}|dur0j}|jksB||krFdS|j}|j}t||d}fdd|D}jtfdd|D7_g}tt|d d d D]\}} t t t d| } | dj} t | d jjj} j | | 7_ td | dd| d d| d| d || | fj| jt| 7_q|svj||Sjrtd |t|D]:\} } || \} } td| d| | j| | <qn<|D]6\} } td| d| | | j| | <qވj||S)NzMMap cache fetching -rr-rHc3s|]}|jvr|VqdSrBr@.0ir1r%r& r-z#MMapCache._fetch..c3s|]}|jvrdVqdS)rHNrYrZr1r%r&r]r-cSs|d|dS)NrrHr%)xr%r%r&r-z"MMapCache._fetch..)keyzMMap get blocks z ()zMMap copy block (zMMap get block ()loggerdebugrrranger sumr enumeratetuplemaprminr"appendr@updater!lenrFrAr)r$r+rWZ start_blockZ end_blockZ block_rangeZneedranges_Z_blocksZsstartsendidxrr%r1r&r/sH    $   zMMapCache._fetchdict[str, Any]cCs|j}|d=|S)NrF)__dict__copyr$stater%r%r& __getstate__s zMMapCache.__getstate__rwrcCs|j|||_dSrB)rtrlrErFrvr%r%r& __setstate__s zMMapCache.__setstate__)NNN) r7r9r:r;rr'rEr/rxrz __classcell__r%r%rGr&r=is >r=csBeZdZdZdZdddddfdd Zd d d d d d ZZS)ReadAheadCachea!Cache which reads only when we get beyond a block of data This is a much simpler version of BytesCache, and does not attempt to fill holes in the cache or keep fragments alive. It is best suited to many small reads in a sequential order (e.g., reading lines from a file). Z readaheadrrrrcs&t|||d|_d|_d|_dS)Nr-r)rCr'rFr+rWr#rGr%r&r'szReadAheadCache.__init__r(r)rVcCs>|dur d}|dus||jkr$|j}||jks6||kr:dS||}||jkr~||jkr~|jd7_|j||j||jS|j|kr|jkrnn6|jd7_|j||jd}|t|8}|j}n|jd7_d}t|j||j}|j ||7_ | |||_||_|jt|j|_||jd|SNrr-rH) rr+rWr rFr!rmrjrr"r)r$r+rWlpartr%r%r&r/s. zReadAheadCache._fetchr7r9r:r;rr'r/r{r%r%rGr&r|sr|csBeZdZdZdZdddddfdd Zd d d d d d ZZS)FirstChunkCachezCaches the first block of a file only This may be useful for file types where the metadata is stored in the header, but is randomly accessed. firstrrrrcs&||kr |}t|||d|_dSrB)rCr'rFr#rGr%r&r'.szFirstChunkCache.__init__r(r)rVcCs0|pd}||jkr tddSt||j}||jkr|jdur|jd7_||jkr|j|7_|d|}|d|j|_||dS|d|j|_|j|j7_|j||}||jkr|j||j7_|||j|7}|j d7_ |S|jd7_|j||7_|||SdS)Nrz,FirstChunkCache: requested start > file sizer-rH) rrcrdrjrrFr!r"rr )r$r+rWdatarr%r%r&r/5s0         zFirstChunkCache._fetchrr%r%rGr&r%srcseZdZdZdZdddddddfdd Zd d Zd d ddZd ddddZddddddZ dddfdd Z ddddddddZ Z S) BlockCachea Cache holding memory as a set of blocks. Requests are only ever made ``blocksize`` at a time, and are stored in an LRU cache. The least recently accessed block is discarded when more than ``maxblocks`` are stored. Parameters ---------- blocksize : int The number of bytes to store in each block. Requests are only ever made for ``blocksize``, so this should balance the overhead of making a request against the granularity of the blocks. fetcher : Callable size : int The total size of the file being cached. maxblocks : int The maximum number of blocks to cache for. The maximum memory use for this cache is then ``blocksize * maxblocks``. Z blockcache rrrrrr maxblocksrcs<t|||t|||_||_t||j|_ dSrB) rCr'mathceilrr functools lru_cache _fetch_block_fetch_block_cachedr$rrrrrGr%r&r'lszBlockCache.__init__cCs |jSz The statistics on the block cache. Returns ------- NamedTuple Returned directly from the LRU Cache used internally. r cache_infor1r%r%r&rts zBlockCache.cache_inforsr0cCs|j}|d=|S)Nrrtrvr%r%r&rxszBlockCache.__getstate__rycCs&|j|t|d|j|_dS)Nr)rtrlrrrrrvr%r%r&rzs  zBlockCache.__setstate__r(r)rVcCst|dur d}|dur|j}||jks,||kr0dS||j}||j}t||dD]}||qR|j||||dS)Nrr-rHstart_block_numberend_block_number)rrrer _read_cache)r$r+rWrr block_numberr%r%r&r/s    zBlockCache._fetch)rrcst||jkr"td|d|jd||j}||j}|j||7_|jd7_td|t||}|S)= Fetch the block of data for `block_number`. 'block_number=(' is greater than the number of blocks (rbrHzBlockCache fetching block %d) r ValueErrorrr"r!rcinforCr/)r$rr+rWblock_contentsrGr%r&rs    zBlockCache._fetch_blockr+rWrrrc Cs||j}||j}|jd7_||kr@||}|||S|||dg}|t|jt|d||||d|d|SdSz Read from our block cache. Parameters ---------- start, end : int The start and end byte positions. start_block_number, end_block_number : int The start and end block numbers. rHNr-rr rextendrirerkjoin r$r+rWrrZ start_posZend_posblockoutr%r%r&rs     zBlockCache._read_cache)r) r7r9r:r;rr'rrxrzr/rrr{r%r%rGr&rSs rcs^eZdZUdZdZded<dddddd d fd d Zd d ddddZddddZZ S) BytesCacheaKCache which holds data in a in-memory bytes object Implements read-ahead by the block size, for semi-random reads progressing through the file. Parameters ---------- trim: bool As we read more data, whether to discard the start of the buffer when we are more than a blocksize ahead of it. r)rrTrrboolr)rrrtrimrcs,t|||d|_d|_d|_||_dS)Nr-)rCr'rFr+rWr)r$rrrrrGr%r&r's zBytesCache.__init__r(rVcCs|dur d}|dur|j}||jks,||kr0dS|jdur||jkr|jdur||jkr||j}|jd7_|j||||S|jrt|j||j}n|}||ks||jkrdS|jdus||jkr|jdus||jkr|j||7_|jd7_| |||_||_nF|jdus.J|jdus>J|jd7_||jkr|jdusv|j||jkr|j||7_| |||_||_n4|j|j|7_| ||j}||_||j|_n|jdurd||jkrd|j|jkrnh||j|jkr6|j||7_| |||_||_n.|j||j7_| |j|}|j||_|jt |j|_||j}|j||||}|j r|j|j|jd}|dkr|j|j|7_|j|j|d|_|Sr}) rr+rWr rFrrjr"r!rrmr)r$r+rWoffsetZbendnewrnumr%r%r&r/s|      zBytesCache._fetchr0cCs t|jSrB)rmrFr1r%r%r&__len__=szBytesCache.__len__)T) r7r9r:r;rr<r'r/rr{r%r%rGr&rs   IrcsPeZdZUdZdZded<dddddd d fd d Zddd dddZZS)AllBytesz!Cache entire contents of the fileallrrNr(zFetcher | Nonez bytes | Noner)rrrrrcsNt||||durD|jd7_|j|j7_|d|j}||_dS)NrHr)rCr'r!r"rrr)r$rrrrrGr%r&r'Fs zAllBytes.__init__r)r*cCs|jd7_|j||S)NrH)r rr.r%r%r&r/TszAllBytes._fetch)NNNN r7r9r:r;rr<r'r/r{r%r%rGr&rAs  rcsVeZdZUdZdZded<ddddd d d d fd d Zddddfdd ZZS)KnownPartsOfAFilea Cache holding known file parts. Parameters ---------- blocksize: int How far to read ahead in numbers of bytes fetcher: func Function of the form f(start, end) which gets bytes from remote as specified size: int How big this file is data: dict A dictionary mapping explicit `(start, stop)` file-offset tuples with known bytes. strict: bool, default True Whether to fetch reads that go beyond a known byte-range boundary. If `False`, any read that ends outside a known part will be zero padded. Note that zero padding will not be used for reads that begin outside a known byte-range. partsrrNTrrz#dict[tuple[int, int], bytes] | Nonerr )rrrrstrictroc st|||||_|rt|}|dg}||dg} |ddD]f\} } |d\} } | | kr| | f|d<| d|| | f7<qL|| | f| || | fqLtt|| |_ ni|_ dS)NrrHra) rCr'rsortedkeyspoprkdictzipr)r$rrrrrroZ old_offsetsoffsetsr@r+r,Zstart0Zstop0rGr%r&r'rs     zKnownPartsOfAFile.__init__r(r)r*cs:|dur d}|dur|j}d}|jD]\\}}}||krH|kr(nq(||}|||||}|jr||kr|krnn.|d||t|7}|jd7_|S|}qq(|jdurtd||fdt d||fdt d|d ||j ||7_ |j d7_ |t||S) Nrr-rHz&Read is outside the known file parts: z. z%. IO/caching performance may be poor!z!KnownPartsOfAFile cache fetching rX)rritemsrrmr rrwarningswarnrcrdr"r!rCr/)r$r+r,rZloc0loc1roffrGr%r&r/s0 zKnownPartsOfAFile._fetch)NTrr%r%rGr&rYs  rc@sveZdZdZGdddeZdddddd d Zd d d dddZdddddZd dddddZ ddddZ dS) UpdatableLRUzg Custom implementation of LRU cache that allows updating keys Used by BackgroudBlockCache c@s.eZdZUded<ded<ded<ded<dS)UpdatableLRU.CacheInforhitsmissesmaxsizecurrsizeN)r7r9r:r<r%r%r%r& CacheInfos rzCallable[P, T]rr)funcmax_sizercCs0t|_||_||_d|_d|_t|_ dSr) collectionsr_cache_func _max_size_hits_misses threadingLock_lock)r$rrr%r%r&r's  zUpdatableLRU.__init__zP.argszP.kwargsr)argskwargsrcOs|rtd||jL||jvrZ|j||jd7_|j|WdSWdn1sn0Y|j|i|}|jF||j|<|jd7_t|j|j kr|jj ddWdn1s0Y|S)Nz Got unexpected keyword argument rHFlast) TypeErrorrrr move_to_endrrrrmrpopitem)r$rrresultr%r%r&__call__s  6 ,zUpdatableLRU.__call__r r)rrcGs4|j||jvWdS1s&0YdSrB)rr)r$rr%r%r& is_key_cachedszUpdatableLRU.is_key_cached)rrrcGsR|j8||j|<t|j|jkr0|jjddWdn1sD0YdS)NFr)rrrmrr)r$rrr%r%r&add_keys zUpdatableLRU.add_keyrr0cCsH|j.|j|jt|j|j|jdWdS1s:0YdS)N)rrrr)rrrrmrrrr1r%r%r&rszUpdatableLRU.cache_infoN)r) r7r9r:r;r rr'rrrrr%r%r%r&rsrcseZdZUdZdZded<d"dddddd fd d Zd d ddZdd ddZdd ddZ ddddddZ d#ddddfdd Z ddddddd d!Z Z S)$BackgroundBlockCachea Cache holding memory as a set of blocks with pre-loading of the next block in the background. Requests are only ever made ``blocksize`` at a time, and are stored in an LRU cache. The least recently accessed block is discarded when more than ``maxblocks`` are stored. If the next block is not in cache, it is loaded in a separate thread in non-blocking way. Parameters ---------- blocksize : int The number of bytes to store in each block. Requests are only ever made for ``blocksize``, so this should balance the overhead of making a request against the granularity of the blocks. fetcher : Callable size : int The total size of the file being cached. maxblocks : int The maximum number of blocks to cache for. The maximum memory use for this cache is then ``blocksize * maxblocks``. Z backgroundrrrrrrrcsZt|||t|||_||_t|j||_t dd|_ d|_ d|_ t |_dS)NrH max_workers)rCr'rrrrrrrr_thread_executor_fetch_future_block_number _fetch_futurerr_fetch_future_lockrrGr%r&r's zBackgroundBlockCache.__init__rr0cCs |jSrrr1r%r%r&r#s zBackgroundBlockCache.cache_inforscCs(|j}|d=|d=|d=|d=|d=|S)Nrrrrrrrvr%r%r&rx.sz!BackgroundBlockCache.__getstate__cCsD|j|t|j|d|_tdd|_d|_d|_t |_ dS)NrrHr) rtrlrrrrrrrrrrrvr%r%r&rz7s   z!BackgroundBlockCache.__setstate__r(r)rVc Cs|dur d}|dur|j}||jks,||kr0dS||j}||j}d}d}|j|jdur|jduslJ|jrtd|j |j |jd|_d|_n:t ||jko|kn}|r|j}|j}d|_d|_Wdn1s0Y|dur"td|j | |t ||dD]}||q0|d} |jP|jdur| |j kr|j| s| |_|j|j| d|_Wdn1s0Y|j||||dS)Nrr-z3BlockCache joined background fetch without waiting.z(BlockCache waiting for background fetch.rHasyncr)rrrrrdonercrrrrrrerrrZsubmitrr) r$r+rWrrZfetch_future_block_numberZ fetch_futureZ must_joinrZend_block_plus_1r%r%r&r/?sr      $   &zBackgroundBlockCache._fetchsyncr3)rlog_inforcsv||jkr"td|d|jd||j}||j}td|||j||7_|jd7_t||}|S)rrrrbz!BlockCache fetching block (%s) %drH) rrrrcrr"r!rCr/)r$rrr+rWrrGr%r&rs   z!BackgroundBlockCache._fetch_blockrc Cs||j}||j}|jd7_||kr@||}|||S|||dg}|t|jt|d||||d|d|SdSrrrr%r%r&rs     z BackgroundBlockCache._read_cache)r)r)r7r9r:r;rr<r'rrxrzr/rrr{r%r%rGr&rs     Lrz!dict[str | None, type[BaseCache]]cachesFztype[BaseCache]rr)clsclobberrcCs6|j}|s*|tvr*td|dt||t|<dS)z'Register' cache implementation. Parameters ---------- clobber: bool, optional If set to True (default is False) - allow to overwrite existing entry. Raises ------ ValueError zCache with name z is already known: N)rrr)rrrr%r%r&register_caches  r)F)3 __future__rrrloggingrrLrrrconcurrent.futuresrr itertoolsroperatorrtypingrr r r r r rr>Ztyping_extensionsrrr getLoggerrcrr)rlistZ MultiFetcherrr=r|rrrrrrrrr<rcr%r%r%r&s`   $    ?.. ee<Q