0Fies5ddlmZddlZddlmZddlmZmZmZm Z m Z m Z ddl m Z ddlmZddlmZmZmZmZddlZddlmZdd lmZdd lmZmZdd lmZdd lm Z dd l!m"Z"m#Z#ej$e%Z&e&'e"de&'e"dGddeZ(GddeZ)Gddej*ee+fZ,d'dZ-Gdde.Z/Gdd e.Z0Gd!d"e.Z1Gd#d$ej2Z3Gd%d&ej4ee+e5fZ6dS)() annotationsN) defaultdict)CallableHashableIteratorMappingMutableMappingSized)contextmanager)partial)Literal NamedTupleProtocolcast)Key) context_meter)deserialize_bytesserialize_bytelist)get_compression_settings safe_sizeof)RateLimiterFilternbytesz#Spill file on disk reached capacityzSpill to disk failedc8eZdZUdZded<ded<d dZd dZd S) SpilledSizez7Size of a key/value pair when spilled to disk, in bytesintmemorydiskotherreturncVt|j|jz|j|jzSNrrrselfrs 1lib/python3.11/site-packages/distributed/spill.py__add__zSpilledSize.__add__ $4;5ty5:7MNNNcVt|j|jz |j|jz Sr"r#r$s r&__sub__zSpilledSize.__sub__#r(r)N)rrr r)__name__ __module__ __qualname____doc____annotations__r'r+r)r&rrsbAAKKK IIIOOOOOOOOOOr)rc2eZdZdZeddZd dZdS) ManualEvictProtoaSDuck-type API that a third-party alternative to SpillBuffer must respect (in addition to MutableMapping) if it wishes to support spilling when the ``distributed.worker.memory.spill`` threshold is surpassed. This is public API. At the moment of writing, Dask-CUDA implements this protocol in the ProxifyHostFile class. r Sized | boolcdS)zAccess to fast memory. This is normally a MutableMapping, but for the purpose of the manual eviction API it is just tested for emptiness to know if there is anything to evict. Nr1r%s r&fastzManualEvictProto.fast0s r)rcdS)aManually evict a key/value pair from fast to slow memory. Return size of the evicted value in fast memory. If the eviction failed for whatever reason, return -1. This method must guarantee that the key/value pair that caused the issue has been retained in fast memory and that the problem has been logged internally. This method never raises. Nr1r6s r&evictzManualEvictProto.evict8s  r)N)r r4r r)r,r-r.r/propertyr7r9r1r)r&r3r3'sR   X        r)r3ceZdZUdZded<ded< d&d'fd Zed(dZed)dZd*fd Z d+dZ d,fd Z d-fd Z d.d/dZ ed0d Zed0d!Zed1d#Zed2d%ZxZS)3 SpillBufferaMutableMapping that automatically spills out dask key/value pairs to disk when the total size of the stored data exceeds the target. If max_spill is provided the key/value pairs won't be spilled once this threshold has been reached. Parameters ---------- spill_directory: str Location on disk to write the spill files to target: int Managed memory, in bytes, to start spilling at max_spill: int | False, optional Limit of number of bytes to be spilled on disk. Set to False to disable. zset[Key]logged_pickle_errorsz#defaultdict[tuple[str, str], float]cumulative_metricsFspill_directorystrtargetr max_spillint | Literal[False]c t||}tj|tj}t i||t t|_tt|_ dS)N)r7slownweight) SlowzictCacheWeakValueMappingsuper__init___in_memory_weightsetr>rfloatr?)r%r@rBrCrF slow_cached __class__s r&rNzSpillBuffer.__init__XsrOY//jt'<'>'>??  b{fEVWWW$'EE!"-e"4"4r)r Iterator[None]c#zKd fd }tj|5d Vd d d d S#1swxYwYd S) asCapture metrics re. disk read/write, serialize/deserialize, and compress/decompress. Note that this duplicates capturing from gather_dep, get_data, and execute. It is repeated here to make it possible to split serialize/deserialize and compress/decompress triggered by spill/unspill from those triggered by network comms. labelrvaluerQunitrAr Nonecdt|tsJj||fxx|z cc<dSr") isinstancerAr?)rVrWrXr%s r&metrics_callbackz6SpillBuffer._capture_metrics..metrics_callbackrsBeS)) ) ) )  #E4K 0 0 0E 9 0 0 0 0 0r)N)rVrrWrQrXrAr rY)r add_callback)r%r\s` r&_capture_metricszSpillBuffer._capture_metricsgs : : : : : : '(8 9 9   EEE                  s 044key Key | Nonec#@K dVdS#t$rL}|j\}||jvsJ||jvsJtdt d}~wt$r+tddt t$r}|j\}}||jvsJ||jvsJ||kr |J||=|||j vr7td|d|j |t d}~wwxYw)Nz;Spill file on disk reached capacity; keeping data in memoryz,Spill to disk failed; keeping data in memoryT)exc_infozFailed to pickle %r) MaxSpillExceededargsr7rFloggerwarning HandledErrorOSErrorerror PickleErrorr>add)r%r_ekey_eorig_es r&_handle_errorszSpillBuffer._handle_errorsysp# % EEEEE ! ! !vHUDI%%%% )))) NNM   ..  ! ! ! LLGRVL W W W..  % % %FME6DI%%%% ))))|| I  999LL!6LMMM-11%888"nn$) %s# DAAADA=DDrrWobjectrYc |5||5t|||j|dddn #1swxYwYddddS#1swxYwYdS#t $r||jvsJ||jvsJYdSwxYw)aIf sizeof(value) < target, write key/value pair to self.fast; this may in turn cause older keys to be spilled from fast to slow. If sizeof(value) >= target, write key/value pair directly to self.slow instead. Raises ------ Exception sizeof(value) >= target, and value failed to pickle. The key/value pair has been forgotten. In all other cases: - an older value was evicted and failed to pickle, - this value or an older one caused the disk to fill and raise OSError, - this value or an older one caused the max_spill threshold to be exceeded, this method does not raise and guarantees that the key/value that caused the issue remained in fast. N) r^rorM __setitem__r>discardrgr7rF)r%r_rWrSs r&rrzSpillBuffer.__setitem__sc( (&&(( 7 7$*=*=c*B*B 7 7##C///)11#666 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 ( ( ($)####di''''''' (sWBB=A6* B6A: :B=A: >B BBBBB B?>B?cZ |5|d5|j\}}}t t |cdddcdddS#1swxYwYddddS#1swxYwYdS#t $rYdSwxYw)aImplementation of :meth:`ManualEvictProto.evict`. Manually evict the oldest key/value pair, even if target has not been reached. Returns sizeof(value). If the eviction failed (value failed to pickle, disk full, or max_spill exceeded), return -1; the key/value pair that caused the issue will remain in fast. The exception has been logged internally. This method never raises. N)r^ror7r9rrrg)r%_rHs r&r9zSpillBuffer.evictsU &&(( ) )$*=*=d*C*C ) )#y00 1fC(( ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) )   22 s]BB2A7 B* B7A; ;B>A; ?B BBBBB B*)B*cV|5||jvrQtt|jj|}t jdddt jd|dt|cdddS#1swxYwYdS)Nz memory-readcountbytes) r^r7rrweightsr digest_metricrM __getitem__)r%r_ memory_sizerSs r&r}zSpillBuffer.__getitem__s  " " $ $ , ,di#3 (9#(>?? +M1gFFF+M;PPP77&&s++ , , , , , , , , , , , , , , , , , ,sA;BB"%B"c~t||j|dSr")rM __delitem__r>rsr%r_rSs r&rzSpillBuffer.__delitem__s8 C    !))#.....r)Ndefaultc td)NzAre you calling .pop(key, None) as a way to discard a key if it exists?It may cause data to be read back from disk! Please use `del` instead.)NotImplementedError)r%r_rs r&popzSpillBuffer.pops! U   r)Mapping[Key, object]c|jS)zxKey/value pairs stored in RAM. Alias of zict.Buffer.fast. For inspection only - do not modify directly! )r7r6s r&rzSpillBuffer.memory yr)c|jS)z~Key/value pairs spilled out to disk. Alias of zict.Buffer.slow. For inspection only - do not modify directly! )rFr6s r&rzSpillBuffer.diskrr)rIctttj|j}tt|jSr")rrJrKrFrIdata)r%caches r&_slow_uncachedzSpillBuffer._slow_uncacheds'TZ++D%*%%%r)rc|jjS)zNumber of bytes spilled to disk. Tuple of - output of sizeof() - pickled size The two may differ substantially, e.g. if sizeof() is inaccurate or in case of compression. )r total_weightr6s r& spilled_totalzSpillBuffer.spilled_totals"//r)F)r@rArBrrCrD)r rT)r_r`r rTr_rrWrpr rYr:r_rr rpr_rr rYr")r_rrrpr rp)r r)r rI)r r)r,r-r.r/r0rNr r^rorrr9r}rrr;rrrr __classcell__rSs@r&r=r=Es  #""";;;; +0 5 5 5 5 5 5 5^"$%$%$%^$%L((((((8" , , , , , ,//////     X X &&&X& 0 0 0X 0 0 0 0 0r)r=r_rrWrpr rc t|Sr"r)r_rWs r&rOrOs u  r)ceZdZdS)rcNr,r-r.r1r)r&rcrc Dr)rcceZdZdS)rjNrr1r)r&rjrjrr)rjceZdZdS)rgNrr1r)r&rgrgrr)rgc eZdZdfd ZxZS) AnyKeyFiler_rr rAc`tt|Sr")rM _safe_keyrArs r&rzAnyKeyFile._safe_keys!ww  S***r))r_rr rA)r,r-r.rrrs@r&rrs=++++++++++r)rc^eZdZUded<ded<ded<ddfd ZddZddZdfd ZxZS)rIrD max_weightzdict[Key, SpilledSize] weight_by_keyrrFr@rAc td}tttgtft t |d}t|tttttft|||_ i|_tdd|_dS)Nz+distributed.worker.memory.spill-compressionraise) compressionon_errorr)rrrrprzr rrMrNrr rrrrrr)r%r@rrdumprSs r&rNz Slow.__init__!s. 9   fXu_ % &K' R R R      U +Z-H-H I I   %'1--r)r_rr rpctjdd5|j|}dddn #1swxYwYtjdddtjdt |d||}|S)Nz disk-readsecondsrxryrz)rmeterdr|lenload)r%r_pickledouts r&r}zSlow.__getitem__8s  i 8 8 " "fSkG " " " " " " " " " " " " " " "#KG<<<#KWwGGGii   s 044rWrYc| ||}n"#t$r}t||d}~wwxYw||jvsJ||jvsJt t t|}|jdur'|j j |z|jkrt|tj dd5||j|<dddn #1swxYwYtjdddtjd|dtt!||}||j|<|xj |z c_ dS)NFz disk-writerrxryrz)r Exceptionrjrrsummaprrrrrcrrr|rr)r%r_rWrrl pickled_sizerHs r&rrzSlow.__setitem__@s &ii&&GG & & &c1%% %  &$&    $,,,,,3vw//00  O5 ( (!&5GG#3'' ' y 9 9 " "!DF3K " " " " " " " " " " " " " " "#L!W===#L,HHH[//>>"(3 V#s" 7277 CCCct||xj|j|zc_dSr")rMrrrrrs r&rzSlow.__delitem__csF C    T/33C888r)r)r@rArrDrrr) r,r-r.r0rNr}rrrrrs@r&rIrIs$$$$))))........!$!$!$!$F9999999999r)rI)r_rrWrpr r)7 __future__rlogging collectionsrcollections.abcrrrrr r contextlibr functoolsr typingr rrrrJ dask.typingrdistributed.metricsrdistributed.protocolrr distributed.protocol.compressionrdistributed.sizeofrdistributed.utilsrr getLoggerr,re addFilterrr3Bufferrpr=rOrrcrjrgFilerFuncrzrIr1r)r&rs#""""""######XXXXXXXXXXXXXXXX%%%%%%666666666666 ------FFFFFFFFEEEEEE******77777777  8 $ $""#HIIJJJ""#9::;;; O O O O O* O O O     x   <}0}0}0}0}0$+c6k*}0}0}0@      y        )        9   ++++++++ I9I9I9I9I949S&%' (I9I9I9I9I9r)