ԋg5VddlmZddlZddlmZddlmZmZmZm Z m Z m Z ddl m Z ddlmZddlmZmZmZmZddlZddlmZdd lmZdd lmZmZdd lmZdd lm Z dd l!m"Z"m#Z#ejHe%Z&e&jOe"de&jOe"dGddeZ(GddeZ)GddejTee+fZ,d!dZ-Gdde.Z/Gdde0Z1Gdde.Z2GddejfZ4Gdd ejjee+e6fZ7y)") 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 failedc6eZdZUdZded<ded<ddZddZy) SpilledSizez7Size of a key/value pair when spilled to disk, in bytesintmemorydiskcvt|j|jz|j|jzSNrrrselfothers 1lib/python3.12/site-packages/distributed/spill.py__add__zSpilledSize.__add__ )4;;5tyy5::7MNNcvt|j|jz |j|jz Sr r!r"s r%__sub__zSpilledSize.__sub__#r'r(N)r$rreturnr)__name__ __module__ __qualname____doc____annotations__r&r*r(r%rrsA K IOOr(rc*eZdZdZeddZddZy)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. cy)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(cy)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. Nr1r5s r%evictzManualEvictProto.evict8s r(N)r+z Sized | boolr+r)r,r-r.r/propertyr6r8r1r(r%r3r3's     r(r3ceZdZUdZded<ded< d dfd ZeddZeddZdfd Z dd Z dfd Z dfd Z ddd Z eddZeddZeddZeddZxZS) 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_metricsct||}tj|tj}t|i||t t|_tt|_ y)N)r6slownweight) SlowzictCacheWeakValueMappingsuper__init___in_memory_weightsetr=rfloatr>)r#spill_directorytarget max_spillr@ slow_cached __class__s r%rHzSpillBuffer.__init__XsXOY/jjt'<'<'>?  b{fEVW$'E!"-e"4r(c#vKdfd }tj|5ddddy#1swYyxYww)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. c\t|tsJj||fxx|z cc<yr ) isinstancestrr>)labelvalueunitr#s r%metrics_callbackz6SpillBuffer._capture_metrics..metrics_callbackrs-eS) ))  # #E4K 0E 9 0r(N)rUrrVrKrWrTr+None)r add_callback)r#rXs` r%_capture_metricszSpillBuffer._capture_metricsgs+ : ' '(8 9 : 9 9s9- 969c#K dy#t$rR}|j\}||jvsJ||jvsJtj dt d}~wt$r"tjddt t$r}|j|jvsJ|j|jvsJ|j|k(r|J||=|j|jvrGtjd|jd|jj|jt d}~wwxYww)Nz;Spill file on disk reached capacity; keeping data in memoryz,Spill to disk failed; keeping data in memoryT)exc_infozFailed to pickle %r) MaxSpillExceededargsr6r@loggerwarning HandledErrorOSErrorerror PickleErrorkeyr=add)r#rfekey_es r%_handle_errorszSpillBuffer._handle_errorsys(" %  !vvHUDII% %% ) )) NNM .  ! LLGRVL W.  %55DII% %%55 ) ))uu|& I 55 9 99LL!6LM--11!%%8"n$' %s2E E E A A5E B4EE  EcR |j5|j|5t| |||jj |ddddddy#1swYxYw#1swYyxYw#t $r#||jvsJ||jvsJYywxYw)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[rjrG __setitem__r=discardrbr6r@)r#rfrVrPs r%rlzSpillBuffer.__setitem__s( (&&($*=*=c*B#C/))11#6+C((*B*B(( ($))# ##dii' '' (sEA:A.,A"A.A:"A+ 'A..A73A:7A::)B&%B&c* |j5|jd5|jj\}}}t t |cdddcdddS#1swYnxYwdddy#1swYyxYw#t $rYywxYw)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[rjr6r8rrrb)r#_rBs r%r8zSpillBuffer.evictss &&($*=*=d*C#yy0 1fC(+D*C((*C*C(((  sKBA:.A% A: B%A. *A:1B:B?BB BBc4|j5||jvrUtt|jj|}t j dddt j d|dt|!|cdddS#1swYyxYw)Nz memory-readcountbytes) r[r6rrweightsr digest_metricrG __getitem__)r#rf memory_sizerPs r%rwzSpillBuffer.__getitem__sw  " " $dii#3 (9(9#(>? ++M1gF++M;P7&s+% $ $s A2BBcZt|||jj|yr )rG __delitem__r=rmr#rfrPs r%rzzSpillBuffer.__delitem__s$ C  !!))#.r(ctd)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#rfdefaults r%popzSpillBuffer.pops! U  r(c|jS)zxKey/value pairs stored in RAM. Alias of zict.Buffer.fast. For inspection only - do not modify directly! )r6r5s r%rzSpillBuffer.memory yyr(c|jS)z~Key/value pairs spilled out to disk. Alias of zict.Buffer.slow. For inspection only - do not modify directly! )r@r5s r%rzSpillBuffer.diskrr(c~ttj|j}tt|j Sr )rrDrEr@rCdata)r#caches r%_slow_uncachedzSpillBuffer._slow_uncacheds'TZZ+D%**%%r(c.|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_weightr5s r% spilled_totalzSpillBuffer.spilled_totals""///r(F)rLrTrMrrNint | Literal[False])r+Iterator[None])rfz Key | Noner+rrfrrVobjectr+rYr9rfrr+rrfrr+rYr )rfrr~rr+r)r+zMapping[Key, object])r+rC)r+r)r,r-r.r/r0rHr r[rjrlr8rwrzrr:rrrr __classcell__rPs@r%r<r<Es #";; +0 5 5 5( 5"#%#%J(8" ,/   && 0 0r(r<ct|Sr r)rfrVs r%rIrIs u r(c eZdZy)r^Nr,r-r.r1r(r%r^r^ r(r^c&eZdZddZeddZy)rec d|jS)NzFailed to pickle )rfr5s r%__str__zPickleError.__str__s"488,//r(c |jdS)Nr)r_r5s r%rfzPickleError.keysyy|r(N)r+rT)r+r)r,r-r.rr:rfr1r(r%rere s0r(rec eZdZy)rbNrr1r(r%rbrbrr(rbc eZdZdfd ZxZS) AnyKeyFilec4t|t|Sr )rG _safe_keyrTr{s r%rzAnyKeyFile._safe_keysw S**r()rfrr+rT)r,r-r.rrrs@r%rrs ++r(rc^eZdZUded<ded<ded<d d fd Zd dZdd Zdfd ZxZS)rCr max_weightzdict[Key, SpilledSize] weight_by_keyrrc (td}tttgtft t |d}t|!|tttttft|||_ i|_tdd|_y)Nz+distributed.worker.memory.spill-compressionraise) compressionon_errorr)rrrrrtr rrGrHrr rrrrrr)r#rLrrdumprPs r%rHz Slow.__init__%s. 9  fXu_ % &K' R     U +Z-H I %'1-r(c tjdd5|j|}dddtjdddtjdt d|j |}|S#1swYSxYw)Nz disk-readsecondsrrrsrt)rmeterdrvlenload)r#rfpickledouts r%rwzSlow.__getitem__<sh  i 8ffSkG9##KG<##KWwGii  9 8s A99Bc |j|}||jvsJ||jvsJt t t|}|jdur1|jj|z|jkDr t|tjdd5||j|<dddtjdddtjd|dtt!||}||j|<|xj|z c_ y#t$r}t||d}~wwxYw#1swYxYw)NFz disk-writerrrrsrt)r Exceptionrerrsummaprrrrr^rrrvrr)r#rfrVrrh pickled_sizerBs r%rlzSlow.__setitem__Ds* *ii&G$&&   $,,,,,3vw/0  OO5 (!!&&5G#3' ' y 9!DFF3K:##L!W=##L,H[/>"(3 V#= *c" )  *.: 9s#D!!D>! D;* D66D;>Ect|||xj|jj |zc_yr )rGrzrrrr{s r%rzzSlow.__delitem__gs2 C  T//33C88r(r)rLrTrrrrr) r,r-r.r0rHrwrlrzrrs@r%rCrC s/$$))..!$F99r(rC)rfrrVrr+r)8 __future__rlogging collectionsrcollections.abcrrrrr r contextlibr functoolsr typingr rrrrD dask.typingrdistributed.metricsrdistributed.protocolrr distributed.protocol.compressionrdistributed.sizeofrdistributed.utilsrr getLoggerr,r` addFilterrr3Bufferrr<rIrr^ TypeErrorrerbFilerFuncrtrCr1r(r%rs"#XX%66 -FE*7   8 $"#HIJ"#9:; O* O x <|0$++c6k*|0~  y ) 9 ++ I9499S&%' (I9r(