0Fie, UddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z mZmZddlmZmZmZmZmZmZmZmZmZddlmZddlmZmZddlmZm Z m!Z!m"Z"ddl#m$Z$dd l%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,dd l-m.Z.ddl/Z/dd l0m1Z1dd l2m3Z3m4Z4m5Z5dd l6m7Z7ddl8m9Z9ddl:m;Z;ddlZ>ddl?m@Z@mAZAmBZBddlCmDZDddlEmFZFmGZGddlHmIZJddlKmLZLejMdZNe&r'ddlOmPZPmQZQmRZReQdZSddlTmUZUddlVmWZWddlXmYZYe)dZZde[d<hdZ\d e[d!<d"d#hZ]d e[d$<hd%Z^d e[d&<d'Z_d(Z`Gd)d*e*ZaGd+d,ebZcGd-d.ecZdGd/d0ebZeGd1d2ebZfedd5Zge jhd6krd7d8iniZiedd9d9d:eiGd;d<ZjeGd=d>ZkGd?d@ZleGdAdBekZmeGdCdDekZneGdEdFekZoGdGdHekZpeGdIdJepZqeGdKdLepZreGdMdNepZseGdOdPepZteGdQdRepZueGdSdTepZveGdUdVepZweGdWdXepZxeGdYdZZyeGd[d\eyZzeGd]d^eyZ{eGd_d`eyZ|eGdadbeyZ}eGdcdde}Z~eGdedfe}ZeGdgdhe}ZeGdidje}ZeGdkdleyZeGdmdneyZeGdodpeyZeGdqdreZeGdsdteZeGdudveZeGdwdxeyZeGdydzeyZeGd{d|eyZeGd}d~eyZeGddeyZeGddeyZeGddeyZeGddeyZeGddeyZeeje+eZeffZde[d<eekZde[d<eeefZde[d<ddZGddZGddejZGddZGddZdS)) annotationsN)Counter defaultdictdeque) AwaitableCallable Collection ContainerHashableIteratorMappingMutableMappingSet)copy) dataclassfield) lru_cachepartialsingledispatchmethodwraps)chain) TYPE_CHECKINGAnyClassVarLiteral TypedDictUnioncast)peekn)Key) key_split parse_bytestypename) worker_story)HeapSet)get_address_host) ErrorMessage error_message)DelayedMetricsLedger monotonictime)pickle) SerializeToPickle) safe_sizeof)recursive_to_dictz distributed.worker.state_machine) NotRequired ParamSpec TypeAliasP) WorkerPlugin) T_runspec)Worker) cancelled constrainederror executingfetchflight forgotten long-runningmemorymissingreadyreleased rescheduledresumedwaitingr3TaskStateState>rBrFr;r9r?zSet[TaskStateState] PROCESSINGrBr9READY> r<rBr=rArErFr;r9r?WAITING_FOR_DATAz--no-value-sentinel--c8eZdZUded<ded<ded<ded<dS) StartStopzHLiteral['compute', 'transfer', 'disk-read', 'disk-write', 'deserialize']actionfloatstartstopzNotRequired[str]sourceN)__name__ __module__ __qualname____annotations__@lib/python3.11/site-packages/distributed/worker_state_machine.pyrMrMks<TTTTLLLKKKrXrMc2eZdZddZdd Zdd ZeZddZdS)InvalidTransitionkeyr rPrGfinishstory list[tuple]c>||_||_||_||_dSNr\rPr]r^)selfr\rPr]r^s rY__init__zInvalidTransition.__init__ss$   rXreturntuple[Callable, tuple]cTt||j|j|j|jffSra)typer\rPr]r^rcs rY __reduce__zInvalidTransition.__reduce__s$DzzDHdj$+tzJJJrXstrc|jjd|jd|jd|jdzdzdt t|jzS)N:  :: -> Story:  ) __class__rSr\rPr]joinmaprkr^ris rY__repr__zInvalidTransition.__repr__se~& U U$( U U$* U U U U  mmCTZ0011 2 rXtuple[str, dict[str, Any]]c<d|j|j|j|jdfS)Nzinvalid-worker-transitionrbrbris rYto_eventzInvalidTransition.to_events- 'x+     rXN)r\r rPrGr]rGr^r_rerfrerkrerwrSrTrUrdrjrv__str__ryrWrXrYr[r[rso    KKKK    G       rXr[c eZdZdfd ZxZS)TransitionCounterMaxExceededrerwcRt\}}d|fS)Nztransition-counter-max-exceeded)superry)rctopicmsgrss rYryz%TransitionCounterMaxExceeded.to_events'WW%%'' s0#55rXr|)rSrTrUry __classcell__rss@rYrrs=6666666666rXrc2eZdZddZdd Zdd ZeZddZdS)InvalidTaskStater\r staterGr^r_c0||_||_||_dSrar\rr^)rcr\rr^s rYrdzInvalidTaskState.__init__s   rXrerfcHt||j|j|jffSra)rhr\rr^ris rYrjzInvalidTaskState.__reduce__s DzzDHdj$*===rXrkc|jjd|jd|jdzdzdt t |jzS)Nrmrnrprqrr)rsrSr\rrtrurkr^ris rYrvzInvalidTaskState.__repr__s[~& F F$( F F$* F F  mmCTZ0011 2 rXrwc0d|j|j|jdfS)Nzinvalid-worker-task-staterrris rYryzInvalidTaskState.to_events( 'x    rXN)r\r rrGr^r_rzr{r|r}rWrXrYrrsk>>>>    G      rXrceZdZdZdS)RecommendationsConflictzOTwo or more recommendations for the same task suggested different finish statesN)rSrTrU__doc__rWrXrYrrsYYYYrXrreintcZttjdS)Nz'distributed.scheduler.default-data-size)r"daskconfiggetrWrXrY_default_data_sizers t{'PQQ R RRrX) slotsTF)repreqceZdZUdZded<edZded<eZded <d Z d ed <ee Z ded<ee Z ded<ee Z ded<ee Zded<dZded<d Zded<d Zded<d Zded<d Zded<ee Zded<d Zd ed!<ee Zd"ed#<d Zd$ed%<d Zd$ed&<d'Zded(<d'Zded)<d Zd*ed+<d,Zded-<ee Zd.ed/<d Z ded0<d Z!ded1<ee Z"d2ed3<d Z#d4ed5<d Z$d6ed7<d Z%d ed8<dZ&d9ed:<e'j(Z)d;ed<<edZ*d=ed><dJdAZ+dKdBZ,dLdCZ-dLdDZ.dEdFdMdIZ/d S)N TaskStatezHolds volatile state relating to an individual Dask task. Not to be confused with :class:`distributed.scheduler.TaskState`, which holds similar information on the scheduler side. r r\F)initrkprefixrrun_idNT_runspec | Nonerun_spec)default_factoryset[TaskState] dependencies dependentswaiting_for_datawaitersrCrGrz4Literal['executing', 'long-running', 'flight', None]previousz!Literal['fetch', 'waiting', None]next float | Nonedurationztuple[int, ...] | Nonepriorityset[str]who_has str | None coming_fromdict[str, float]resource_restrictionsSerialize | None exception tracebackexception_texttraceback_text type | Nonerhrsuspicious_countlist[StartStop] startstops start_time stop_timedictmetadata int | Nonenbytes dict | Nonerspan_idbooldonez$ClassVar[weakref.WeakSet[TaskState]] _instancesr __weakref__reNonecvtj|t|j|_dSra)rraddr!r\rris rY __post_init__zTaskState.__post_init__$s.  &&&)) rXc|jdkr d|jd}n&|jdkrd|jd|jd}n|j}d|jd|d S) Nr8z cancelled()rEzresumed(roz )rrrr\rcrs rYrvzTaskState.__repr__(sp : $ $1111EE Z9 $ $z.TaskState._to_dict_no_nest..Ys(DDDADa8mm1mmmrX)r0items)rcrouts rY_to_dict_no_nestzTaskState._to_dict_no_nestFs4" gtDDDDDDDDDrXrerr{rerrrrer)0rSrTrUrrVrrRUN_ID_SENTINELrrsetrrrrrrrrrrrrrrrrrrhrlistrrrrrrrrweakrefWeakSetrrrrvrrrrWrXrYrrs\ HHH%U###F####!F!!!! "&H%%%%$)5#=#=#=L====!&s!;!;!;J;;;;',uS'A'A'AAAAA#eC888G8888'E&&&&FJHIIII.2D2222"H!!!!'+H++++c222G2222"K"""".3eD.I.I.IIIII"&I&&&&"&I&&&&NND"'%"="="=J====#J####"I""""U4000H0000F#K####GD7Fw7H7HJHHHHu%(((K((((****3333"FFFF=?EEEEEEEErXrcLeZdZUdZdZded<edd Zdfd ZxZ S) InstructionzLCommand from the worker state machine to the Worker, in response to an event stimulus_idrkrkwargsrre_InstructionMatchc t|fi|S)aGenerate a partial match to compare against an Instruction instance. The typical usage is to compare a list of instructions returned by :meth:`WorkerState.handle_stimulus` or in :attr:`WorkerState.stimulus_log` vs. an expected list of matches. Examples -------- .. code-block:: python instructions = ws.handle_stimulus(...) assert instructions == [ TaskFinishedMsg.match(key="x"), ... ] )rclsrs rYmatchzInstruction.matchcs$!/////rXotherobjectrc|t|tr||kSt|Sra) isinstancerr__eq__)rcrrss rYrzInstruction.__eq__ws8 e. / / )D= 77>>%(( (rX)rrrerrrrer) rSrTrUr __slots__rV classmethodrrrrs@rYrr\stVV I000[0&))))))))))rXrc@eZdZUdZded<ded<ddZdd ZddZdS)rz_Utility class, to be used to test an instructions list. See :meth:`Instruction.match`. type[Instruction]rdict[str, Any]rrc "||_||_dSrar)rcrrs rYrdz_InstructionMatch.__init__s rXrerkc|jj}dd|jD}|d|dS)N, c3*K|]\}}|d|VdS)=NrWrs rY z-_InstructionMatch.__repr__..s0JJda!zzazzJJJJJJrX(z) (partial match))rrSrtrr)rccls_str kwargs_strs rYrvz_InstructionMatch.__repr__sQ(#YYJJdk6G6G6I6IJJJJJ 99J9999rXrrrct|jurdStfd|jDS)NFc3FK|]\}}t||kVdSragetattr)rrrrs rYr z+_InstructionMatch.__eq__..s6JJda75!$$)JJJJJJrX)rhrallrr)rcrs `rYrz_InstructionMatch.__eq__sL ;;dh & &5JJJJdk6G6G6I6IJJJJJJrXN)rrrrr{r)rSrTrUrrVrdrvrrWrXrYrrs|:::: KKKKKKrXrc2eZdZUdZded<ded<ded<dS) GatherDep)worker to_gather total_nbytesrkrzset[Key]rrrNrSrTrUrrVrWrXrYrrs87IKKKrXrceZdZUdZded<dS)Executer\r r\NrrWrXrYrrI HHHHHrXrceZdZUdZded<dS)RetryBusyWorkerLaterrrkrNrrWrXrYrrIKKKKKrXrc&eZdZUded<dZddZdS) SendMessageToSchedulerz ClassVar[str]oprWrercVfdjD}j|d<j|d<|S)z@Convert object to dict so that it can be serialized with msgpackc2i|]}|t|SrWr)rrrcs rYrz2SendMessageToScheduler.to_dict..s% ? ? ?QQa  ? ? ?rXr"r)rVr"r)rcds` rYto_dictzSendMessageToScheduler.to_dicts< ? ? ? ?$*> ? ? ?'$+-rXNrer)rSrTrUrVrr&rWrXrYr!r!s;IrXr!ceZdZUdZded<ded<ded<ded <d ed <d ed <ded<ded<eeZdfd ZxZS)TaskFinishedMsg task-finishedr r\rrrrbytesrhrkr#rrthreadrrrercRt}d|d<|S)NOKstatusrr&rcr%rss rYr&zTaskFinishedMsg.to_dicts$ GGOO  ( rXr') rSrTrUr"rVtuplerr&rrs@rYr)r)s B HHHKKKKKKMMMNNNo&&IrXr)ceZdZUdZded<ded<ded<ded <d ed <d ed <d ed<ded<eeZdfd Ze dddZ xZ S) TaskErredMsgz task-erredr r\rrr-rrrrkrrrr,rrrercRt}d|d<|S)Nr:r/r0r1s rYr&zTaskErredMsg.to_dicts$ GGOO  ( rXNtsrrc |jsJt|j||j|j|j|j||j| S)N) r\rrrrrr,rr)rr4r\rrrr)r6rrr,s rY from_taskzTaskErredMsg.from_tasksR|ll,,}#    rXr'ra) r6rrrrrkr,rrer4) rSrTrUr"rVr2rr& staticmethodr8rrs@rYr4r4s B HHHKKKo&&I KO    \     rXr4c"eZdZUdZdZded<dS)ReleaseWorkerDataMsgzrelease-worker-datarr r\NrSrTrUr"rrVrWrXrYr;r;s# BI HHHHHrXr;c"eZdZUdZdZded<dS) RescheduleMsg reschedulerr r\Nr<rWrXrYr>r>s# BI HHHHHrXr>c,eZdZUdZdZded<ded<dS)LongRunningMsgr?r\compute_durationr r\rrCNr<rWrXrYrArAs1 B+I HHH""""""rXrAc"eZdZUdZdZded<dS) AddKeysMsgadd-keyskeysCollection[Key]rHNr<rWrXrYrErE s( BIrXrEc&eZdZUdZdZdZded<dS)RequestRefreshWhoHasMsga{Worker -> Scheduler asynchronous request for updated who_has information. Not to be confused with the scheduler.who_has synchronous RPC call, which is used by the Client. See also -------- RefreshWhoHasEvent distributed.scheduler.Scheduler.request_refresh_who_has distributed.client.Client.who_has distributed.scheduler.Scheduler.get_who_has zrequest-refresh-who-hasrGrIrHNrSrTrUrr"rrVrWrXrYrKrKs4   #BIrXrKc0eZdZUdZdZdZded<ded<dS) StealResponseMsgzkWorker->Scheduler response to ``{op: steal-request}`` See also -------- StealRequestEvent zsteal-response)r\rr r\zTaskStateState | NonerNrLrWrXrYrNrN&s= B I HHH      rXrNcveZdZUdZdZded<iZded<dd Zdd ZddZ ddddZ e ddZ ddZ dS) StateMachineEventzDBase abstract class for all stimuli that can modify the worker state)rhandledrkrz,ClassVar[dict[str, type[StateMachineEvent]]]_classesargsrrrecHt|}d|_|S)z@H???? 777757888888$   \ UUUUUUrXrPceZdZdZdS) PauseEventrWNrSrTrUrrWrXrYrmrmwIIIrXrmceZdZdZdS) UnpauseEventrWNrnrWrXrYrqrq|rorXrqceZdZUdZded<dS)RetryBusyWorkerEventrrkrNrrWrXrYrsrsrrXrsc,eZdZUdZdZded<ded<dS)GatherDepDoneEventz?:class:`GatherDep` instruction terminated (abstract base class))rrrkrrrNrSrTrUrrrVrWrXrYrurus2II*IKKKrXruc2eZdZUdZdZded<d d Zdd Zd S)GatherDepSuccessEventzV:class:`GatherDep` instruction terminated: remote worker fetched successfully datadict[Key, object]rzrQrOrerPc^t|}||_d|jD|_|S)Nci|]}|dSrarWrrs rYrz5GatherDepSuccessEvent.to_loggable..s///At///rX)rrQrzrcrQrs rYr[z!GatherDepSuccessEvent.to_loggables04jj //TY/// rXrc2d|jD|_dS)Nci|]}|dSrarWr~s rYrz:GatherDepSuccessEvent._after_from_dict..s000Q000rXryris rYrfz&GatherDepSuccessEvent._after_from_dicts00di000 rXNrkr)rSrTrUrrrVr[rfrWrXrYrxrxs[I 111111rXrxceZdZdZdZdS)GatherDepBusyEventzI:class:`GatherDep` instruction terminated: remote worker is busy rWNrSrTrUrrrWrXrYrrIIIrXrceZdZdZdZdS)GatherDepNetworkFailureEventzr:class:`GatherDep` instruction terminated: network failure while trying to communicate with remote worker rWNrrWrXrYrrrrXrcreZdZUdZded<ded<ded<ded<eeZdd ZeddZ dS)GatherDepFailureEventzclass:`GatherDep` instruction terminated: generic error raised (not a network failure); e.g. data failed to deserialize. r-rrrrkrrrercTtt|_d|_dSrar- Exceptionrrris rYrfz&GatherDepFailureEvent._after_from_dict "9;;//rXerr BaseExceptionrrrrc tt|}||||d|d|d|d|S)Nrrrr)rrrrrrr)r()rrrrrrs rYfrom_exceptionz$GatherDepFailureEvent.from_exceptionsUC  s%+&+&/0/0#    rXNr) rrrrkrrrrkrer) rSrTrUrrVr2rrfrrrWrXrYrrso&&I   [   rXrceZdZUded<dZdS)RemoveWorkerEventrkrrN)rSrTrUrVrrWrXrYrrsKKKIIIrXrc eZdZUded<ded<ded<ded<d ed <d ed <d ed<ded<ded<ded<ded<eeZd4dZddd5dZd6d Zd7d"Z d4d#Z e d4d$Z e d8d&Z ed'd(d(d)d*d(d+d(d,d9d3Zd(S):ComputeTaskEventr r\rrdict[Key, Collection[str]]rdict[Key, int]rtuple[int, ...]rrOrrrrrractorrrrrrerct|jtrt|j|_t|jt r|jj|_dSdSra)rrrr2rr.rzris rYrzComputeTaskEvent.__post_init__sW dmT * * 1!$-00DM dmX . . /!M.DMMM / /rXrWrrrc^t||S)Nr)rPrc_clean)rcrs rYrczComputeTaskEvent._to_dicts# ))$++--)IIIrXrPc2t|}d|_|Sra)rr)rcrs rYrzComputeTaskEvent._cleans4jj  rXrQc<|}||_|Sra)rrQrs rYr[zComputeTaskEvent.to_loggableskkmm  rXcd|_dSra)rris rYrfz!ComputeTaskEvent._after_from_dicts  rXcdSrarWrWs rY_fzComputeTaskEvent._f srXtuple[Callable, tuple, dict]c|jdifS)NrW)rrWs rY dummy_runspeczComputeTaskEvent.dummy_runspecsBrXrNr?F)rrrrrrrr!dict[Key, Collection[str]] | Nonedict[Key, int] | Nonedict[str, float] | Nonerrrkc t|||pi|p d|pdD||t|pi||pid|  S)Build a dummy event, with most attributes set to a reasonable default. This is a convenience method to be used in unit testing only. ci|]}|dSrWr~s rYrz*ComputeTaskEvent.dummy..'s:::qa:::rXrWN) r\rrrrrrrrrrr)rr) r\rrrrrrrrrs rYdummyzComputeTaskEvent.dummysq" Mr:::GMr:::%3355"7"=2#)r#    rXrrrerPrk)rer)r\r rrrrrrrrrrOrrrrrrrrkrer)rSrTrUrVr2rrrcrr[rfrrrr9rrWrXrYrrs HHHKKK''''OOO++++KKKo&&I////57JJJJJJ  [   [ 59(,$(9=#'     \   rXrc"eZdZUdZded<dZdS)ExecuteDoneEventz\Abstract base event for all the possible outcomes of a :class:`Compute` instruction r r\rN)rSrTrUrrVrrWrXrYrr3s* HHHIIIrXrceZdZUded<ded<ded<ded<ded<d ed <eeZddZddd fdZd!dZe d"dddd#dZ xZ S)$ExecuteSuccessEventrrrvaluerOrPrQrrrhrQrerPc@t|}||_d|_|Sra)rrQrrs rYr[zExecuteSuccessEvent.to_loggableGs 4jj   rXrWrrrrct|}d|vrt|j|d<|S)Nrrh)rrcrkrh)rcrr%rss rYrczExecuteSuccessEvent._to_dictMs> GG  W  - -  DIAfIrXrc"d|_d|_dSra)rrhris rYrfz$ExecuteSuccessEvent._after_from_dictTs  rXNr)rrr\r rrkc 0t|||dd|d|S)rgrN)r\rrrPrQrrhr)r)r\rrrrs rYrzExecuteSuccessEvent.dummyXs3##    rXrkrrra) r\r rrrrrrrrkrer) rSrTrUrVr2rr[rcrfr9rrrs@rYrr=sKKKMMMLLLKKKKKKo&&I 57       \     rXrceZdZUded<ded<ded<ded<ded <d ed <d ed <eeZddZedddddZe ddddZ dS)ExecuteFailureEventrrrrPrQr-rrrrkrrrercTtt|_d|_dSrarris rYrfz$ExecuteFailureEvent._after_from_dict{rrXN)rPrQ err_or_msgBaseException | ErrorMessager\r rc t|tr|}nt|}||||||d|d|d|d| S)Nrrrr r\rrPrQrrrrr)rrr()rrr\rrPrQrrs rYrz"ExecuteFailureEvent.from_exceptionst j$ ' ' ,CC ++Cs+&+&/0/0#    rXr)rc Lt||ddtdddd| S)rNrr)rr-)r\rrs rYrzExecuteFailureEvent.dummys<#oo#    rXr)rrr\r rrrPrrQrrrkrer)r\r rrrrkrer) rSrTrUrVr2rrfrrr9rrWrXrYrrpsKKKo&&I#!     [ 6     \   rXrc*eZdZdZed dZdS) RescheduleEventrWr\r rrkrec$t||S)zqBuild an event. This method exists for compatibility with the other ExecuteDoneEvent subclasses. r\r)rrs rYrzRescheduleEvent.dummys 3K@@@@rXN)r\r rrkrer)rSrTrUrr9rrWrXrYrrs=IAAA\AAArXrceZdZUdZded<dS)CancelComputeEventrr r\NrrWrXrYrrrrXrceZdZdZdS)FindMissingEventrWNrnrWrXrYrrrorXrc"eZdZUdZdZded<dS)RefreshWhoHasEventzScheduler -> Worker message containing updated who_has information. See also -------- RequestRefreshWhoHasMsg )rrrNrvrWrXrYrrs/I''''''rXrc(eZdZUdZded<ded<dS)AcquireReplicasEvent)rrrrrrNrrWrXrYrrs/%I''''rXrceZdZUdZded<dS)RemoveReplicasEventrGrIrHNrrWrXrYrr#IrXrceZdZUdZded<dS) FreeKeysEventrGrIrHNrrWrXrYrrrrXrc"eZdZUdZdZded<dS)StealRequestEventzEvent that requests a worker to release a key because it's now being computed somewhere else. See also -------- StealResponseMsg rr r\NrvrWrXrYrrs*I HHHHHrXrc&eZdZUdZded<d dZd S) UpdateDataEventryr{rzrQrOrerPczt|}||_t|j|_|Sra)rrQrfromkeysrzrs rYr[zUpdateDataEvent.to_loggables.4jj ==++ rXNrk)rSrTrUrrVr[rWrXrYrrs;IrXrc(eZdZUdZded<ded<dS) SecedeEventrBr r\rOrCNrrWrXrYrrs,+I HHHrXrRecs Instructions RecsInstrsrSc i}g}|D]\\}}|D]=\}}||vr/|||kr#td|jd||d||||<>||z }]||fS)z|Merge multiple (recommendations, instructions) tuples. Collisions in recommendations are only allowed if identical. zMismatched recommendations for rmz vs. )rrr\)rSrecsinstrrecs_iinstr_ir6r]s rYmerge_recs_instructionsrsDE ,,..  JBTzzd2h&00-YbfYY$r(YYQWYYDHH  ;rXc  eZdZUdZded<ded<ded<ded <d ed <d ed <d ed<ded<ded<ded<ded<ded<ded<ded<ded<ded<ded<ded <ded!<ded"<ded#<d$ed%<ded&<d'ed(<d'ed)<ded*<ded+<ded,<ded-<d.ed/<d0ed1<d2ed3<ded4<d5ed6<ded7<d8ed9<ded:<d;ed<<eeZd=d>d>d>d>d>d?d@dAejejdB d1dIZ d2dNZ e d3dOZ e d4dPZ e d3dQZe d3dRZd5dYZd6d]Zd7d_Zd8d`Zd9dbZd:ddZd;dgZddlZd?doZd@dpZd@dqZd@drZd@dsZd@dtZd@duZ d@dvZ!d@dwZ"d@dxZ#dAdzZ$d@d{Z%d@d|Z&d@d}Z'd@d~Z(d@dZ)d@dZ*dBdZ+dBdZ,d@dZ-d@dZ.d@dZ/d@dZ0d@dZ1d@dZ2dCdZ3d@dZ4d@dZ5d@dZ6d@dZ7d@dZ8dDdZ9dDdZ:dEdZ;dEdZ<dEdZ=dEdZ>dFdZ?d@dZ@ide3de1de:de3de3de3de3de2de,de.de:de>de0de-de5de"de"ide+de9de;de4de)de$de%de&de+de7de=dede8de+de;de)de4ide'de+dededede6de"de+de!de@de<de de#de(de*de"ZAded<dGdڄZBdHd݄ZCdIdބZDd7d߄ZEd7dZFdJdZGeHdKdZIeIjJdLdZKeIjJdMdZLeIjJdNdZMeIjJdOdZNeIjJdPdZOdQdZPeIjJdRdZQeIjJdSdZReIjJdTdZSeIjJdUdZTeIjJdVdZUeIjJdWdZVeIjJdXdZWeIjJdYdZXeIjJdZdZYeIjJd[dZZeIjJd\dZ[d]d Z\eIjJd^d Z]eIjJd_dZ^eIjJd`dZ_eIjJdadZ`eIjJdbdZadcdZbdddZcddded#Zdd7d$Zed7d%Zfd7d&Zgd7d'Zhd7d(Zid7d)Zjd7d*Zkd7d+Zld7d,Zmd7d-Znd7d.Zodfd/Zpdfd0Zqd>S(g WorkerStateayState machine encapsulating the lifetime of all tasks on a worker. Not to be confused with :class:`distributed.scheduler.WorkerState`. .. note:: The data attributes of this class are implementation details and may be changed without a deprecation cycle. .. warning:: The attributes of this class are all heavily correlated with each other. *Do not* modify them directly, *ever*, as it is extremely easy to obtain a broken state this way, which in turn will likely result in cluster-wide deadlocks. The state should be exclusively mutated through :meth:`handle_stimulus`. rkaddresszdict[Key, TaskState]tasksrthreadszMutableMapping[Key, object]rzzdict[str, WorkerPlugin]pluginsHeapSet[TaskState]rBr9rnthreadsrrunningrrFzdefaultdict[str, set[Key]]has_whatz$defaultdict[str, HeapSet[TaskState]] data_needed fetch_countrOtransfer_message_bytes_limitmissing_dep_flightin_flight_taskszdict[str, set[Key]]in_flight_workerstransfer_incoming_bytestransfer_incoming_count_limittransfer_incoming_count_total*transfer_incoming_bytes_throttle_thresholdr busy_workers generationrtotal_resourcesavailable_resourcesr; long_runningexecuted_countrr{actorsz deque[tuple]logzdeque[StateMachineEvent] stimulus_logvalidate TaskCounter task_countertransition_counterint | Literal[False]transition_counter_maxtransfer_incoming_bytes_limitz random.RandomrngrNi'TF) rrrzrr resourcesrrrrrr"MutableMapping[Key, object] | Nonerdict[str, WorkerPlugin] | Noner Mapping[str, float] | Nonec ||_|r||_||ni|_||ni|_||ni|_|t |ni|_|j|_||_ i|_ d|_ t|_ tt|_tt!t"t$t'jd|_d|_i|_t|_||_d|_t7d|_d|_t|_d|_t#t'jd|_ t#t'jd|_!t|_"t|_#d|_$d|_%t|_&| |_'tPj)*d} tW| |_,tW| |_-t]|_/d|_0| |_1| |_2i|_3tij5d|_6dS)NTrrrgcAz&distributed.admin.low-level-log-length)maxlen)7rrrzrrrrrrrrrrrFrrrr%roperator attrgetterrrrrrrrrrrrrBr9r;rrrrrrrrrrrrrrrrrrandomRandomr) rcrrrzrrr rrrrrrs rYrdzWorkerState.__init__s"!   #"DL!,DD" ")"5ww2 ")"5ww2 2;2GtIR#'#7#<#<#>#>     uu #C(( & GI&H,? ,K,K L L L  !#EE-J*-.*:=d))7'($"%%%!4Z!@!@AAA "x'::'F'FGGG"uu EE,H)!IJJ'''!000'MM"#&<#-J* =##rXstimsrPrerc.g}t}|D]}t|ts.|j||||\}}||z }||||jz }|S)zProcess one or more external events, transition relevant tasks to new states, and return a list of instructions to be executed as a consequence. See also -------- BaseWorker.handle_stimulus rZr) r+rrrappendr[ _handle_event _transitionsr)rcr instructionsrQstimrrs rYhandle_stimuluszWorkerState.handle_stimulus2s && R RDd$455 L!(()9)9')9)J)JKKK,,T22KD% E !L D--d@P-QQ QLLrXc*t|jS)aCount of tasks currently executing on this worker and counting towards the maximum number of threads. It includes cancelled tasks, but does not include long running (a.k.a. seceded) tasks. See also -------- WorkerState.executing WorkerState.executed_count WorkerState.nthreads WorkerState.all_running_tasks )lenr;ris rYexecuting_countzWorkerState.executing_countHs4>"""rXc |j|jzS)ahAll tasks that are currently occupying a thread. They may or may not count towards the maximum number of threads. These are: - ts.status in (executing, long-running) - ts.status in (cancelled, resumed) and ts.previous in (executing, long-running) See also -------- WorkerState.executing_count )r;rris rYall_running_taskszWorkerState.all_running_tasksYs~ 111rXc*t|jS)zNumber of tasks currently being replicated from other workers to this one. See also -------- WorkerState.in_flight_tasks )rrris rYin_flight_tasks_countz!WorkerState.in_flight_tasks_countjs4'(((rXc*t|jS)zCurrent number of open data transfers from other workers. See also -------- WorkerState.in_flight_workers )rrris rYtransfer_incoming_countz#WorkerState.transfer_incoming_countts4)***rXr\r rrrrcf |j|}td||nC#t$r6t |x|j|<}|j|YnwxYw|js |sJ||_|j |d|j |tf|S)Nz+Data task %s already known (stimulus_id=%s)zensure-task-exists) rloggerdebugKeyErrorrrnew_taskrrrrr+)rcr\rrr6s rY_ensure_task_existszWorkerState._ensure_task_existss +CB LLFK X X X X + + +#,S>> 1DJsOb   & &r * * * * * +{ #OOO"BK 2BHk466RSSS s),=A,+A,rMapping[Key, Collection[str]]rc|D]D\}}|j|}|s#t|}|j|vrF||j|jdkr!td|j||j |kr|j |z D]M}|j |||jdkr |j ||N||j z D]M}|j | ||jdkr |j | |N||_ FdS)Nr@zEScheduler claims worker %s holds data for task %s, which is not true.r<) rrrrrremoverr%r&rrrr)rcrr\workersr6rs rY_update_who_haszWorkerState._update_who_hassw#MMOO& !& !LC$$B 'llG|w&&t|,,,8x''LL-  zW$$*w. 8 8 f%,,S1118w&&$V,33B777!BJ. 5 5 f%))#...8w&&$V,00444 BJJM& !& !rXr6cZtd||j|jvr |j|j=|j|jd|j|jd|jD]G}|j| |j|j | |H|j |j D]6}|j ||j |7|j d|_d|_d|_d|_d|_d|_d|_d|_d|_|j ||j ||j ||j ||j ||j ||j |dS)zEnsure that TaskState attributes are reset to a neutral default and Worker-level state associated to the provided key is cleared (e.g. who_has) This is idempotent zPurge task: %sNFr)r%r&r\rzrrerrrdiscardrclearrrrrrrrrrrrrrBr9r;rrrF)rcr6rr%s rY _purge_statezWorkerState._purge_states  %r*** 6TY   "&! %%% &&&j 1 1F M& ! ) )"& 1 1 1  V $ , ,R 0 0 0 0  " "A   ' ' * * * I  b ! ! ! ! !!###     ''+++ 2   $$$ r""" !!"%%% $$R((( R     rXcn|j|jk}|j|jk}|j|jk}|r|p|S)aDecides whether the WorkerState should throttle data transfers from other workers. Returns ------- * True if the number of incoming data transfers reached its limit and the size of incoming data transfers reached the minimum threshold for throttling * True if the size of incoming data transfers reached its limit * False otherwise )r#rrrr)rcreached_count_limitreached_throttle_thresholdreached_bytes_limits rY#_should_throttle_incoming_transfersz/WorkerState._should_throttle_incoming_transferssX  (D,N N   (> ? #  (D,N N #A'AXEXXrXrc v|jr|jsigfS|rigfSi}g}|D]r\}}||jksJ||\}}|s |jsJ|sn9d|D}tdt||t|t|j|j |j t|j |j d|||tf|D].} |jr| jdksJ|| jvsJ| |vsJd|f|| </|t'||||||j|<|xjdz c_|xj|z c_|rnt||fS)zTransition tasks from fetch to flight, until there are no more tasks in fetch state or a threshold has been reached. ch|] }|j SrWrrr6s rY z4WorkerState._ensure_communicating..s???bf???rXz]Gathering %d tasks from %s; %d more remain. Pending workers: %d; connections: %d/%d; busy: %dzgather-dependenciesr<r=)rrrrr)rrr7_select_workers_for_gatherr_select_keys_for_gatherrr%r&rr#rrrrr+rrrrrr) rcrrecommendationsrravailable_tasksto_gather_tasksmessage_nbytesto_gather_keysr6s rY_ensure_communicatingz!WorkerState._ensure_communicatingsU| 4#3 r6M  3 3 5 5 r6M "%' '+'F'F'H'H5 5  #FOT\)))).2.J.J// +O^# Bd&B B B B" ?????N LLDO$$O$$D$%%,2D%&&    HOO& TVVT   & 9 9=58w....!RZ////_4444'/&8##   !,!/ +    .>## #   ."''))C44S99 .'++---trXrr)c |jr=|jdksJ|j|jvs|j|jvsJ|jJ|jJ tj|j}n6#t$r)tjt|j}YnwxYwt|j||j|t|j|j |j |j|j| S)Nr@) r\rrrhr#rr,rr)rrr\rzrrhrr,dumpsrr#r)rrrr)rcr6rrtype_serializeds rY_get_task_finished_msgz"WorkerState._get_task_finished_msgs  = )8x''''6TY&&"&DK*?*?*?*?7&&&9((( >$l2733OO > > >%l8BG+<+<==OOO > 9 bg&&[<##BF++}#    sA 0BBc|js|digfSd|_d|_|xjdz c_|jsJ|jD]"}|j||#igfS)NrAr<Fr)rrrrrrr)rcr6rws rY_transition_generic_fetchz%WorkerState._transition_generic_fetch sz ' ?B& & A{ ( (A  Q  # #B ' ' ' '2v rXc|j||||||SNr)rr0r2_transition_released_waitingrcr6rs rY_transition_missing_waitingz'WorkerState._transition_missing_waitingsH ''+++ "000MMMrXc|jr |jdksJ|jsigfS|j||||S)NrAr)rrrrr0rfrjs rY_transition_missing_fetchz%WorkerState._transition_missing_fetch sd = )8y((((z r6M ''+++--bk-JJJrXc|j||||\}}|j|jvsJ||fSrh)rr0_transition_generic_releasedr\rrcr6rrrs rY_transition_missing_releasedz(WorkerState._transition_missing_released,sa ''+++!>> K?  lv####\!!rXcB|jsJ|||Srh)r_transition_generic_missingrjs rY_transition_flight_missingz&WorkerState._transition_flight_missing6s(w// /LLLrXcz|jr |jrJd|_|j|d|_igfS)NrAF)rrrrrrrjs rYrsz'WorkerState._transition_generic_missing<sJ = "z ! ! ! ##B'''2v rXcX|jr |jdksJ|||S)NrCr)rrrfrjs rY_transition_released_fetchz&WorkerState._transition_released_fetchGs: = *8z))))--bk-JJJrXc||i}|jD](}|js|jtt zdhzvrd||<)d|_|jsd||<|gfS)Nr@rCr>)r2rrrrIrHr)rcr6rr dependencys rYroz(WorkerState._transition_generic_releasedNs "/ . .J& .$EJ,>(,KKK#-Z } #"DHRxrXcjr"tfd|jDsJi}|j|jD]F}|jdkr9|j||j|d||<G|jsd||<d|_j||gfS)Nc34K|]}|jjvVdSra)r\r)rr%rcs rYr z;WorkerState._transition_released_waiting..ds,DDqqu *DDDDDDrXr@r<rBrF) rrrrr1rrrrF)rcr6rr>dep_tss` rYriz(WorkerState._transition_released_waiting`s = EDDDDBODDDDD D D D " !!###o 2 2F|x''#''///""2&&&*1'" *")OB  ""rXrc|jr1|jdksJ|jsJ|jD]}||j|vsJd|_d|_||_|j||xjdzc_igfS)Nr<Fr=r) rrrrrrrrr)rcr6rrres rY_transition_fetch_flightz$WorkerState._transition_fetch_flightus = 58w&&&&:   Z 5 5!1!!444444   $$$ A2v rXcP|xjdzc_|||SNrr)rrsrjs rY_transition_fetch_missingz%WorkerState._transition_fetch_missings1 A// /LLLrXcP|xjdzc_|||Sr)rrorjs rY_transition_fetch_releasedz&WorkerState._transition_fetch_releaseds1 A000MMMrXc|jJ|xj|jzc_|||\}}|t|j|||fS)Nrr)rrorr;r\rps rY_transition_memory_releasedz'WorkerState._transition_memory_releasedsyy$$$ ry !>> K?  l 0RVUUUVVV\!!rXcjrn|jdksJ|jrJtfd|jDsJtd|jDsJ|jvsJ|jvsJd|_j|j | S)NrFc3PK|] }|jjvp |jjvV!dSrar\rzrrdeprcs rYr z>WorkerState._transition_waiting_constrained..N49$>4;(>rXc3,K|]}|jdkVdS)r@Nr)rrs rYr z>WorkerState._transition_waiting_constrained..s)HHsyH,HHHHHHrXr9) rrrrrrBr9rFr,rrZrjs` rY_transition_waiting_constrainedz+WorkerState._transition_waiting_constraineds  = .8y((((* * * *?   HHHHHHH H H HTZ''''T-----  B R   %%'''rXc|jsJtit|j|gf|||S)zNote: this transition is triggered exclusively by a task raising the Reschedule() Exception; it is not involved in work stealing. rr)rrr>r\rorjs rY!_transition_executing_rescheduledz-WorkerState._transition_executing_rescheduledsS w& -BF DDDE F  - -bk - J J    rXc|jra|jdksJ||jvsJ||jvsJ|jrJ|jD]-}|j|jvs|j|jvsJ|jdksJ.|j r|digfSd|_|j J|j ||j ||S)NrFr@r9rB)rrrBr9rrr\rzrrrrFr,rrZ)rcr6rrs rY_transition_waiting_readyz%WorkerState._transition_waiting_readys  = -8y((((TZ''''T-----* * * * - -w$)++sw$+/E/E/E/EyH,,,,, # + &* *{&&& B r%%'''rXrr-rrrrc||_||_||_||_d|_t ||||j|j }i|gfS)Nr:)rrr,) rrrrrr4r8rrr\) rcr6rrrrrrsmsgs rY_transition_generic_errorz%WorkerState._transition_generic_errorso!   **%% #<##BF++ &  D6zrXc|jsJ|jdvr|jdksJ|di}n|jdksJ|jdksJ|di}d|_d|_d|_d|_|gfS)zIn case of failure of the previous state, discard the error and kick off the next state without informing the scheduler r;r?r<r=rFrCFNrrrr) rcr6rrrrrrrs rY_transition_resumed_errorz%WorkerState._transition_resumed_errorsw ;7 7 77g%%%%gDD;(****7i'''' ?D RxrXc|jsJ|jdvsJ|jdksJd|_d|_d|_d|_|digfS)a If the task raises the Reschedule() exception, but the scheduler already told the worker to fetch it somewhere else, silently transition to fetch. Note that this transition effectively duplicates the logic of _transition_resumed_error. rr<rCFNrrjs rY_transition_resumed_rescheduledz+WorkerState._transition_resumed_rescheduled saw{;;;;;w'!!!! G}b  rXc|jdkrS|jr |jdksJ|jr"d|_d|_d|_d|_|digfSd|_d|_d|_n(|jr!|jdvsJ|jdksJ|jrJigfS)z See also -------- _transition_cancelled_fetch _transition_cancelled_waiting _transition_flight_fetch r=rFrCFNrr<)rrrrrrjs rY_transition_resumed_fetchz%WorkerState._transition_resumed_fetchs ;( " "} ,w)++++w  &" I**$"  ] ;"?????7g%%%%w   2v rXc|digfS)Nr<rWrjs rY_transition_resumed_missingz'WorkerState._transition_resumed_missingCsG}b  rXc8|jrJd|_d|_igfS)Nr8)rrrrjs rY_transition_resumed_releasedz(WorkerState._transition_resumed_releasedHs) 72v rXc|jdkr|jr|digfSd|_d|_igfS|jdvsJ|jrJd|_d|_igfS)zs See also -------- _transition_cancelled_waiting _transition_resumed_fetch r=rCNrrEr<)rrrrrjs rY_transition_cancelled_fetchz'WorkerState._transition_cancelled_fetchRs ;( " "w J'++$" 2v ;"?????w    BHBGr6MrXc|jrJ|jdkrd|_d|_igfS|jdkr*d|_d|_t|jd|}i|gfS|jdksJd|_d|_igfS)z See also -------- _transition_cancelled_fetch _transition_cancelled_or_resumed_long_running _transition_resumed_fetch r;Nr?r\rCrr=rErF)rrrrAr\r)rcr6rrs rY_transition_cancelled_waitingz)WorkerState._transition_cancelled_waitingms7 ;+ % %"BHBKr6M [N * * &BHBK!FT{Dv: ;(**** BHBGr6MrXrSrcb|jsigfSd|_d|_|||S)NFr)rrro)rcr6rrSs rY_transition_cancelled_releasedz*WorkerState._transition_cancelled_releaseds> w r6M 000MMMrXc|jr|jdvsJ|jrJ|jrJt t d|j|_d|_igfS)zWe can't stop executing a task just because the scheduler asked us to, so we're entering cancelled state and waiting until it completes. rr8)rrrrrrrrjs rY_transition_executing_releasedz*WorkerState._transition_executing_releasedsg = 8<<<<<w   w   7#>?JJ 2v rXc|jrd|jdksJ|jrJ|j|jvsJ||jvsJ||jvsJ|jD] }|j|jvs|j|jvsJ!d|_t|j|}i|gfS)Nr9r;r) rrrr\rzrBr9rrr)rcr6rrrs rY!_transition_constrained_executingz-WorkerState._transition_constrained_executings = F8},,,,* * * *6****TZ''''T----- F Fw$)++sw$+/E/E/E/EBF <<<E7{rXcjr^|jdksJ|jrJ|jjvsJ|jvsJ|jvsJtfd|jDsJd|_t|j|}i|gfS)NrBc3PK|] }|jjvp |jjvV!dSrarrs rYr z:WorkerState._transition_ready_executing..rrXr;r) rrrr\rzrBr9rrr)rcr6rrs` rY_transition_ready_executingz'WorkerState._transition_ready_executings = 8w&&&&* * * *6****TZ''''T-----?    BF <<<E7{rXcF|jsigfS|||Srh)rrfrjs rY_transition_flight_fetchz$WorkerState._transition_flight_fetchs0 w r6M--bk-JJJrXcF|jrJd|_d|_d|_igfS)Nr=r8rrjs rY_transition_flight_releasedz'WorkerState._transition_flight_releaseds07 2v rXrCcd|_|j||j|t |j||}ti|gf|S)za See also -------- _transition_cancelled_or_resumed_long_running r?r) rr;r0rrrAr\rrZ)rcr6rCrrs rY"_transition_executing_long_runningz.WorkerState._transition_executing_long_runnings" r""" b!!!)9{   ' $L  " " $ $   rXc|jdvsJd|_|j||j||S)aHandles transitions: - cancelled(executing) -> long-running - cancelled(long-running) -> long-running (user called secede() twice) - resumed(executing->fetch) -> long-running - resumed(long-running->fetch) -> long-running (user called secede() twice) Unlike in the executing->long_running transition, do not send LongRunningMsg. From the scheduler's perspective, this task no longer exists (cancelled) or is in memory on another worker (resumed). So it shouldn't hear about it. Instead, we're going to send the LongRunningMsg when and if the task transitions back to waiting. See also -------- _transition_executing_long_running _transition_cancelled_waiting rr?)rr;r0rrrZ)rcr6rCrs rY-_transition_cancelled_or_resumed_long_runningz9WorkerState._transition_cancelled_or_resumed_long_runnings^*{;;;;;$  r""" b!!!%%'''rXrrc6|||d||S)zThis transition is *normally* triggered by ExecuteSuccessEvent. However, beware that it can also be triggered by scatter(). r*rr)_transition_to_memoryrcr6rrrs rY_transition_executing_memoryz(WorkerState._transition_executing_memory s, )) v;*   rXc@|||dt|S)zThis transition is triggered by scatter(). This transition does not send any message back to the scheduler, because the scheduler doesn't know this key exists yet. Frrrrs rY_transition_released_memoryz'WorkerState._transition_released_memory s,)) u_+*   rXc@|||dt|S)zThis transition is *normally* triggered by GatherDepSuccessEvent. However, beware that it can also be triggered by scatter(). rFrrrs rY_transition_flight_memoryz%WorkerState._transition_flight_memory' s, )) z/{*   rXc|jdvr|jdksJd}n|jdksJ|jdksJd}d|_d|_||||||S) a(Normally, we send to the scheduler a 'task-finished' message for a completed execution and 'add-data' for a completed replication from another worker. The scheduler's reaction to the two messages is fundamentally different; namely, add-data is only admissible for tasks that are already in memory on another worker, and won't trigger transitions. In the case of resumed tasks, the scheduler's expectation is set by ts.next - which means, the opposite of what the worker actually just completed. rr<rFr=rFr*Nr)rrr)rcr6rrrmsg_types rY_transition_resumed_memoryz&WorkerState._transition_resumed_memory1 s ;7 7 77g%%%%!HH;(****7i''''&H )) xK*   rXr+Literal[False, 'add-keys', 'task-finished']c2i}g}|jr|j|jvsJ|jdksJ|j|jvr||j|j<nt } ||j|j<nO#t $rB} t| } |t| |fzigfcYd} ~ Sd} ~ wwxYwt } | |z dkr|j d|| dd|_|j t||_ |xj |j z c_ t||_|jD]3} | j|| js| jdkrd|| <4|j |jd|t f|d kr+| t'|jg| nE|d kr9|t(ksJ| |||| n|d usJ||fS)a4Insert a task's output in self.data and set the state to memory. This method is the one and only place where keys are inserted in self.data. There are three ways to get here: 1. task execution just terminated successfully. Initial state is one of - executing - long-running - resumed(prev=executing next=fetch) - resumed(prev=long-running next=fetch) 2. transfer from another worker terminated successfully. Initial state is - flight - resumed(prev=flight next=waiting) 3. scatter. In this case *normally* the task is in released state, but nothing stops a client to scatter a key while is in any other state; these race conditions are not well tested and are expected to misbehave. r@Ng{Gzt?z disk-writerNrPrQrFrBz put-in-memoryrFrHrr*rF)rr\rzrrr+rr(r2valuesrrrsizeofrhrrr0rrErrc) rcr6rrrrr>rrPerrQrs rYrz!WorkerState._transition_to_memoryL s2!#%' = (6****8x'''' 6T[ "'DK  FFE A$) "&!! A A A$A&&E#**,,//6);Nr)rr%s rYr z=WorkerState._transition_released_forgotten.. s)IIa17k1IIIIIIrXrCr>) ranyrrr0rr2rrer\)rcr6rr>rs rY_transition_released_forgottenz*WorkerState._transition_released_forgotten s!# = JII2=IIIII I I I? 3 3C N " "2 & & &yJ&&s~&'2$ " rvt$$$""rX)r8r:)r8r<)r8r?)r8r@)r8rA)r8rC)r8rD)r8rF)rEr:)rEr<)rEr?)rEr@)rErC)rErD)r9r;)r9rC)r:rC)r;r:r)r;r@)r;rC)r;rD)r<r=)r<rA)r<rC)r=r:)r=r<)r=r@)r=rA)r=rC)r?r:)r?r@)r?rD)r?rC)r@rC)rAr:)rAr<)rArC)rArF)rBr;)rBrC)rCr:)rCr<)rCr>)rCr@)rCrA)rCrF)rFr9)rFrB)rFrCzSClassVar[Mapping[tuple[TaskStateState, TaskStateState], Callable[..., RecsInstrs]]]_TRANSITIONS_TABLE method_namerc|jD]Z\}}t||rE t|||i|-#t$r t d|dYVwxYw[dS)Nz!Plugin '%s' failed with exceptionT)exc_info)rrhasattrrrr%rb)rcrrSrnameplugins rY_notify_pluginszWorkerState._notify_plugins s L..00  LD&v{++ 0GFK00$A&AAAA KK;TD    sA'A10A1r]TaskStateState | tuplec t|tr)|rJ|dd}tt|d}|j|krigfS|j}|j||f}|xjdz c_|jr:|j|jkr*t|j ||| ||/|||g|Rd|i\}}| d|j ||n1d||fvr | |d|\}}||dx} rbt| tr| ^} } n| d} } | d kr=t||f|j || g| Rd|i\}}||dx} bt||f|j ||g|Rd|i\}}nn#t t"f$r0} t!|j ||| || d} ~ wwxYwt!|j ||| ||j|j |||jd |D|t+f||fS) zTransition a key from its current state to the finish state See Also -------- Worker.transitions: wrapper around this method rNrr transitionrCrrWr>c^i|]*\}}|jt|tr|dn|+Sr)r\rr2)rr6news rYrz+WorkerState._transition..T sGCFje&<&<ECFF#rX)rr2rrGrrrrrrr\r^r _transitionrerr[rrrrr+) rcr6r]rrSrPfuncrrrv_statev_argsrs rYrzWorkerState._transition s fe $ $ 5OOO!"":D.&)44F 8v  r6M&**E6?;; 1$  ' V'4+FFF.rvufdjjQSnnUU U  !%dB!O!O!O!O;!O!O D,  rvuf E E E E v . . V%)%5%5 &6&&"l  88B---a !!U++0+,(&&*+R+--!)@|,((WWvWWW;WW**&D, 88B---a &=<($D$RP$PPPKPP&&"ll&'>? V V V'vtzz"~~NNTUU V$BFE64::b>>JJ J #'::<<    &\!!s;B:F66G7+G22G7chtfd|jDS)Nc3@K|]\}}j||kVdSra)r)rresourceneededrcs rYr z?WorkerState._resource_restrictions_satisfied.._ sH   &  $X .& 8      rX)rrrrcr6s` rYr]z,WorkerState._resource_restrictions_satisfied^ sJ    $&$<$B$B$D$D      rXcn|jD]\}}|j|xx|zcc<dSrarrrrcr6rrs rYrYzWorkerState._acquire_resourcesd Q " 8 > > @ @ 9 9 Hf  $X . . .& 8 . . . . 9 9rXcn|jD]\}}|j|xx|z cc<dSrarrs rY_release_resourceszWorkerState._release_resourcesh rrXr>rcJ gt id fd }||\}}|z ||jjr D]}|S)zProcess transitions until none are left This includes feedback from previous transitions and continues until we reach a steady state rrrerc,|r|\}} |||j||\}}||||dSdSrh)popitemr setdefaultrrupdateextend) rr6r]a_recsa_instructionsinitial_statesrrcrrs rY process_recsz.WorkerState._transitions..process_recsv s 4!\\^^ F " ))"bh777)-)9)9K*:**& F#####N333 4 4 4 4 4rXr)rrrer)rrrCr transitionsr validate_task) rcr>rrrrr6rrrs ` ` @@@rYrzWorkerState._transitionsl s  :< 4 4 4 4 4 4 4 4 4 4  _))++,,,"&!;!; !;!T!T&  V %%n555 = ' ' '""2&&&&rXevc t|ra) TypeErrorrcrs rYrzWorkerState._handle_event smmrXrc ti}|jD]\}} |j|}nC#t$r6t |x|j|<}|j|YnwxYwd|tf||<|j |d|j |j tf|gfS)Nr@zreceive-from-scatter) rzrrr'rrr(rrrrrr+)rcrr>r\rr6s rY_handle_update_datazWorkerState._handle_update_data s "'--//  JC /Z_ / / /'0~~5 3"!**2..... /$,UO"DOB  HOO,bhO    ""s 0=A0/A0rc|jd|j|jt fi}|jD]#}|j|}|rd||<$|gfS)aHandler to be called by the scheduler. The given keys are no longer referred to and required by the scheduler. The worker is now allowed to release the key, if applicable. This does not guarantee that the memory is released since the worker may still decide to hold on to the data and task since it is required by an upstream dependency. z free-keysrC)rrrHrr+rr)rcrr>r\r6s rY_handle_free_keyszWorkerState._handle_free_keys sp bgr~tvvFGGG "7 1 1C$$B 1&0#""rXrc ^i}g}|jD]}|j|}| |jdkr*|jD]3}|jdvr(t d|jd|jd|jd4|j|jd|j tfd ||<||fS) aStream handler notifying the worker that it might be holding unreferenced, superfluous data. This should not actually happen during ordinary operations and is only intended to correct any erroneous state. An example where this is necessary is if a worker fetches data for a downstream task but that task is released before the data arrives. In this case, the scheduler will notify the worker that it may be holding this unnecessary data, if the worker hasn't released the data itself, already. This handler only releases tasks that are indeed in state memory. For stronger guarantees, see handler free_keys Nr@rzCannot remove replica of z while z in state .zremove-replicarC) rHrrrr RuntimeErrorr\rrrr+)rcrr>rr\r6rs rY_handle_remove_replicasz#WorkerState._handle_remove_replicas s !#%' 7 - -C$$BzRX11}  9 ===&hBFhhSWhhZ]Zchhh> HOORV%5r~tvvN O O O",OB   ,,rXrc|jr^|j|jksJt |jsJi}|jD]9\}}||d|j}|j dkr ||_d||<:| |j|gfS)Nrr\rrr@r<) rrrHrrrrr)rrr.)rcrr>r\rr6s rY_handle_acquire_replicasz$WorkerState._handle_acquire_replicas s = ,:??$$ (8(88888rz((**++ + + + "9??,, . .KC)) N*Bx8##" &-# RZ(((""rXrc r |j|j}td||jdnM#t $r@t |jx|j|j<}|j|YnwxYw|j |jd|j |jtf|j |_ i}g}|j thdzvrnQ|j dkr7| |||j |jn|j dvrd||<|j|_|j|jfz}|xjd zc_|jrd|j|j<d|_d|_d |_d |_||_|j|_|j|_|j|_|j d vr |jd vs |j|_|jr|j |j! ksJ|j"D]5}|sJtG|tGtI|ksJ6|j!%D]h\}}|&|||j } | j dkr|| _!|j'(| | j)(|i|*|jn0tWd|d|jd|,|||fS)Nz)Asked to compute an already known task %s)taskrz compute-task>rFr;r?r@r>r:r<r=rArErCr8rFrrr8rErrz&Unexpected task state encountered for z; stimulus_id=z; story=)-rr\r%r&rr'rrr(rrrr+rrIrcrrrrrrrrrrrrrrrrrHrrrrrr)rrrr.rr^) rcrr6r>rr dep_workersdep_keyrr|s rY_handle_compute_taskz WorkerState._handle_compute_task s +BF#B LL;BN;;     + + +&/&7&7 7DJrv    & &r * * * * * + 2>466RSSSI "%' 8u       X ! !   ++rybn,     X   #,OB +BK{do%77H OOq OOx +&* BF#BLBL "B  "B "BK+BK^BNBJ444K#@@@+-+C(} Ez((BINN,<,<<<<<#%:#4#4#6#6EEK&&&&{++s3{3C3C/D/DDDDDD#%9??#4#4 * *11% "2 <8++$*FM##F+++!%%b))))   , , , ,HHH!~HH7;zz"~~HH   ,,s58ABBruIterator[TaskState]c#K|xj|jzc_|j|j}|D];}|j|}d|_d|_|j ||Vr6s rY_handle_gather_dep_successz&WorkerState._handle_gather_dep_successu s !#..r22 . .Bv  '/")&L## OPPP=A8w....T%5bi%@@@@@ ""29---9 --M"),44RV<<<&-##""rXrcj|j|ji}g}||D]0}d||<|j|jz s||j1t|j|jg}|r)|t||j||fS)z,gather_dep terminated: remote worker is busyr<)rrr) rrrrrrr\rrrK)rcrr>refresh_who_hasr6rs rY_handle_gather_dep_busyz#WorkerState._handle_gather_dep_busy s bi((( "..r22 / /B")OB : 11 /&&rv... ! r~ N N N&      '(bn     ,,rXrci}||D];}|j|jd|jt fd||<<|j|jdD]<}|j r|j dksJ|j|j vsJ|j |jhkrd||<=|j |jdD].}|j |}|j |j/|gfS)agather_dep terminated: network failure while trying to communicate with remote worker Though the network failure could be transient, we assume it is not, and preemptively act as though the other worker has died (including removing all keys from it, even ones we did not fetch). This optimization leads to faster completion of the fetch, since we immediately either retry a different worker, or ask the scheduler to inform us of a new worker if no other worker is available. rr<rWrA)rrrr\rr+rrerrrrrrr,rcrr>r6r\s rY"_handle_gather_dep_network_failurez.WorkerState._handle_gather_dep_network_failure s!#..r22 * *B HOORV]BNDFFK L L L")OB  "&&ry"55 0 0B} /x7****yBJ....zbi[((&/#=$$RY33 ) )CCB J  bi ( ( ( (""rXrcNfd|D}|gfS)zvgather_dep terminated: generic error raised (not a network failure); e.g. data failed to deserialize. c Ti|]$}|djjjj|jf%S)r:)rrrrr)rr6rs rYrz:WorkerState._handle_gather_dep_failure.. sN ! ! !    !!  ! ! ! rX)r)rcrr>s ` rY_handle_gather_dep_failurez&WorkerState._handle_gather_dep_failure sH ! ! ! ! 22266 ! ! ! ""rXrcfi}|j|jdD]<}|jr|jdksJ|j|jvsJ|j|jhkrd||<=|j|jdD].}|j|}|j|j/|gfS)NrWr<rA) rrerrrrrrr,rs rY_handle_remove_workerz!WorkerState._handle_remove_worker s ""&&ry"55 0 0B} /x7****yBJ....zbi[((&/#=$$RY33 ) )CCB J  bi ( ( ( (""rXrcf|j|j}|sigfS|d|jfigfS)Nr?)rrr\rCrcrr6s rY_handle_secedezWorkerState._handle_secede s@ Z^^BF # # r6M^R%89:B>>rXrc|j|j}||jnd}t |j||j}|t dhzvr |sJ|di|gfSi|gfS)N)r\rrrFrC)rrr\rrNrrI)rcrr6rrs rY_handle_steal_requestz!WorkerState._handle_steal_request s{ Z^^BF # #NBF%R^TTT EYK' ' 'III #dV+ +v: rXrmcd|_igfS)zPrevent any further tasks to be executed or gathered. Tasks that are currently executing or in flight will continue to progress. F)rrs rY _handle_pausezWorkerState._handle_pause s  2v rXrqc8d|_|S)zEmerge from paused statusT)rrZrs rY_handle_unpausezWorkerState._handle_unpause s %%'''rXrscH|j|jigfSra)rr0rrs rY_handle_retry_busy_workerz%WorkerState._handle_retry_busy_worker s$ !!"),,,2v rXrc|j|j}|r|jtdhzvrigfS|j|jd|jtf|j rJ|digfS)zCancel a task on a best-effort basis. This is only possible while a task is in state `waiting` or `ready`; nothing will happen otherwise. rFzcancel-computerC) rrr\rrIrrrr+rrs rY_handle_cancel_computez"WorkerState._handle_cancel_compute s Z^^BF # # RXUi[%888r6M !12>466JKKK=   J##rXr$tuple[TaskState, Recs, Instructions]c|j|j}|s"J||j|jr||jv||jvksJd|_|xjdz c_| ||j ||j || \}}||vsJ|||fS)aCommon code for the handlers of all subclasses of ExecuteDoneEvent. The task state can be executing, cancelled, or resumed, but in case of scatter() it can also be in memory or error state. See also -------- _gather_dep_done_common Tr) rrr\r^rr;rrrrr0rZrcrr6rrs rY_execute_done_commonz WorkerState._execute_done_common) sZ^^BF # #%%4::bf%%%%% = G$.(bD4E.EFFFF q  ### r""" !!"%%%,,.. e~~~~4rXrc||\}}}|jd|j|jd|j|_|j|_d|j|jf||<||fS)zTask completed successfullycomputerr@) r.rrrPrQrrhrrr-s rY_handle_execute_successz#WorkerState._handle_execute_successE ss33B77D%  BHbgVVWWWI 'bh 2RU{rXrc||\}}}|j/|j(|jd|j|jdd|j|j|j|j|j f||<||fS)zTask execution failedNr0rr:) r.rPrQrrrrrrrr-s rY_handle_execute_failurez#WorkerState._handle_execute_failureR s33B77D% 8 BG$7 M $rxII     L L     I  RU{rXrcF||\}}}d||<||fS)zTask raised Reschedule() exception while it was running. Note: this has nothing to do with work stealing, which instead causes a FreeKeysEvent. rD)r.r-s rY_handle_reschedulezWorkerState._handle_rescheduled s033B77D% RU{rXrc|jsigfS|jr.|jD]&}|jrJ||'t d|jD|j}i|gfS)Ncg|] }|j SrWrr:s rY z4WorkerState._handle_find_missing..y s;;;R"&;;;rXr)rrrr^rKr)rcrr6rs rY_handle_find_missingz WorkerState._handle_find_missingo s& r6M = 6- 6 6:55tzz"~~5555&;;4#:;;;   D6zrXrc||ji}g}|jD]N}|j|}|s|jr|jdkrd||<7|js|jdkrd||<O||fS)NrAr<)r.rrrr)rcrr>rr\r6s rY_handle_refresh_who_hasz#WorkerState._handle_refresh_who_has~ s RZ((( "%' : 0 0C$$B z 0bh)33&-##Z 0BH$7$7&/# ,,rXkeys_or_tasks_or_stimulistr | Key | TaskStater_cDd|D}t||jS)ziReturn all records from the transitions log involving one or more tasks or stimulus_id's cJh|] }t|tr|jn|!SrWrrr\rrs rYr;z$WorkerState.story.. s:   9:Z9-- 4AEE1   rX)r$r)rcr<keys_or_stimulis rYr^zWorkerState.story s4  >V   OTX666rX keys_or_tasksKey | TaskStatelist[StateMachineEvent]cBd|Dfd|jDS)z;Return all state machine events involving one or more taskscJh|] }t|tr|jn|!SrWr@rAs rYr;z-WorkerState.stimulus_story.. s-PPPQAy118qPPPrXc:g|]}t|ddv|S)r\Nr)rrrHs rYr8z.WorkerState.stimulus_story.. s/SSSr'"eT2J2Jd2R2R2R2R2RrX)r)rcrCrHs @rYstimulus_storyzWorkerState.stimulus_story s6QP-PPPSSSST.SSSSrXrWrrrrcbid|jd|jd|jdd|jDdd|jDdt |jd d |j Dd d |j Dd t |j dd|j Ddd|jDd|jdd|jDd|jd|jd|jd|j|jt |jt |jd}fd| D}t3|S)zDictionary representation for debugging purposes. Not type stable and not intended for roundtrips. See also -------- Client.dump_cluster_state distributed.utils.recursive_to_dict rrrrBcg|] }|j SrWrr:s rYr8z(WorkerState._to_dict.. s;;;bf;;;rXr9cg|] }|j SrWrr:s rYr8z(WorkerState._to_dict.. sGGGrBFGGGrXrzrcRi|]$\}}|d|D%S)cg|] }|j SrWrr:s rYr8z3WorkerState._to_dict... s222rBF222rX)sorted)rretsss rYrz(WorkerState._to_dict.. sEAs22SZZ\\222rXr;ch|] }|j SrWrr:s rYr;z'WorkerState._to_dict.. s:::R"&:::rXrrch|] }|j SrWrr:s rYr;z'WorkerState._to_dict.. s@@@RV@@@rXrch|] }|j SrWrr:s rYr;z'WorkerState._to_dict.. sFFF2FFFrXrrcg|] }|j SrWrr:s rYr8z(WorkerState._to_dict.. s"L"L"Lb26"L"L"LrXrrrr)r task_countstask_cumulative_elapsedc$i|] \}}|v || SrWrW)rrrrs rYrz(WorkerState._to_dict.. s)BBBA'1A1A11A1A1ArXr)rrrrBrOr9rrrzrrr;rrrrrrrrrrr current_countcumulative_elapsedr0)rcrrbs ` rYrczWorkerState._to_dict s* t|    t|  ;;tz'8'8':':;;;  GGT-=-D-D-F-FGGG  DMM$),,   ".4466  ::4>:::  T]++  @@d.?@@@  FF1EFFF  !7 !"L"LD4K"L"L"L! " D-# $ 48% & D-' ( !$"9) *Z 1 ? ? A ABB'+D,=,P,P,R,R'S'S/   2CBBBBBB w7777rXc|j|jvs|j|jvsJt|jt sJ|jrJdSra)r\rzrrrrrrs rY_validate_task_memoryz!WorkerState._validate_task_memory sRv""bf &;&;&;&;")S)))))&&&&&&rXc@|jdks |jdkr||jvsJ||jvsJn.|jdks |jdksJ||jvsJ||jvsJ|jJ|j|jvsJ|jrJ|jD]}||j vsJ||j vsJdS)ajValidate tasks: - ts.state == executing - ts.state == long-running - ts.state == cancelled, ts.previous == executing - ts.state == cancelled, ts.previous == long-running - ts.state == resumed, ts.previous == executing, ts.next == fetch - ts.state == resumed, ts.previous == long-running, ts.next == fetch r;r?N) rrr;rrr\rzrrrBr9rcr6rs rY_validate_task_executingz$WorkerState._validate_task_executing s 8{ " "bk[&@&@''''T......8~--1N1N1N1NT^++++*****{&&&vTY&&&&&&&&= / /Cdj((((d...... / /rXc>|jdkr |jrJ|jvsJ|jvsJn,|jsJ|jdksJ|jvsJ|jvsJ|jjvsJ|jrJ|jrJtfd|j DsJdS)zWValidate tasks: - ts.state == ready - ts.state == constrained rBr9c3PK|] }|jjvp |jjvV!dSrarrs rYr z3WorkerState._validate_task_ready.. sK  ?BCGty :CGt{$:      rXN) rrrBr9r\rzrrrrrs` rY_validate_task_readyz WorkerState._validate_task_ready s 8w  / / / /####T------+ + + +8},,,,TZ'''')))))vTY&&&&7&&&&    FHo          rXc|jjvsJ|jrJ|jvsJ|jsJ|jfd|jDksJ|jD]}|jvsJ|jvsJdS)NcLh|] }|jjv|jjv|!SrWrrs rYr;z5WorkerState._validate_task_waiting.. s@' ' ' wdi''CG4;,F,F ,F,F,FrX) r\rzrrFrrrrBr9r]s` rY_validate_task_waitingz"WorkerState._validate_task_waiting svTY&&&&7T\!!!!"""""' ' ' ' ' ' '     = / /Cdj((((d...... / /rXc|j|jvsJ|j|jvsJ||jvsJ|jD]}||jvsJ||jvsJ|jsJ|j|jvsJ|j|j|jvsJdS)zValidate tasks: - ts.state == flight - ts.state == cancelled, ts.previous == flight - ts.state == resumed, ts.previous == flight, ts.next == waiting N) r\rzrrrrBr9rrr]s rY_validate_task_flightz!WorkerState._validate_task_flight svTY&&&&vT[((((T)))))= / /Cdj((((d......~~!77777v/???????rXc,|j|jvsJ|j|jvsJ|j|jvsJ|jrJ|jsJ|jD])}|j|j|vsJ||j|vsJ*|jD]}||j vsJ||j vsJdSra) r\rzrrrrrrrrBr9)rcr6rers rY_validate_task_fetchz WorkerState._validate_task_fetch/ svTY&&&&vT[((((|2:----7z - -A6T]1-----)!,,,,,,= / /Cdj((((d...... / /rXc*j|jvsJj|jvsJjrJjrJt fd|jDrJ|jvsJj D]}||j vsJ||j vsJdS)Nc3*K|] }j|vVdSrar)rrr6s rYr z5WorkerState._validate_task_missing..A s*QQhrv)QQQQQQrX) r\rzrrrrrrrrrBr9r]s ` rY_validate_task_missingz"WorkerState._validate_task_missing< svTY&&&&vT[((((:7QQQQ$-:N:N:P:PQQQQQQQQT,,,,,= / /Cdj((((d...... / /rXc|jJ|jdvr||dS|jdksJ||dS)Nrr=)rrr^rfrs rY_validate_task_cancelledz$WorkerState._validate_task_cancelledG sbw ;7 7 7  ) )" - - - - -;(****  & &r * * * * *rXc|jdvr#|jdksJ||n/|jdksJ|jdksJ|||jD]}||jvsJ||jvsJdS)Nrr<r=rF)rrr^rfrrBr9r]s rY_validate_task_resumedz"WorkerState._validate_task_resumedO s ;7 7 77g%%%%  ) )" - - - -;(****7i''''  & &r * * *= / /Cdj((((d...... / /rXc8|j|jvsJ|j|jvsJ|jrJ|jrJ|jD]}||vsJ ||jvsJ||jvsJ||j vsJ|j rJ|j rJ|j rJ|j rJdSra)r\rzrrrrrr;rrrrrr)rcr6rPs rY_validate_task_releasedz#WorkerState._validate_task_released[ svTY&&&&vT[((((7;#**,, ! !CS=====''''-----00000&&&&7<<rXc |j|jvr|j|j|usJ|jdkr||dS|jdkr||dS|jdkr||dS|jdkr||dS|jdkr||dS|jdvr||dS|jdvr| |dS|jdkr| |dS|jd kr| |dS|jd kr| |dSdS#t$rO}t|t!|j|j|| |d}~wwxYw) Nr@rFrAr8rE)rBr9rr=r<rCr)r\rrr[rdrkrmrorar^rfrhrqrr%rrr^)rcr6rs rYrzWorkerState.validate_taskr s" v##z"&)R////x8##**2.....Y&&++B/////Y&&++B/////[((--b11111Y&&++B/////555))"-----:::--b11111X%%**2.....W$$))"-----Z'',,R00000('      Q   "F"($**R..  sOAE8 E8* E8 E8. E8E80E8 E82 E8 E88 GA G  Gc, jD]}|jD]%}|jksJ|jj|vsJ&|jD]>}j|j|usJ||jvsJ|?|j D]C}j|j|usJ|j tvsJ|Dj D]S\}}|jksJ|D]>}|jvsJ||j|jvsJ?Tt}j D]d\}}|D]\}|||j dksJ|||jvsJ|d|j]et!|jksJjD]*}|j dksJ|+jD]*}|j dksJ|+jD]*}|j dksJ|+jD]>}|j dks1|j dvr |jdksJ|?jD]>}|j dks1|j dvr |jdksJ|?jD]>}|j d ks1|j dvr |jd ksJ|?jD]*}|j d ksJ|+t5gjjjjjjjjRD]5}j|j|usJj|jd |6t7fd t5jjD} j| ksJd jd| jD](} | jvsJ| )jD](} | jvsJ| )jD]}|tAdjD} j!"| ks#Jj!"| fj#rj$j#ksJ%dS)Nr<z ; ts.who_has=rArBr9r;r r?r=rFz is not c3@K|]}j|jpdVdS)rN)rr)rr\rcs rYr z-WorkerState.validate_state.. sB  ,/DJsO " 'a      rXz self.nbytes=z ; expected c32K|]}|j|jfVdSra)rrr:s rYr z-WorkerState.validate_state.. s<% % &(RY !% % % % % % rX)&rrrrr\rrrr^rrrJrrrrrrrrBr9r;rrrrFrsumrzrrrrrrXrr_validate_resources) rcr6rrts_waitrHr fetch_tssrP expect_nbytesr\expect_state_counts ` rYvalidate_statezWorkerState.validate_state s*##%% N NB* 7 7----vv!666666 < < z#'*c1111S^+++TZZ^^++++. N Nz'+.'9999}(8888$**W:M:M8888 N!M//11 7 7LFDT\)))) 7 7DJ 1 A!666666 7 EE +1133 D DKFC D D b!!!x7***DJJrNN***+++-C-Cbj-C-C++++ D9~~!11111) 9 9B8y((($**R..((((* 7 7B8w&&& 2&&&&" = =B8},,,djjnn,,,,.  B8{**444 9S9S9Szz"~~:T9S9S#  B8~--4449V9V9Vzz"~~:W9V9V&  B8x''4449P9P9Pzz"~~:Q9P9P, 9 9B8y((($**R..((((    $ $ & &   #   J        N       L    Q QB:bf%+++ 260B-P-PB-P-P++++    38DK3P3P     {m+++-X -X-X-X-X+++9 6 6C$*$$$djjoo$$$$; 6 6C$*$$$djjoo$$$$*##%% # #B   r " " " "$% % ,0J,=,=,?,?% % %    ..004FFFF   + + - - I FFF  & I*T-HHHHH   """""rXc6|j|jksJ|j}|jD]*\}}|dksJ|j||xx|zcc<+|jD]H}|jD],\}}|dksJ||jf||xx|zcc<-Itd|Ds J|dS)zKAssert that available_resources + resources held by tasks = total_resourcesg& .rc3<K|]}t|dkVdS)g& .>N)abs)rrs rYr z2WorkerState._validate_resources.. s,;;qCFFTM;;;;;;rXN) rrHrrrrrrr)rctotalrrr6s rYrwzWorkerState._validate_resources s:#((**d.F.K.K.M.MMMMM$))++,2244  DAqu999d6999 !HHHMHHHH(  B06688  1AvvvB$<=vvvaA  ;;ELLNN;;;;;BBUBBBBBrX)rrrrrzr rrrr r r rrrrrrrrOrrO)rrPrerr)rer)r\r rrrrkrer)rr*rerr6rrer)rer)rrkrer)rerD)rPrrerQ)r6rrArrer)rer)rer[)r6rrrrrkrer))r6rrrkrer)r6rrrkrrkrer)r6rrr-rrrrkrrkrrrrkrer)r6rrSrrrkrer)r6rrCrOrrkrer) r6rrrrrrrkrer) r6rrrrrrrrrkrer)rrkrSrrrrer) r6rr]rrSrrrkrer)r6rrer)r>rrrkrer)rrPrer)rrrer)rrrer)rrrer)rrrer)rrrer)rrurer )rrxrer)rrrer)rrrer)rrrer)rrrer)rrrer)rrrer)rrmrer)rrqrer)rrsrer)rrrer)rrrer+)rrrer)rrrer)rrrer)rrrer)rrrer)r<r=rer_)rCrDrerErr)rrSrTrUrrVr2rmathinfrdrr`rrr!r#r)r.r2r7rCr<r=rSrZrXrcrfrkrmrqrtrsrwrorir~rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr]rYrrrrregisterrrrrr rrrrrrr r"r$r&r(r*r.r1r3r5r9r;r^rIrcr[r^rardrfrhrkrmrorqrr|rwrWrXrYrr$s(LLL  &%%% %$$$ $###MMM MMM)((( 6555(''''&&&$### +***!   '&&&'&&&4333 OOO&%%% *)))!   KKK  +***NNN 1000)(((o&&I "37)-2604-171>1>1>1@))))0$:$:$:$:L(0    F    NNNN K K K K""""MMMM    KKKK$####*"MMMM NNNN " " " "(((($     ((((,2:!!!!$%%%%N!!!! 6!!!!F N N N N     $ K K K K        (((((6                 6\-\-\-\-|####23  >3  ;3  &'T3  !? 3  !"@ 3  "#A 3  %&D3  !"?3  73  73  $%R3  93  !=3  #$C3  %&G3  $%A!3 " ;#3 3 $  9%3 & &'I'3 ( !=)3 * "#A+3 , %&G-3 . 5/3 0 713 2 933 4 653 6 573 8 793 : 9;3 <  ;=3 > "#d&Z e j d?d-Ze j d@d0Ze j dAd1Ze j dBd5ZdS)C BaseWorkerzWrapper around the :class:`WorkerState` that implements instructions handling. This is an abstract class with several ``@abc.abstractmethod`` methods, to be subclassed by :class:`~distributed.worker.Worker` and by unit test mock-ups. rrzset[asyncio.Task]_async_instructionsc||_t|_tjd|_dS)Nzdistributed.worker.validate)rrrrrrr&rs rYrdzBaseWorker.__init__ s/ #&55 [__%BCC rXN)r task_namerkr)Callable[P, Awaitable[StateMachineEvent]]rSP.argsrrrP.kwargsrerc*ttdfd }tj||}|j||t|j|dS)z:Execute an asynchronous instruction inside an asyncio taskrerPcKd5id{VcdddS#1swxYwYdS)Nasync-instructionr)record)rSrrledgersrYwrapperz4BaseWorker._start_async_instruction..wrappers#677 3 3!T42622222222 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3s 599r)rrNr) r)rasyncio create_taskrradd_done_callbackr_finish_async_instruction) rcrrrrSrrrrs ` `` @rY_start_async_instructionz#BaseWorker._start_async_instructions&'' t 3 3 3 3 3 3 3 3  3"77999===  $$T***  D267 S S S     rXrasyncio.Task[StateMachineEvent]rr)c|j| |}n[#tj$r"t d|dYdSt$rt d|wxYw| d5| |dddn #1swxYwY| |||dS)z\The asynchronous instruction just completed; process the returned stimulus. zAsync instruction for z ended with CancelledErrorNz-Unhandled exception in async instruction for rr) rr,resultrCancelledErrorr%warningrrrr_finalize_metrics)rcrrrrs rYrz$BaseWorker._finish_async_instruction%sD  ''--- ;;==DD%    NNTDTTT U U U FF      STSS T T T  ]]2] 3 3 ' '   & & & ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' tVW55555s!1-B !(B "CC CrrPct|ts|Jt|trId}|jD]2}|jj|}|r|jdkrd}n 3|jrdnd}nt|ttfrd}d}nt|trd}d}nt|trMd|t|j f}|jj|j }|r |jdkrdnd}nt|trd|t|j f}d}nPt|trd|t|j f}d}n!t|ts J|dS|| D]#\}} } |g||| R| $dS) N)z gather-depr@Fr8rAfailedbusyexecute) coarse_time)rrrxrzrrrrrrrr!r\rrrsfinalize digest_metric) rcrrractivityr\r6rlabelrunits rYrzBaseWorker._finalize_metricsDs$ 011 #??? d1 2 2" &Hy F FZ%))#.."(h.."'KE-1YEkkI 46RS T T &H"KK 0 1 1 &H KK 1 2 2 !7Idh,?,?@H!%%dh//B#%O"(h*>*>%%KKK 1 2 2 !7Idh,?,?@H"KK o . . !7Idh,?,?@H%KKd$899 ? ?4 ? ? ? F"(//k/"J"J @ @ E5$   7757$77 ? ? ? ? @ @rXrc |jj|}|D]}t|tr(||@t|t r|jsJ|jrbd tttd|jd}t|dkr |dddz}nd}|d|jd |d |j|j|j|j|j  t|t&rP|jj|j}|d |jd |j|j|j|jpt|t0r,|d|jd |j|jt5|dS)azForward one or more external stimuli to :meth:`WorkerState.handle_stimulus` and process the returned instructions, invoking the relevant Worker callbacks (``@abc.abstractmethod`` methods below). Spawn asyncio tasks for all asynchronous instructions and start tracking them. See also -------- WorkerState.handle_stimulus rrPNMz...z gather_dep(z, {z}))rrzexecute(r)rrzretry_busy_worker_later()rrrr! batched_sendr&rrr&rtrurkrrrr gather_deprrrrr\rrrretry_busy_worker_laterr)rcrrrgkeys_strr6s rYrzBaseWorker.handle_stimulusws2tz159  ( &( &D$ 677' &!!$,,..1111D),,$ &~%%%:%#yyS%DN2K2KA2N)O)OPPH8}}r))#+CRC=5#8$H--@$+@@8@@@OKN!%!2 $ 0 .D'** &Z%dh/--,tx,,,LHJ $ 0 .D"677 &--=t{===0K oo%Q( &( &rXtimeoutrOcK|jsdS|jD]}|tj|j|d{V\}}|D]"}td|d|#dS)z$Cancel all asynchronous instructionsN)rz$Failed to cancel asyncio task after z seconds: )rcancelrwaitr%r:)rcrrr]pendings rYclosezBaseWorker.closes'  F,  D KKMMMM#<(@'RRRRRRRRR 7  D LLPwPP$PP      rXrrcdS)aSend a fire-and-forget message to the scheduler through bulk comms. Parameters ---------- msg: dict msgpack-serializable message to send to the scheduler. Must have a 'op' key which is registered in Scheduler.stream_handlers. NrW)rcrs rYrzBaseWorker.batched_sendrjrXrrrIrrrc KdS)aGather dependencies for a task from a worker who has them Parameters ---------- worker : str Address of worker to gather dependencies from to_gather : list Keys of dependencies to gather from worker -- this is not necessarily equivalent to the full list of dependencies of ``dep`` as some dependencies may already be present on this worker. total_nbytes : int Total number of bytes for all the dependencies in to_gather combined NrW)rcrrrrs rYrzBaseWorker.gather_dep rXr\r c KdS)zExecute a taskNrW)rcr\rs rYrzBaseWorker.executerrXc KdS)z9Wait some time, then take a peer worker out of busy stateNrW)rcrs rYrz"BaseWorker.retry_busy_worker_laterrrXrr rcdS)z!Log an arbitrary numerical metricNrW)rcrrs rYrzBaseWorker.digest_metricrjrX)rr) rrkrrrSrrrrrrer)rrrr)rrrerra)rrPrr)rrrer)rrPrer)r)rrOrer)rrrer) rrkrrIrrrrkrerP)r\r rrkrerP)rrkrerP)rr rrOrer)rSrTrUrrVrdrrrrrabcabstractmethodrrrrrrWrXrYrrs ****DDDD#       .6666F# 1@1@1@1@1@f5&5&5&5&n               ,  HHHH 000000rXrczeZdZUded<ded<ded<ded<eeZdd ZdddZdddZddZ ddZ dS) rr _previous_tsz#Counter[tuple[str, TaskStateState]]_current_countr _new_tasksz.defaultdict[tuple[str, TaskStateState], float]_cumulative_elapsedrercd|_t|_t|_t t |_dSra)rrrrrrrOrris rYrdzTaskCounter.__init__s7 %ii%%#.u#5#5   rXT by_prefixrrc|r|jSt}|jD]\\}}}||xx|z cc<|S)aReturn current count of tasks. Parameters ---------- by_prefix: bool, optional True (default) Return counter of (task prefix, task state) -> count False Return counter of task state -> count )rrr)rcrrr]rns rYrXzTaskCounter.current_counts`  '& &'.yy!06688  MJQ JJJ!OJJJJ rXMapping[Any, float]cp|jr_|jJt}||jz }||_|jD]\}}|j|xx||zz cc<|r|jSt t }|jD]\\}}} ||xx| z cc<|S)aEver-increasing cumulative task runtimes, including tasks that have left a state or even that don't exist anymore, updated as of the moment when this method is called. Parameters ---------- by_prefix: bool, optional True (default) Return mapping of (task prefix, task state) -> seconds False Return mapping of task state -> seconds )rrr*rrrrO) rcrnowelapsedrn_tasksrr]rrs rYrYzTaskCounter.cumulative_elapseds   A$000++CD--G #D "17799 A A 7(+++w/@@++++  ,+ +2=e2D2D!5;;==  MJQ JJJ!OJJJJ rXr6rc:|j|dS)zvA new task has just been created and will be immediately fed into the recommendations for transitions N)rrrs rYr(zTaskCounter.new_task0s  BrX prev_statesdict[TaskState, TaskStateState]c|s |jsdSt}|jrJ|jJ||jz }|jD]\}}|j|xx||zz cc<||_|D]\}}|jdkr!|j|j|jfxxdz cc<||jvr|j|U|j|j|fdz }|dkr||j|j|f<|dksJ|j|j|f=|jD]#}|j|j|jfxxdz cc<$|j dS)z+Tasks have just transitioned to a new stateNr>rr) rr*rrrrrrr0r1) rcrrrrrr6 prev_state dec_counts rYrzTaskCounter.transitions6s 4?  Fkk   A$000D--G"17799 A A 7(+++w/@@++++)//11 C CNB x;&&#BIrx$7888A=888T_$$''++++ / :0EFJ q==AJD' :(=>>$>>>>+BIz,ABB/ : :B   28 3 4 4 4 9 4 4 4 4 rXNr)T)rrrer)rrrerr)rrrer) rSrTrUrVr2rrdrXrYr(rrWrXrYrrs7777HGGGo&&I6666 &:          rXrcNeZdZUded<ded<dddZdd Zdd ZddZddZdS)DeprecatedWorkerStateAttributerkrrtargetNc||_dSra)r)rcrs rYrdz'DeprecatedWorkerStateAttribute.__init__]s  rXownerrhrerc||_dSrar)rcrrs rY __set_name__z+DeprecatedWorkerStateAttribute.__set_name__`s  rXcjtjd|jd|jp|jdtdS)Nz The `Worker.z,` attribute has been moved to `Worker.state.`)warningswarnrr FutureWarningris rY_warn_deprecatedz/DeprecatedWorkerStateAttribute._warn_deprecatedcsQ  949 9 9![5DI 9 9 9      rXinstance Worker | None type[Worker]rct|dS|t|j|jp|jSra)rrrrr)rcrrs rY__get__z&DeprecatedWorkerStateAttribute.__get__js:  4 x~t{'?di@@@rXr7rcr|t|j|jp|j|dSra)rsetattrrrr)rcrrs rY__set__z&DeprecatedWorkerStateAttribute.__set__qs7  8ty%@@@@@rXra)rr)rrhrrkrerr)rrrrrer)rr7rrrer) rSrTrUrVrdrrrrrWrXrYrrYs III    AAAAAAAAAArXrrrW)rSrrer) __future__rrrrGloggingrrrsysrr collectionsrrrcollections.abcrrr r r r r rrr dataclassesrr functoolsrrrr itertoolsrtypingrrrrrrrtlzrr dask.typingr dask.utilsr!r"r#distributed._storiesr$distributed.collectionsr%distributed.commr&distributed.corer'r(distributed.metricsr)r*r+distributed.protocolr,distributed.protocol.serializer-r.distributed.sizeofr/rdistributed.utilsr0 getLoggerr%typing_extensionsr1r2r3r4distributed.diagnostics.pluginr5distributed.schedulerr6distributed.workerr7rGrVrHrIrJNO_VALUErrMrr[rrrr version_infoDC_SLOTSrrrrrrr!r)r4r;r>rArErKrNrPrmrqrsrurxrrrrrrrrrrrrrrrrrrrr2rrrrrrABCrrrrWrXrYrs"""""""    3333333333                      ((((((((EEEEEEEEEEEEPPPPPPPPPPPPPPPPPP 7777777777------++++++------88888888EEEEEEEEEE''''''>>>>>>>>444444//////  = > > *DCCCCCCCCC #A<;;;;;//////))))))$&### &}55555 ) ) )     # # # # # # # # # L66666#4666      y    FZZZZZiZZZ SSS S ".'99GT??r ,%,,8,,IEIEIEIEIEIEIE-,IEX ))))))) )DKKKKKKKK.         k     ;      [    , & ! ! ! ! ! )! !  ! H      1          *     #####+## # '  4 &  ! ! ! ! !- ! !  ! =U=U=U=U=U=U=U =U@ "  $  ,   *  11111.11 1& +  #5  ! ! ! ! ! .! !  ! H )   N N N N N (N N  N b (  / / / / / */ /  / d > > > > > *> >  > D AAAAA&AA A      *     (   ( ( ( ( (* ( (  ( ,   +   %         )       '  # y%(=">>?????{+ ++++dL01 1111"Y'CY'CY'CY'CY'CY'CY'CY'CxNh0h0h0h0h0h0h0h0Vk k k k k k k k \AAAAAAAAAArX