0FieP80ddlmZddlZddlZddlmZddlmZddlm Z m Z m Z ddl m Z ddlmZddlmZdd lmZdd lmZmZmZmZmZdd lmZdd lmZdd lmZm Z e r ddl!Z"ddl#m$Z$ej%e&Z'GddZ(GddeZ)dS)) annotationsN)Sequence)ThreadPoolExecutor) TYPE_CHECKINGAnyoverload) thread_state) parse_bytes) WorkerPlugin)ToPickle)NDIndex ShuffleId ShuffleRunShuffleRunSpec ShuffleSpec)ShuffleClosedError)ResourceLimiter) log_errorssync)WorkerceZdZUded<ded<ded<ded<d ed <d ed <d.dZd/dZd0dZd1dZd2dZd3dZ d4d"Z d5d%Z d6d7d*Z e d8d+Ze d9d,Z d6d:d-Zd&S);_ShuffleRunManagerboolclosedzdict[ShuffleId, ShuffleRun] _active_runszset[ShuffleRun]_runszdict[ShuffleId, int]_stale_run_idszasyncio.Condition_runs_cleanup_conditionShuffleWorkerPlugin_pluginpluginreturnNonecd|_i|_t|_i|_t j|_||_dS)NF) rrsetrrasyncio Conditionrr )selfr!s Blib/python3.11/site-packages/distributed/shuffle/_worker_plugin.py__init__z_ShuffleRunManager.__init__/s@ UU  '.'8':':$ dict[ShuffleId, Any]cHd|jDS)Nc>i|]\}}||S) heartbeat).0id shuffle_runs r) z0_ShuffleRunManager.heartbeat..8s7   ,;B B %%''   r+)ritemsr(s r)r0z_ShuffleRunManager.heartbeat7s2  ?C?P?V?V?X?X    r+ shuffle_idrrun_idintmessagestrc|j||}||kr ||j|<|j|d}| |j|krdS|j|t |}|||jj j |j |dSN) r setdefaultrgetr8pop RuntimeErrorfailr worker_ongoing_background_tasks call_soonclose)r(r7r8r: stale_run_idr3 exceptions r)rBz_ShuffleRunManager.fail<s*55j&II & .4D  +'++J==  +"4">"> F j))) )) ### 5?? KXXXXXr+r3rc *Kt5 |d{V|j4d{V|j||jdddd{Vn#1d{VswxYwYni#|j4d{V|j||jdddd{Vw#1d{VswxYwYwxYwddddS#1swxYwYdSr=)rrFrrremove notify_all)r(r3s r)rFz_ShuffleRunManager.closeJsl \\ > > >!'')))))))))7>>>>>>>>J%%k2220;;===>>>>>>>>>>>>>>>>>>>>>>>>>>>47>>>>>>>>J%%k2220;;===>>>>>>>>>>>>>>>>>>>>>>>>>>>>  > > > > > > > > > > > > > > > > > >snDB D4B.D B D B DC8 4C& C8& C0 0C83C0 4C88DD D c~KjrJd_jrMj\}}jjjj|jMj4d{Vj fdd{Vdddd{VdS#1d{VswxYwYdS)NTcj Sr=)rr6sr)z-_ShuffleRunManager.teardown..^s DJr+) rrpopitemr rCrDrErFrwait_for)r(_r3s` r)teardownz_ShuffleRunManager.teardownSs;  !.6688NA{ L  9 C C K     / P P P P P P P P.778N8N8N8NOO O O O O O O O P P P P P P P P P P P P P P P P P P P P P P P P P P P P P Ps5$B,, B69B6c`K|j|d}| |j|kr||d{V}|j|krt d|d||j|krt d|d||jrt |d|jr|j|S)aGet the shuffle matching the ID and run ID. If necessary, this method fetches the shuffle run from the scheduler plugin. Parameters ---------- shuffle_id Unique identifier of the shuffle run_id Unique identifier of the shuffle run Raises ------ KeyError If the shuffle does not exist RuntimeError If the run_id is stale N)r7zrun_id=z stale, got z invalid, got  has already been closed)rr?r8_refreshrArr _exceptionr(r7r8r3s r)get_with_run_idz"_ShuffleRunManager.get_with_run_id`s&'++J==  +"4v"="= $ %!.!!K   & &D&DD{DDEE E  & ( (F&FFFFGG G ; H$%F%F%FGG G  ! )( (r+specrkeycK|j|jd}|#||j||d{V}|jrt |d|jr|j|S)aGet or create a shuffle matching the ID and data spec. Parameters ---------- shuffle_id Unique identifier of the shuffle type: Type of the shuffle operation key: Task key triggering the function Nr7rYrZrT)rr?r2rUrrrV)r(rYrZr3s r) get_or_createz _ShuffleRunManager.get_or_creates'++DGT::   $ 7!.!!K ; H$%F%F%FGG G  ! )( (r+run_ids Sequence[int]cZK||t|d{VS)aGet the shuffle matching the ID and most recent run ID. If necessary, this method fetches the shuffle run from the scheduler plugin. Parameters ---------- shuffle_id Unique identifier of the shuffle run_ids Sequence of possibly different run IDs Raises ------ KeyError If the shuffle does not exist RuntimeError If the most recent run_id is stale r7r8N)rXmax)r(r7r^s r)get_most_recentz"_ShuffleRunManager.get_most_recents9*))ZG )UUUUUUUUUr+NShuffleSpec | None str | NonercPK|<|jjj||jjjd{V}nI|jjjt |||jjjd{V}t|t r|j}|S)N)r2rC)rYrZrC) r rC scheduler shuffle_getaddressshuffle_get_or_creater isinstancedata)r(r7rYrZresults r)_fetchz_ShuffleRunManager._fetchs <<.8DD|*2EFF  <.8NNd^^|*2OF fh ' ' ! r+c KdSr=r/)r(r7s r)rUz_ShuffleRunManager._refreshs r+c KdSr=r/)r(r7rYrZs r)rUz_ShuffleRunManager._refreshs  r+cDK||||d{V}|jrt|d|j|dx}r8|j|jkr|S|||j|d|j|j|d}|%||jkrtd|jd||j |j|j |j }||j|<|j ||S)Nr\rTz stale, expected run_id==z'Received stale shuffle run with run_id=z; expected run_id > )rnrrrr?r8rBrrArYcreate_run_on_worker worker_forr radd)r(r7rYrZrmexistingrGr3s r)rUz_ShuffleRunManager._refreshsf {{jt{MMMMMMMM ; H$%F%F%FGG G(,,Z>> >8 &-// OKKFMKK *..z4@@  #  (E(E5&-55&255 k66 M6,dl  )4*% {###r+)r!rr"r#r"r,r7rr8r9r:r;r"r#)r3rr"r#)r"r#r7rr8r9r"rrYrrZr;r"r)r7rr^r_r"r)NN)r7rrYrdrZrer"r)r7rr"r)r7rrYrrZr;r"r)r7rrYrdrZrer"r)__name__ __module__ __qualname____annotations__r*r0rBrFrRrXr]rcrnrrUr/r+r)rr#sLLL---- )(((....        Y Y Y Y>>>> P P P P""""H4VVVV4$( .   X    X $( r+rceZdZUdZded<ded<ded<ded<d ed <d3d Zd4dZd4dZd5dZd6dZ d7dZ d8dZ d9d#Z d:d&Z d;d(Zdd?d2Zd/S)@raInterface between a Worker and a Shuffle. This extension is responsible for - Lifecycle of Shuffle instances - ensuring connectivity between remote shuffle instances - ensuring connectivity and integration with the scheduler - routing concurrent calls to the appropriate `Shuffle` based on its `ShuffleID` - collecting instrumentation of ongoing shuffles and route to scheduler/worker rrCr shuffle_runsrmemory_limiter_commsmemory_limiter_diskrrr"r#c|j|jd<|j|jd<|j|jd<||jd<||_t||_ttd|_ ttd|_ d|_ t|jjj|_dS)Nshuffle_receiveshuffle_inputs_donez shuffle-failshufflez100 MiBz1 GiBF)rhandlersr shuffle_failstream_handlers extensionsrCrrrr rrrrstatenthreads _executorr(rCs r)setupzShuffleWorkerPlugin.setups-1-A)*151I-.151B~.'+)$ .t44$3K 4J4J$K$K!#2;w3G3G#H#H  +DK,=,FGGr+r;c d|jjS)NzShuffleWorkerPlugin on )rCrir6s r)__str__zShuffleWorkerPlugin.__str__s>)<>>>r+c2d|jjd|jdS)Nz)rC address_saferr6s r)__repr__zShuffleWorkerPlugin.__repr__ s$bt{/GbbTXT_bbbbr+r,c4|jSr=)rr0r6s r)r0zShuffleWorkerPlugin.heartbeat's **,,,r+r7rr8r9rllist[tuple[int, Any]] | bytescxK|||d{V}||d{VdS)z Handler: Receive an incoming shard of data from a peer worker. Using an unknown ``shuffle_id`` is an error. N)_get_shuffle_runreceive)r(r7r8rlr3s r)rz#ShuffleWorkerPlugin.shuffle_receive*s[!11*fEEEEEEEE !!$'''''''''''r+cKt5|||d{V}|d{VddddS#1swxYwYdS)z Handler: Inform the extension that all input partitions have been handed off to extensions. Using an unknown ``shuffle_id`` is an error. N)rr inputs_donerWs r)rz'ShuffleWorkerPlugin.shuffle_inputs_done7s \\ , , $ 5 5j& I IIIIIIIK))++ + + + + + + + , , , , , , , , , , , , , , , , , ,s7AAAr:c@|j|||dS)aCFails the shuffle run with the message as exception and triggers cleanup. .. warning:: To guarantee the correct order of operations, shuffle_fail must be synchronous. See https://github.com/dask/distributed/pull/7486#discussion_r1088857185 for more details. )r7r8r:N)rrB)r(r7r8r:s r)rz ShuffleWorkerPlugin.shuffle_fail@s( *VWUUUUUr+r partition_id int | NDIndexrYrkwargsc L||}|jd||d|S)N)rlrr/)get_or_create_shuffle add_partition)r(rlrrYrr3s r)rz!ShuffleWorkerPlugin.add_partitionKsH0066 ({( %     r+r^r_c~K|j||d{V}||d{VS)z Task: Note that the barrier task has been reached (`add_partition` called for all input partitions) Using an unknown ``shuffle_id`` is an error. Calling this before all partitions have been added is undefined. N)rrcbarrier)r(r7r^r3s r)_barrierzShuffleWorkerPlugin._barrierYsZ!-==j'RRRRRRRR !((111111111r+rcJK|j||d{VS)Nra)rrXr(r7r8s r)rz$ShuffleWorkerPlugin._get_shuffle_runesK &66!&7         r+rZcJK|j||d{VS)N)rYrZ)rr]r(rYrZs r)_get_or_create_shufflez*ShuffleWorkerPlugin._get_or_create_shufflens6 &44$C4HHHHHHHHHr+cK|jrJd|_|jd{V |jddS#t $r|jYdSwxYw)NT)cancel_futures)rrrRrshutdown Exceptionrs r)rRzShuffleWorkerPlugin.teardownus; ((********* & N # #4 # 8 8 8 8 8 & & & N # # % % % % % % &sA#A76A7cHt|jj|j||}|Sr=)rrCloopr)r(r7r^rms r)rzShuffleWorkerPlugin.barriers!dk& z7KK r+cNt|jj|jj||Sr=)rrCrrrXrs r)get_shuffle_runz#ShuffleWorkerPlugin.get_shuffle_runs-  K    -      r+cftj}t|jj|jj||Sr=)r rZrrCrrr]rs r)rz)ShuffleWorkerPlugin.get_or_create_shuffles5 K    +      r+Nmetapd.DataFrame | Nonecv|||}tj}||||S)z Task: Retrieve a shuffled output partition from the ShuffleWorkerPlugin. Calling this for a ``shuffle_id`` which is unknown or incomplete is an error. )rrZr)rr rZget_output_partition)r(r7r8rrr3rZs r)rz(ShuffleWorkerPlugin.get_output_partitionsG**:v>> //%0   r+)rCrr"r#)r"r;rv)r7rr8r9rlrr"r#)r7rr8r9r"r#rw) rlrrrrYrrrr"r9)r7rr^r_r"r9rxry)rYrr"rr=) r7rr8r9rrrrr"r)rzr{r|__doc__r}rrrr0rrrrrrrrRrrrrr/r+r)rrs  NNN$$$$))))((((LLL H H H H????cccc---- ( ( ( (,,,, V V V V      2 2 2 2    IIII&&&&          "%)        r+r)* __future__rr&loggingcollections.abcrconcurrent.futuresrtypingrrr dask.contextr dask.utilsr distributed.diagnostics.pluginr distributed.protocol.serializer distributed.shuffle._corer rrrrdistributed.shuffle._exceptionsrdistributed.shuffle._limiterrdistributed.utilsrrpandaspddistributed.workerr getLoggerrzloggerrrr/r+r)rs""""""$$$$$$111111//////////%%%%%%""""""777777333333?>>>>>888888........*))))))  8 $ $VVVVVVVVru u u u u ,u u u u u r+