>ieddlmZ ddlmZmZddlmZmZddlm Z ddl m Z m Z ddl mZmZmZmZmZddlmZ d'd(dZd)dZd*dZGddZeddZ d+d,dZd-d!Zd.d$Zd/d&ZdS)0) annotations) defaultdict namedtuple)MappingMutableMapping)log)Anycast)get_dependenciesget_depsgetcycleistask reverse_dict)KeyNdskMutableMapping[Key, Any] dependencies$MutableMapping[Key, set[Key]] | Nonereturndict[Key, int]c @ABCDEFsiStfdDt@t@\BFt@FAt At kr?t d}t ddd|Dzd@D}t |dkrHttt}d+d }|g|R|<||<t}||=|Sd@AfdDD}|j }d,@ABfd } d,@ABFfd } Ft|dEEFfdADCiDd} t||g} | j} g}|j}|j}|j}t't}g}|j}|j}t)|}|j}|j}i}|j}g}|j}|j}t(j}d} t)} d}!| r| }"|"Dvr"B|"r|"|vr| |"||"D} dt | cxkrdkr)nn&| t3| | dn| | || |s| rB| drd}#n| D|"<| dz } @|"} d}!| r| D]}$B|$xxdzcc<d}#n|r|} | j} |rd}#n|r|D]_}%|%Dvr @|%}&| D|%<| dz } |&rB|&D]}$B|$xxdzcc<t |&dkr|&\}%B|%s @|%}&R| |&z} ~&`||| D} | sd}!d}#nnm|#rh|re| s|rdnd}!|t)|D}'t3|'CfdD]}%||%}(t ||@|(D|dkr ||%D @|%}&| D|%<| dz } |&r|&D]}$B|$xxdzcc<|!rV|&|z})|)rNt |&t |)kr)t |)dkr|)\}%B|%s @|%}&{nQ|&|)z }&nt)})t |&dkr#|&\}%B|%s|)s @|%}&||%n| |&z} ~& ~'|| D} || sd}!| |z})|)rNt | t |)kr't |)dkr|)\}$B|$s|"||$<~$Vd}!| |)z } t | dkrT| \}$|!r| s|$g} | j} ||$C|$}*B|$s|"||$<n||*| ~$~*nDt | dkr| \}$}+C|$}*C|+},|,|*ks|*|,kr | |+| |$kr|+|$}+}$|,|*},}*| r C| d}-|,|-krH|| ||+g|$g} | j} || B|+s|#r ||+n|"||+<n|*|-krY|| |$g} | j} ||$B|+s|#r ||+nr|"||+<nl||,|+gnOC|"}.|*|$f|,|+ffD]=\}/}0B|0s|#r ||0|"||0<!||/|0g>~.~-n| rJ|!rp|$g} | j} ||$B|+s|"||+<nw|*|,kr*dC|"zd |*zkr||+g||+nG||,|+gn*|*|$f|,|+ffD]!\}/}0||/|0g"~$~+~*~,n:t't(}1t't(}2| D]J}$C|$}3B|$s|#s|2|3|$|1|3|$KC|"}.| r1C| d}-g}4t)}5|1D]U\}*}6|*|-kr|4|*n2|2|*}5|5D]}7|"||7<|6|5z}6||*|6~6~*V~5|4r|| dt |4kr|4d!|4D]Z}*|1|*}8dt |8cxkrd"krnnt3|8| d}8|d#|8D||8~8[|} | j} ~4~-n|!r8t|1}9|1|9}:t |:dkrt|:} || nd$|.zd%t |:zt |:z|9zkrPt |:d"krt3|:| d}:|d&|:D|} ||:nat |:d"krt|:| g} n|:g} ||9|:|| ~:~9| j} |1D]9\}*}6|2|*}5|5D]}7|"||7<|6|5z}6||*|6~*~6: t t Dkrn|rPt3|d!D]*}*|d't7||*D+|g};|roDfd(|D};|;rUdt |;cxkrd"krnn|;| d|d)|;D||;n|o~;|r |st |}|>|=ks/|=|=|>z t9|=|>z zz|=t9|=zkrt||}" t |d*krt3||d}nt|}|j}?d}|"|vr |?}"|"Dvr|?}"|"Dv| |" DS)-aOrder nodes in dask graph This produces an ordering over our tasks that we use to break ties when executing. We do this ahead of time to reduce a bit of stress on the scheduler and also to assist in static analysis. This currently traverses the graph as a single-threaded scheduler would traverse it. It breaks ties in the following ways: 1. Begin at a leaf node that is a dependency of a root node that has the largest subgraph (start hard things first) 2. Prefer tall branches with few dependents (start hard things first and try to avoid memory usage) 3. Prefer dependents that are dependencies of root nodes that have the smallest subgraph (do small goals that can terminate quickly) Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> dsk = {'a': 1, 'b': 2, 'c': (inc, 'a'), 'd': (add, 'b', 'c')} >>> order(dsk) {'a': 0, 'c': 1, 'b': 2, 'd': 3} Nc2i|]}|t|S)r ).0krs *lib/python3.11/site-packages/dask/order.py zorder..{s&AAA+C33AAAz2Cycle detected between the following keys: -> %sz -> c34K|]}t|VdSN)str)rxs r zorder..s(33SVV333333rch|] \}}|| Srrrrvs r zorder..s!<<<1!._fs Drc Zi|](\}}\}}}}}|||z ||z | |t|f)Sr) StrComparable)rkeynum_dependentstotal_dependents_ min_heights max_heightss rrzorder..sk C"   ! [ ( [ (    #    rc3^K|]'\}}||t||fV(dSr len)rr0val dependentsmetricss rr#zorder..sX  S #jo&& 5      rr"rtuplecJt|t|z |z|d t|fS)zChoose a path from our starting task to our tactical goal This path is connected to a large goal, but focuses on completing a small goal and being memory efficient. Nr8r/)r"rr:r; num_neededs rdependents_keyzorder..dependents_keys` '''  1  \!_!5!5 5 1 EQZ]N !    rc Jt|}|\}}}}}||z||z | |t|z |z||t|fS)z}Choose which dependency to run as part of a reverse DFS This is very similar to both ``initial_stack_key``. r?) r"r1r2r3r4r5rr:r;r@total_dependenciess rdependencies_keyzorder..dependencies_keys '''Z]++ AJ     [ ( [ (  " " Sa11 1JqM A   !    rrcLi|] \}\}}}}}||z dz||z z!S)r(r)rr0r2r3r4root_total_dependenciesrCs rrzorder..sa     C    $'9#'> > B+- /   rr0FTi)r0reversec|Sr r)r"partition_keyss rzorder..s PQARr)rHdc3K|]}|gVdSr rrdeps rr#zorder..s$/F/F#/F/F/F/F/F/Fr c3K|]}|gVdSr rrRs rr#zorder..s$/J/J#/J/J/J/J/J/Jrc34K|]}t|VdSr )list)rels rr#zorder..s("P"P488"P"P"P"P"P"Prcg|]}|v| Srr)rr"results r zorder..sJJJ!6//!///rc3K|]}|gVdSr rrRs rr#zorder..s$#@#@cSE#@#@#@#@#@#@ri')r)r r*r rr+)r"rrr<)dictr ndependencies graph_metricsr8r RuntimeErrorjoinitemsr robjectorder __getitem__rXminpopappendextendrsetupdateaddclear differencesortedsortreversedr)Grrcycle root_nodesrootr-o init_stackinitial_stack_keyrArDi inner_stackinner_stack_pop inner_stacksinner_stacks_appendinner_stacks_extendinner_stacks_pop next_nodes outer_stackouter_stack_extendouter_stack_popseen seen_updateseen_addsingles singles_clear later_singleslater_singles_appendlater_singles_clearset_differenceis_init_sorteddepsadd_to_inner_stackitemprocess_singlesrSsingle deps_singles singles_keysparent already_seenr0dep2key2prev_keyitem_keyrd dep_poolspossible_singlespkeynow_keyspsinglesvalsspoolmin_keymin_pool outer_depsprev_lenNminit_stack_popr:r;r@rKr[rFrCsG`` @@@@@@@rrereYsj8  s))CAAAASAAA l++J%2<%L%L"J"L*6HIIG 7||s3xxd## Ann33U33333 4   =< 0 0 2 2<<>d# !:--#**4000-~l4.@&IIs4yy////4/////'.. &t1A4 P P P(..t444# D)))"! #!z+b/'B! &*OO#$F4LFA%d+D)-&!#'11C&sOOOq0OOOO*/ $ ..00 "-/ "& ,F'' '1&'9 )*vQ' 1'355 *31 4"<00A55,8 '1&'9!-3=f3EL$, L0D(##%%%%~dF33%*""&G +7G +.9%SL%STTe"-~c'llFCC %\7R7R7R7RSSS99F%V_F*N .z&/A6 J J $  -,V444 #'1&'9 )*vQ'1'355 *31 415/;d/B #/ !O(+<'8'8C ::DAq#-a=:#2!6$8$8$;$;$;$;15GAJJ *1 4 4aS 9 9 9 9$ ****) 6'*e *5/  )$/<,0GDMM D[[Q1E-ES-P-P//777$HTNNNN&t,33TF;;;;&)3Z$$>66DAq&qM00!5555sDD(,, #.s#3#3 --C)#.D%c?8?8(.223777dO'',,,,)$/M&-k!n=H!H"uuH%.__%6%6 & & T>>$OOC0000'7'D 3t992222s22222'-dPT'U'U'U///F/F/F/F/FFFF'K--- $&6&6&8&8 *5/ ((*#:"%i..#,==#9#9x==A--*.x..K'K 4444MBX,>X,NQX,XXX #8}}s22+1$,.$,",","0//J/J/J/J/JJJJ*:*:*<* &w/66x@@@'K 444$g*5/%.__%6%6&& T#3C#8!)..A)-GAJJ("3..t444s Y &v |  F + +   j$777 Q Q#""P"Phz#6O6O"P"P"PPPPP        JJJJ__%6%6JJJJ s:,,,,,,,,,OOOEEE###@#@Z#@#@#@@@@ J'''      ":HZJ' F;;JJA1 AAvva!es1q5zz11AAJ>>:+<===:&&#J4EtTTT !*-- '^N!N :  !>##Dfnn!>##Dfnn4   e R!h MrMapping[Key, set[Key]]r:rCMapping[Key, int])dict[Key, tuple[int, int, int, int, int]]cid|D}g}|j}|j}|D]G\}}|s@||} d| | ddf|<||D]%} || xxdzcc<|| s || &H|r|}||} t| dkr&| \} | \} }}}}d| z||d|zd|zf|<nut fd||D\}}}}}dt |zt |t|dt |zdt|zf|<||D]%} || xxdzcc<|| s || &|S)aUseful measures of a graph used by ``dask.order.order`` Example DAG (a1 has no dependencies; b2 and c1 are root nodes): c1 | b1 b2 \ / a1 For each key we return: 1. **total_dependents**: The number of keys that can only be run after this key is run. Note that this is only exact for trees. (undirected) cycles will cause double counting of nodes. Therefore, this metric is an upper bound approximation. 1 | 2 1 \ / 4 2. **min_dependencies**: The minimum value of the total number of dependencies of all final dependents (see module-level comment for more). In other words, the minimum of ``ndependencies`` of root nodes connected to the current node. 3 | 3 2 \ / 2 3. **max_dependencies**: The maximum value of the total number of dependencies of all final dependents (see module-level comment for more). In other words, the maximum of ``ndependencies`` of root nodes connected to the current node. 3 | 3 2 \ / 3 4. **min_height**: The minimum height from a root node 0 | 1 0 \ / 1 5. **max_height**: The maximum height from a root node 0 | 1 0 \ / 2 Examples -------- >>> inc = lambda x: x + 1 >>> dsk = {'a1': 1, 'b1': (inc, 'a1'), 'b2': (inc, 'a1'), 'c1': (inc, 'b1')} >>> dependencies, dependents = get_deps(dsk) >>> _, total_dependencies = ndependencies(dependencies, dependents) >>> metrics = graph_metrics(dependencies, dependents, total_dependencies) >>> sorted(metrics.items()) [('a1', (4, 2, 3, 1, 2)), ('b1', (2, 3, 3, 1, 1)), ('b2', (1, 2, 2, 0, 0)), ('c1', (1, 3, 3, 0, 0))] Returns ------- metrics: Dict[key, Tuple[int, int, int, int, int]] c8i|]\}}||t|Srr7r%s rrz!graph_metrics..Os)@@@1a@!SVV@@@rr(rc3(K|] }|V dSr r)rrr[s rr#z graph_metrics..vs'CCfVnCCCCCCr)rcrhrir8zipsumrgmax)rr:rCr@current current_popcurrent_appendr0rr9childparentsrr2min_dependenciesmax_dependenciesr4r5total_dependents_min_dependencies_max_dependencies_ min_heights_ max_heights_r[s @rr`r`s>bF@@ (8(8(:(:@@@JG+K^N%%''** T *$S)Cc31-F3K%c* * *5!!!Q&!!!!%(*"N5))) %&kmmS/ w<<1  IVv     $$  KK F3KKCCCC:c?CCCD  !!!C)***%&&%&&C %%%C %%% F3K"#& & &E u    "   e$ &u%%%K %&L Mr%tuple[dict[Key, int], dict[Key, int]]c i}i |D]\}}t|||<|sd |<|}g}|j}|j} D]0} || D]%} || xxdzcc<|| s || &1|ra|} dt fd|| Dz | <|| D]%} || xxdzcc<|| s || &|a| fS)aGNumber of total data elements on which this key depends For each key we return the number of tasks that must be run for us to run this task. Examples -------- >>> inc = lambda x: x + 1 >>> dsk = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')} >>> dependencies, dependents = get_deps(dsk) >>> num_dependencies, total_dependencies = ndependencies(dependencies, dependents) >>> sorted(total_dependencies.items()) [('a', 1), ('b', 2), ('c', 3)] Returns ------- num_dependencies: Dict[key, int] total_dependencies: Dict[key, int] r(c3(K|] }|V dSr r)rrr[s rr#z ndependencies..s'KKfUmKKKKKKr)rcr8copyrhrir) rr:r@rr&num_dependenciesrrrr0rr[s @rr_r_s|,J F""$$1A 1  F1I!((G+K^N'' o ' 'F v   ! #   f% 'v&&& ' 'kmm#KKKKc9JKKKKKKs  o ' 'F v   ! #   f% 'v&&& ' V ##rc2eZdZUdZdZded<d dZd d Zd S) r/aWrap object so that it defaults to string comparison When comparing two objects of different types Python fails >>> 'a' < 1 Traceback (most recent call last): ... TypeError: '<' not supported between instances of 'str' and 'int' This class wraps the object so that, when this would occur it instead compares the string representation >>> StrComparable('a') < StrComparable(1) False objr rc||_dSr r)selfrs r__init__zStrComparable.__init__s rotherrboolc |j|jkS#t$r-t|jt|jkcYSwxYwr )r Exceptionr!)rrs r__lt__zStrComparable.__lt__sR 28ei' ' 2 2 2tx==3uy>>1 1 1 1 2s4A A N)rr )rr rr)__name__ __module__ __qualname____doc__ __slots____annotations__rrrrrr/r/sX I HHH222222rr/ OrderInfo)reagenum_data_when_runnum_data_when_releasednum_dependencies_freedrvMapping[Key, int] | None&tuple[dict[Key, OrderInfo], list[int]]cl |t|\}}nt|}|t||}g}d}i iii d|D}t t ||jD]\}}||||<d} ||D]6} || xxdzcc<|| dkr||| z | <|| <| dz } 7| |<||r || dz z}vd |<||<|| z} fd|D} | |fS)aSimulate runtime metrics as though running tasks one at a time in order. These diagnostics can help reveal behaviors of and issues with ``order``. Returns a dict of `namedtuple("OrderInfo")` and a list of the number of outputs held over time. OrderInfo fields: - order : the order in which the node is run. - age : how long the output of a node is held. - num_data_when_run : the number of outputs held in memory when a node is run. - num_data_when_released : the number of outputs held in memory when the output is released. - num_dependencies_freed : the number of dependencies freed by running the node. N)rrc4i|]\}}|t|Srr7)rr0r9s rrzdiagnostics..s$CCCHC#s3xxCCCrrGr(c ni|]1\}}|t|||||2Sr)r)rr0r9rfreedreleasepressure runpressures rrzdiagnostics..sX    C Y S;s+_S-A5:     r)r rrerc enumeraterprfri)rrvrr:pressure num_in_memoryr@ryr0releasedrSrvrrrrs @@@@r diagnosticsrs$#+C== jj!,// y #L 1 1 1HM CKO ECC 0@0@0B0BCCCJF3AM:::;;&&3 &&&( C$  C sOOOq OOO#!##qv:C'4$A c c? & X\ )MMCH#0OC X %MM           B x<rr+cdSr rrrrr-r-sCrtaskr c t|rt|dsJg}|ddD]}t|ttfr||4t|t rHt|r#|t|{||t|tr|d|Dtg|RS|S)Nrr(c,g|]}t|Sr) _convert_task)res rr\z!_convert_task..-s > > >aq!1!1 > > >r) rcallable isinstancer!intrir<rrXr-)rnew_specrYs rrr s  d||Q      qrr( @ @B"sCj)) @####B&& @"::(OOM"$5$56666OOB''''B%% @ > >2 > > >???X rr^ci}|D]\}}|}t|||<t|t|krtd|S)zTake a dask graph and replace callables with a dummy function and remove payload data like numpy arrays, dataframes, etc. z)Sanitization failed to preserve topology.)rcrr ra)rnewr0valuesnew_keys r sanitize_dskr3sg Cyy{{-- V$V,,G }} %%FGGG Jrr )rrrrrr)rrr:rrCrrr)rrr:rrr)NN)rrrvrrrrr)rr+)rr rr )rrrr^) __future__r collectionsrrcollections.abcrrmathrtypingr r dask.corer r r rr dask.typingrrer`r_r/rrr-rrrrrrs""""""LZ0///////33333333PPPPPPPPPPPPPP :>a a a a a HEEEEP.$.$.$.$b22222222> J   #'9=88888v&      r