U h @sUddlmZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z ddl m Z ddlmZddlmZmZmZmZmZmZmZmZmZerddlZddlmZedZnedZed Zed Z ee!e!ge"fZ#Gd d d Z$Gd dde$Z%Gddde$Z&Gddde$Z'Gddde$Z(Gddde$Z)Gddde$Z*Gddde$Z+GdddeeefZ,Gddde$Z-de$iZ.de/d <d(d"d#d$d%d&d'Z0e$e%e)e&e(e'e*e+e-f D]Z1e0e1qdS))) annotationsN)FutureThreadPoolExecutor)groupby) itemgetter) TYPE_CHECKINGAnyCallableClassVarGeneric NamedTupleOptional OrderedDictTypeVar) 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&@/var/www/html/venv/lib/python3.8/site-packages/fsspec/caching.py__init__<szBaseCache.__init__ int | NonebytesstartstoprcCs<|dkr d}|dkr|j}||jks,||kr0dS|||S)Nr)rrr%r,r-r&r&r'_fetchFszBaseCache._fetchrcCsd|_d|_d|_dS)zAReset hit and miss counts for a more ganular report e.g. by file.rN)r!r"r#r%r&r&r' _reset_statsOszBaseCache._reset_statsstrc Cs>|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#r2r&r&r' _log_statsUszBaseCache._log_statscCs@d|jjd|jd|jd|jd|jd|jd|jdS) Nz ) __class____name__rr rr!r"r#r2r&r&r'__repr__\szBaseCache.__repr__N) r8 __module__ __qualname____doc__r__annotations__r(r0r3r6r9r&r&r&r'r*s    rcsteZdZdZdZdddddddd fd d Zd d ddZddddddZdd ddZdddddZ Z S) MMapCachezmemory-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 mmapNrrz str | Nonezset[int] | Noner)rrrlocationblocksrcs8t||||dkrtn||_||_||_dSN)superr(setrAr@ _makefilecache)r%rrrr@rAr7r&r'r(tszMMapCache.__init__zmmap.mmap | bytearrayr1cCsddl}ddl}|jdkr tS|jdks8tj|js|jdkrT|}t |_ n t |jd}| |jd| d|n t |jd}|||jS)Nrzwb+1zr+b)r?tempfiler bytearrayr@ospathexists TemporaryFilerDrAopenseekwriteflushfileno)r%r?rJfdr&r&r'rEs       zMMapCache._makefiler)r*r,endrc sttd|d||dkr"d}|dkr0j}|jksB||krFdS|j}|j}t||d}fdd|D}jtfdd|D7_tt|d d d D]\}}t t t d|}|dj} t |d jjj} j | | 7_ td |dd|d d| d| d | | j| | <j|jt|7_qj||S)NzMMap cache fetching -rr.rHc3s|]}|jkr|VqdSrBrA.0ir2r&r' s z#MMapCache._fetch..c3s|]}|jkrdVqdS)rHNrYrZr2r&r'r]s cSs|d|dS)NrrHr&)xr&r&r'r.z"MMapCache._fetch..)keyzMMap get blocks z ())loggerdebugrrranger!sumr enumeratetuplemaprminr#rrFrAupdater"len) r%r,rWZ start_blockZ end_blockZ block_rangeZneed_Z_blocksZsstartsendr&r2r'r0s0   $ zMMapCache._fetchdict[str, Any]cCs|j}|d=|S)NrF)__dict__copyr%stater&r&r' __getstate__s zMMapCache.__getstate__rsrcCs|j|||_dSrB)rprkrErFrrr&r&r' __setstate__s zMMapCache.__setstate__)NN) r8r:r;r<rr(rEr0rtrv __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>|dkr d}|dks||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"rlrjrr#r)r%r,rWlpartr&r&r'r0s. zReadAheadCache._fetchr8r:r;r<rr(r0rwr&r&rGr'rxsrxcsBeZdZdZdZdddddfdd 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|jdkr|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,rWdatar{r&r&r'r0s0         zFirstChunkCache._fetchr|r&r&rGr'r}sr}cseZdZdZdZdddddddfdd 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(mathceilr r functools lru_cache _fetch_block_fetch_block_cachedr%rrrrrGr&r'r(FszBlockCache.__init__cCs |jSz The statistics on the block cache. Returns ------- NamedTuple Returned directly from the LRU Cache used internally. r cache_infor2r&r&r'rNs zBlockCache.cache_inforor1cCs|j}|d=|S)Nrrprrr&r&r'rtYszBlockCache.__getstate__rucCs&|j|t|d|j|_dS)Nr)rprkrrrrrrr&r&r'rv^s  zBlockCache.__setstate__r)r*rVcCst|dkr d}|dkr|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'r0ds    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"rcinforCr0)r%rr,rWblock_contentsrGr&r'r{s    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!rextendrireappendjoin r%r,rWrr start_posend_posblockoutr&r&r'rs     zBlockCache._read_cache)r) r8r:r;r<rr(rrtrvr0rrrwr&r&rGr'r-s 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|dkr d}|dkr|j}||jks,||kr0dS|jdk r||jkr|jdk r||jkr||j}|jd7_|j||||S|jrt|j||j}n|}||ks||jkrdS|jdks||jkr|jdks||jkr|j||7_|jd7_| |||_||_nF|jdk s.t |jdk s>t |jd7_||jkr|jdksv|j||jkr|j||7_| |||_||_n4|j|j|7_| ||j}||_||j|_n|jdk rd||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|_|Sry) rr,rWr!rFrrjr#r"rAssertionErrorrlr)r%r,rWoffsetZbendnewrnumr&r&r'r0s|      zBytesCache._fetchr1cCs t|jSrB)rlrFr2r&r&r'__len__szBytesCache.__len__)T) r8r:r;r<rr=r(r0rrwr&r&rGr'rs   IrcsPeZdZUdZdZded<dddddd d fd d Zddd dddZZS)AllBytesz!Cache entire contents of the fileallrrNr)zFetcher | Nonez bytes | Noner)rrrrrcsNt||||dkrD|jd7_|j|j7_|d|j}||_dS)NrHr)rCr(r"r#rrr)r%rrrrrGr&r'r( s zAllBytes.__init__r*r+cCs|jd7_|j||S)NrH)r!rr/r&r&r'r0.szAllBytes._fetch)NNNN r8r:r;r<rr=r(r0rwr&r&rGr'rs  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&Optional[dict[tuple[int, int], bytes]]rr)rrrrstrictrmc st|||||_|rt|}|dg}||dg} |ddD]f\} } |d\} } | | kr| | f|d<| d|| | f7<qL|| | f| || | fqLtt|| |_ ni|_ dS)NrrHra) rCr(rsortedkeyspoprdictzipr)r%rrrrrrmZ old_offsetsoffsetsrAr,r-Zstart0Zstop0rGr&r'r(Ls     zKnownPartsOfAFile.__init__r)r*r+cs:|dkr d}|dkr|j}d}|jD]\\}}}||krH|kr(nq(||}|||||}|jr||kr|krnn.|d||t|7}|jd7_|S|}qq(|jdkrtd||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)rritemsrrlr!rrwarningswarnrcrdr#r"rCr0)r%r,r-rZloc0Zloc1roffrGr&r'r0js0 zKnownPartsOfAFile._fetch)NTrr&r&rGr'r3s  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)r8r:r;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)argskwargsrc Os|rtd||j@||jkrX|j||jd7_|j|W5QRSW5QRX|j||}|j<||j|<|jd7_t|j|j kr|jj ddW5QRX|S)Nz Got unexpected keyword argument rHFlast) TypeErrorrrr move_to_endrrrrlrpopitem)r%rrresultr&r&r'__call__s     zUpdatableLRU.__call__rr)rrc Gs(|j||jkW5QRSQRXdSrB)rr)r%rr&r&r' is_key_cachedszUpdatableLRU.is_key_cached)rrrc Gs>|j.||j|<t|j|jkr0|jjddW5QRXdS)NFr)rrrlrr)r%rrr&r&r'add_keys zUpdatableLRU.add_keyrr1c Cs<|j,|j|jt|j|j|jdW5QRSQRXdS)N)rrrr)rrrrlrrrr2r&r&r'rszUpdatableLRU.cache_infoN)r) r8r:r;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``. backgroundrrrrrrrcsZt|||t|||_||_t|j||_t dd|_ d|_ d|_ t |_dS)NrH max_workers)rCr(rrr rrrrr_thread_executor_fetch_future_block_number _fetch_futurerr_fetch_future_lockrrGr&r'r(s zBackgroundBlockCache.__init__rr1cCs |jSrrr2r&r&r'rs zBackgroundBlockCache.cache_inforocCs(|j}|d=|d=|d=|d=|d=|S)Nrrrrrrrrr&r&r'rtsz!BackgroundBlockCache.__getstate__cCsD|j|t|j|d|_tdd|_d|_d|_t |_ dS)NrrHr) rprkrrrrrrrrrrrrr&r&r'rvs   z!BackgroundBlockCache.__setstate__r)r*rVc Cs|dkr d}|dkr|j}||jks,||kr0dS||j}||j}d}d}|j|jdk r|jdk slt|jrtd|j |j |jd|_d|_n:t ||jko|kn}|r|j}|j}d|_d|_W5QRX|dk rtd|j | |t ||dD]}| |q|d} |jF|jdkr~| |jkr~|j | s~| |_|j|j| d|_W5QRX|j||||dS)Nrr.z3BlockCache joined background fetch without waiting.z(BlockCache waiting for background fetch.rHasyncr)rrrrrrdonercrrrrrrer rrZsubmitrr) r%r,rWrrZfetch_future_block_numberZ fetch_futureZ must_joinrZend_block_plus_1r&r&r'r0sz        zBackgroundBlockCache._fetchsyncr4)rlog_inforcsv||jkr"td|d|jd||j}||j}td|||j||7_|jd7_t||}|S)rrrrbz!BlockCache fetching block (%s) %drH) r rrrcrr#r"rCr0)r%rrr,rWrrGr&r'res   z!BackgroundBlockCache._fetch_blockrc Cs||j}||j}|jd7_||kr@||}|||S|||dg}|t|jt|d||||d|d|SdSrrrr&r&r'rws     z BackgroundBlockCache._read_cache)r)r)r8r:r;r<rr=r(rrtrvr0rrrwr&r&rGr'rs     Lrz!dict[str | None, type[BaseCache]]cachesFztype[BaseCache]rr)clsclobberrcCs6|j}|s*|tkr*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)2 __future__rrrloggingrrLrrconcurrent.futuresrr itertoolsroperatorrtypingrrr r r r r rrr?typing_extensionsrrr getLoggerrcrr*rrr>rxr}rrrrrrrr=rcr&r&r&r's^   ,    ?h.. ee<Q