0Fie@UdZddlmZddlZddlZddlmZddlmZm Z ddl m Z ddl m Z ddlmZdd lmZdd lmZmZmZdd lmZddlZdd lmZdd lmZmZddlmZddl m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/m0Z0ddl1m2Z2ddl3m4Z4erddl5Z6ddl7m8Z8ddl9m:Z;ede?d<edfZ@de?d<ee$e!ZNdIdCZOdS)Ju[ Utilities for rechunking arrays through p2p shuffles ==================================================== Tensors (or n-D arrays) in dask are split up across the workers as regular n-D "chunks" or bricks. These bricks are stacked up to form the global array. A key algorithm for these tensors is to "rechunk" them. That is to reassemble the same global representation using differently shaped n-D bricks. For example, to take an FFT of an n-D array, one uses a sequence of 1D FFTs along each axis. The implementation in dask (and indeed almost all distributed array frameworks) requires that 1D axis along which the FFT is taken is local to a single brick. So to perform the global FFT we need to arrange that each axis in turn is local to bricks. This can be achieved through all-to-all communication between the workers to exchange sub-pieces of their individual bricks, given a "rechunking" scheme. To perform the redistribution, each input brick is cut up into some number of smaller pieces, each of which contributes to one of the output bricks. The mapping from input brick to output bricks decomposes into the Cartesian product of axis-by-axis mappings. To see this, consider first a 1D example. Suppose our array is split up into three equally sized bricks:: |----0----|----1----|----2----| And the requested output chunks are:: |--A--|--B--|----C----|---D---| So brick 0 contributes to output bricks A and B; brick 1 contributes to B and C; and brick 2 contributes to C and D. Now consider a 2D example of the same problem:: +----0----+----1----+----2----+ | | | | α | | | | | | | +---------+---------+---------+ | | | | β | | | | | | | +---------+---------+---------+ | | | | γ | | | | | | | +---------+---------+---------+ Each brick can be described as the ordered pair of row and column 1D bricks, (0, α), (0, β), ..., (2, γ). Since the rechunking does not also reshape the array, axes do not "interfere" with one another when determining output bricks:: +--A--+--B--+----C----+---D---+ | | | | | Σ | | | | | | | | | +-----+ ----+---------+-------+ | | | | | | | | | | | | | | | Π | | | | | | | | | | | | | | | | | | | +-----+-----+---------+-------+ Consider the output (B, Σ) brick. This is contributed to by the input (0, α) and (1, α) bricks. Determination of the subslices is just done by slicing the the axes separately and combining them. The key thing to note here is that we never need to create, and store, the dense 2D mapping, we can instead construct it on the fly for each output brick in turn as necessary. The implementation here uses :func:`split_axes` to construct these 1D rechunkings. The output partitioning in :meth:`~.ArrayRechunkRun.add_partition` then lazily constructs the subsection of the Cartesian product it needs to determine the slices of the current input brick. This approach relies on the generic p2p buffering machinery to ensure that there are not too many small messages exchanged, since no special effort is made to minimise messages between workers when a worker might have two adjacent input bricks that are sliced into the same output brick. ) annotationsN) defaultdict)CallableSequence)ThreadPoolExecutor) dataclass)product)Path) TYPE_CHECKINGAny NamedTuple)IOLoop)tokenize)HighLevelGraphMaterializedLayer) PooledRPCCall)NDIndex ShuffleId ShuffleRun ShuffleSpecget_worker_pluginhandle_transfer_errorshandle_unpack_errors)ResourceLimiter)unpickle_bytestream)ShuffleSchedulerPlugin) barrier_keyshuffle_barrier)ShuffleWorkerPlugin)sizeof) TypeAlias.r! ChunkedAxis ChunkedAxesNDSliceinput np.ndarrayidr input_chunkrnewolddiskboolreturnintc t|5t||t||||cdddS#1swxYwYdS)N)r'r)r*r+) partition_idspec)rr add_partitionArrayRechunkSpec)r%r'r(r)r*r+s } | |f| zt"|jf| z|| ||j|f||f| z<?t&|| f||<d|}|j td|DD]} t(|| |f||f| z<tjd 5t-|} t/j|| |g } |j| ||| cdddS#1swxYwYdS) Nr)r>dtypezrechunk-transfer-zdistributed.p2p.diskc34K|]}t|VdSr9len.0dims r4 zrechunk_p2p..s(!?!?s#c((!?!?!?!?!?!?z rechunk-p2p-c34K|]}t|VdSr9rBrDs r4rGzrechunk_p2p..s(!=!=s#c((!=!=!=!=!=!=rHc|ddS)Nkeys r4zrechunk_p2p..s3qrr7rH)shuffle) dependencies)meta)numpy dask.arrayarraysizeemptyshaper@rrrdaskconfiggetndindextupler>appendr5namerr;annotaterrfrom_collectionsArray) r<r>npdadsktoken _barrier_keyr_r+ transfer_keysindexlayergraphs r4 rechunk_p2prlsCv{{rxag>>>>C Q  Ey//00L &u & &D!788DME!?!?ah!?!?!???@@   dWu_---  VI     H  TGeO)%?C  !% ! !DE!=!=f!=!=!===>>LL .ulKTGeO 22 3 3 355!#&&/e1#NNNrxtV!444 555555555555555555s;FF"%F"c2eZdZUdZded<ded<ded<dS)Splita"Slice of a chunk that is concatenated with other splits to create a new chunk Splits define how to slice an input chunk on a single axis into small pieces that can be concatenated together with splits from other input chunks to create output chunks of a rechunk operation. r. chunk_index split_indexsliceN)__name__ __module__ __qualname____doc____annotations__rLrHr4rnrnsALLLLLrHrn SplitChunk SplitAxis SplitAxesc ddlm}|||}g}t|D]\}}d||D}t|D]G\}} t| D]2\} \} } || t || | 3H|D]} | d|||S)aCalculate how to split the old chunks on each axis to create the new chunks Parameters ---------- old : ChunkedAxes Chunks along each axis of the old array new : ChunkedAxes Chunks along each axis of the new array Returns ------- SplitAxes Splits along each axis that determine how to slice the input chunks to create the new chunks by concatenating the resulting shards. r) old_to_newcg|]}gSrLrL)rE_s r4 zsplit_axes..s;;;ar;;;rHc|jjSr9)rqstart)splits r4rOzsplit_axes..s U[->rHrM)dask.array.rechunkr{ enumerater^rnsort)r*r)r{ _old_to_newaxes axis_indexnew_axisold_axisnew_chunk_index new_chunkrpold_chunk_indexrq old_chunks r4 split_axesrs .-----*S#&&K D )+ 6 6   H;;3z?;;;*3H*=*=   &OY9B99M9M  5 5ou)00/;>> " @ @I NN>>N ? ? ? ? H KrHshards&list[list[tuple[NDIndex, np.ndarray]]]cddl}ddlm}i}|D]}|D] \}}|||< dt|D}t ||j|ksJ|j|d}|D]\}}||t|<| } || S)Nr) concatenate3c2g|]}t|dzS)rK)maxrDs r4r~z!convert_chunk..s"===C1 ===rHO)r@) rSdask.array.corerzipkeysrCprodrWitemsr]tolist) rrcrindexedsublistrishardsubshape rec_cat_argarrss r4 convert_chunkr s,,,,,,)+G### # #LE5"GENN #>=W\\^^(<===H w<<7278,, , , , ,"(83///K ** u$) E%LL!!     D <  rHcTeZdZdZd1fd Zd2d Zd3d%Zd4d)Zd5d+Zd6d/Z d7d0Z xZ S)8ArrayRechunkRuna4State for a single active rechunk execution This object is responsible for splitting, sending, receiving and combining data shards. It is entirely agnostic to the distributed system and can perform a rechunk with other run instances using `rpc``. The user of this needs to guarantee that only `ArrayRechunkRun`s of the same unique `ShuffleID` and `run_id` interact. Parameters ---------- worker_for: A mapping partition_id -> worker_address. old: Existing chunking of the array per dimension. new: Desired chunking of the array per dimension. id: A unique `ShuffleID` this belongs to. run_id: A unique identifier of the specific execution of the shuffle this belongs to. local_address: The local address this Shuffle can be contacted by using `rpc`. directory: The scratch directory to buffer data in. executor: Thread pool to use for offloading compute. rpc: A callable returning a PooledRPCCall to contact other Shuffle instances. Typically a ConnectionPool. scheduler: A PooledRPCCall to to contact the scheduler. memory_limiter_disk: memory_limiter_comm: A ``ResourceLimiter`` limiting the total amount of memory used in either buffer. worker_fordict[NDIndex, str]r*r#r)r'rrun_idr. local_addressstr directoryexecutorrrpcCallable[[str], PooledRPCCall] schedulerrmemory_limiter_diskrmemory_limiter_commsr+r,looprc nt|||||| | | | | | ||_||_t t }|D] \}}|||!t||_ ||_ t|||_ dS)N) r'rrrrrrrrr+r) super__init__r*r)rlistrr^dict partitions_ofrr)selfrr*r)r'rrrrrrrrr+rrpartaddr __class__s r4rzArrayRechunkRun.__init__Ks" '!5 3  #D)) $**,, - -JD$ $  & &t , , , ,!-00$$S#..rHdataFlist[tuple[NDIndex, list[tuple[NDIndex, tuple[NDIndex, np.ndarray]]]]]r-NonecK|tt}|D]k}|\}}||jvr|j||D] \}}|||!|xjt|z c_l~|sdS ||d{VdS#t$r }||_ d}~wwxYwr9) raise_if_closedrrreceivedaddr^ total_recvdr _write_to_disk Exception _exception) rrrdid1payloadid2res r4_receivezArrayRechunkRun._receivers T"" * *ALCdm## M  c " " "% * * Us ""5))))   q )      F %%f-- - - - - - - - - -   DO  sB<< CCCr&r0rdict[str, tuple[NDIndex, Any]]cvtt}tdt|jD}|D]\}t|\}}}||} | j| } ||j|||| ff]fd| DS)Nc3,K|]\}}||VdSr9rL)rEaxisis r4rGz3ArrayRechunkRun._shard_partition..s*WWqT!WWWWWWWrHc i|] \}}||f SrLrL)rEkvr0s r4 z4ArrayRechunkRun._shard_partition..s$===AL!$===rH) rrr rrbasecopyrr^r) rrr0outndsplitsndsplitro shard_indexndslicers ` r4_shard_partitionz ArrayRechunkRun._shard_partitionsLW L L WWC4V4VWWWX  G03W -KgMEz%   , - 4 4{E23    >=======rHrNkwargsr c ||}|jd5t|cdddS#1swxYwYdS)Nread)_read_from_disk _disk_buffertimer)rr0rNrrs r4_get_output_partitionz%ArrayRechunkRun._get_output_partitions ##L11  # #F + + ' ' && ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' 'sA  AAbufferc|Sr9rL)rrs r4 deserializezArrayRechunkRun.deserializes rHpathr 2tuple[list[list[tuple[NDIndex, np.ndarray]]], int]c|d5}ttj|d}dddn #1swxYwYt t |}||jfS)aOpen a memory-mapped file descriptor to disk, read all metadata, and unpickle all arrays. This is a fast sequence of short reads interleaved with seeks. Do not read in memory the actual data; the arrays' buffers will point to the memory-mapped area. The file descriptor will be automatically closed by the kernel when all the returned arrays are dereferenced, which will happen after the call to concatenate3. zr+b)moderN)open memoryviewmmapfilenorrnbytes)rrfhrrs r4rzArrayRechunkRun.readsYYEY " " ;b "))++q 9 9::F ; ; ; ; ; ; ; ; ; ; ; ; ; ; ;)&1122v}$$s5AAAc|j|Sr9)r)rr's r4_get_assigned_workerz$ArrayRechunkRun._get_assigned_workersr""rH)rrr*r#r)r#r'rrr.rrrrrrrrrrrrrrr+r,rr)rrr-r)rr&r0rr-r)r0rrNrrr r-r&)rr r-r )rr r-r)r'rr-r) rrrsrtrurrrrrrr __classcell__)rs@r4rr"s&&P%/%/%/%/%/%/N4>>>>, ' ' ' '%%%%"########rHrT)frozenc4eZdZUded<ded<ddZddZdS)r3r#r)r*pluginrr-rcttd|jD}||j|tS)Nc3NK|] }tt|V!dSr9)rangerC)rEcs r4rGz7ArrayRechunkSpec._pin_output_workers..s.>>eCFFmm>>>>>>rH)r r)_pin_output_workersr'_get_worker_for_hash_sharding)rr parts_outs r4rz$ArrayRechunkSpec._pin_output_workerss?>>TX>>>? )) GY =   rHrr.rrrc8t||j|j|j|tj|jjd|jd||j |jj |jj |jj |j |j|j|jjS)Nzshuffle--)rr*r)r'rrrrrrrrr+r)rr*r)r'osrjoinworkerlocal_directory _executoraddressrrrrr+r)rrrrs r4create_run_on_workerz%ArrayRechunkSpec.create_run_on_workers !wgll --47--V--% -/ !m- & :!'!<##    rHN)rrr-r)rr.rrrrr-r)rrrsrtrvrrrLrHr4r3r3sV          rHr3output_partitionworkers Sequence[str]rcPt|t|z}||S)zJGet address of target worker for this output partition using hash sharding)hashrC)rr rs r4rrs' W-A 1:rH)r%r&r'rr(rr)r#r*r#r+r,r-r.)r'rr6rr7r.r-r&)r<r=r>r#r-r=)r*r#r)r#r-ry)rrr-r&)rrr r r-r)Pru __future__rrr collectionsrcollections.abcrrconcurrent.futuresr dataclassesr itertoolsr pathlibr typingr r r tornado.iolooprrY dask.baserdask.highlevelgraphrrdistributed.corerdistributed.shuffle._corerrrrrrrdistributed.shuffle._limiterrdistributed.shuffle._pickler%distributed.shuffle._scheduler_pluginrdistributed.shuffle._shufflerr"distributed.shuffle._worker_pluginrdistributed.sizeofr rSrctyping_extensionsr!rTrUrdr]floatr"rvr#rqr$r5r;rlrnrrwrxryrrrr3rrLrHr4r"s^^^@#""""" ######........111111!!!!!!1111111111!!!!!! AAAAAAAA******988888;;;;;;HHHHHHEEEEEEEEBBBBBB%%%%%%++++++ucz* ****{C/0 00005#:&&&&&         &5&5&5&5RJ&U ####J' ''''I &&&&D.`#`#`#`#`#j,!67`#`#`#F $" " " " " {7+" " " JrH