0FieFddlmZddlZddlZddlZddlZddlmZmZm Z ddlm Z ddl m Z ddl mZddlmZddlmZdd lmZdd lmZmZGd d ZGd deZdS)) annotationsN)Callable GeneratorIterable)contextmanager)Any)concat) ShardsBuffer)ResourceLimiterpickle_bytelist)Deadline log_errorsceZdZUded<ded<ded<dd Zddd ZddZdddZddZe ddZ e ddZ dS) ReadWriteLockzthreading.Condition _conditionint_n_readsbool_write_pendingreturnNonectjtj|_d|_d|_d|_dS)NrF) threading ConditionLockrrr _write_activeselfs 9lib/python3.11/site-packages/distributed/shuffle/_disk.py__init__zReadWriteLock.__init__s9#-in.>.>?? #"timeoutfloatctj|dkr|nd}j5jfd|j}|dur ddddSd_jfd|j}|dur.d_j ddddSd_ ddddS#1swxYwYdS)Nrcj SNrrsr z-ReadWriteLock.acquire_write.." D//r"r$FTcjdkS)Nr)rrsr r*z-ReadWriteLock.acquire_write..)s *r")rafterrwait_for remainingr notify_allrrr$deadlineresults` r acquire_writezReadWriteLock.acquire_writes>W\\''tDD _  _--////9K.F         #'D _--****H4F.F&+#**,,,         "&D #                  s*CAC9CCCc|j5|jdurtdd|_d|_|jddddS#1swxYwYdS)NFz#Tried releasing unlocked write lock)rr RuntimeErrorrr1rs r release_writezReadWriteLock.release_write3s _ ) )!U**"#HIII"'D !&D  O & & ( ( (  ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) )sAAAActj|dkr|nd}j5jfd|j}|dur ddddSxjdz c_ ddddS#1swxYwYdS)Nrcj Sr(r)rsr r*z,ReadWriteLock.acquire_read..?r+r"r,FT)rr.rr/r0rr2s` r acquire_readzReadWriteLock.acquire_read;s>W\\''tDD _  _--////9K.F          MMQ MM                  s*A:A::A>A>c|j5|jdkrtd|xjdzc_|jdkr|jddddS#1swxYwYdS)Nrz"Tired releasing unlocked read lockr;)rrr7r1rs r release_readzReadWriteLock.release_readFs _ - -}!!"#GHHH MMQ MM}!!**,,,  - - - - - - - - - - - - - - - - - -sAA$$A(+A(Generator[None, None, None]c#K| dV|dS#|wxYwr()r5r8rs r writezReadWriteLock.writeNsV  ! EEE    D   2Ac#K| dV|dS#|wxYwr()r<r>rs r readzReadWriteLock.readVsV  EEE        D      rBNrr)r#)r$r%rr)rr?) __name__ __module__ __qualname____annotations__r!r5r8r<r>rrArDr"r rrs####MMM#### *))))     ----!!!^!   ^   r"rc@eZdZdZdfd ZddZddZdfd ZxZS)DiskShardsBufferaAccept, buffer, and write many small objects to many files This takes in lots of small objects, writes them to a local directory, and then reads them back when all writes are complete. It buffers these objects in memory so that it can optimize disk access for larger writes. **State** - shards: dict[str, list[bytes]] This is our in-memory buffer of data waiting to be written to files. - sizes: dict[str, int] The size of each list of shards. We find the largest and write data from that buffer Parameters ---------- directory : str or pathlib.Path Where to write and read data. Ideally points to fast disk. memory_limiter : ResourceLimiter Limiter for in-memory buffering (at most this much data) before writes to disk occur. If the incoming data that has yet to be processed exceeds this limit, then the buffer will block until below the threshold. See :meth:`.write` for the implementation of this scheme. directorystr | pathlib.PathrD)Callable[[pathlib.Path], tuple[Any, int]]memory_limiterr ct|dtj||_|jdd|_||_t|_ dS)Nr;)rPconcurrency_limitT)exist_okF) superr!pathlibPathrMmkdir_closed_readr_directory_lock)rrMrDrP __class__s r r!zDiskShardsBuffer.__init__|sx )    !i00 d+++  ,r"idstrshards list[Any]rrc TKt5|d5|j5|jrt dt |dtr|}ntd|D}t|j t|z d5}| |dddn #1swxYwYdddn #1swxYwYdddn #1swxYwYddddS#1swxYwYdS)a'Write one buffer to file This function was built to offload the disk IO, but since then we've decided to keep this within the event loop (disk bandwidth should be prioritized, and writes are typically small enough to not be a big deal). Most of the logic here is about possibly going back to a separate thread, or about diagnostics. If things don't change much in the future then we should consider simplifying this considerably and dropping the write into communicate above. rAAlready closedrc34K|]}t|VdSr(r ).0shards r z,DiskShardsBuffer._process..s*'S'S5(>(>'S'S'S'S'S'Sr"ab)modeN) rtimerZrDrXr7 isinstancebytesr openrMr] writelines)rr\r^framesfs r _processzDiskShardsBuffer._processs$\\ - -7## - -)..00--|=*+;<<<"&)U33T!'"('S'SF'S'S'S!S!Sdns2ww6TBBB-a V,,,------------------------------ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -s}DDA4C.5C C.C C.C C." D.C2 2D5C2 6D9 DD D D DD!$D!rc@||jstd |d5|j5|jrtd||jt|z \}}dddn #1swxYwYdddn #1swxYwYn#t$rt|wxYw|r|xj |z c_ |St|)z%Read a complete file back into memoryz$Tried to read from file before done.rDraN)raise_on_exception _inputs_doner7rhrZrDrXrYrMr]resolveFileNotFoundErrorKeyError bytes_read)rr\datasizes r rDzDiskShardsBuffer.reads !!!  GEFF F 6"" R R)..00RR|=*+;<< B" B B"B B""B&)B&)rMrNrDrOrPr )r\r]r^r_rr)r\r]rrrE) rFrGrH__doc__r!rorDrz __classcell__)r[s@r rLrL_s8//////"!-!-!-!-F*..........r"rL) __future__rr{rUr}rcollections.abcrrrrtypingrtoolzr distributed.shuffle._bufferr distributed.shuffle._limiterr distributed.shuffle._pickler distributed.utilsrrrrLrJr"r rsR"""""" 9999999999%%%%%%44444488888877777722222222I I I I I I I I Xk.k.k.k.k.|k.k.k.k.k.r"