U /e@sddlZddlZddlZddlmZddlmZmZmZddl m Z m Z m Z m Z mZmZmZddZddZd"d d Zd d Zd#ddZd$ddZddZddZd%ddZGdddeZejZddeeeeeefddZddZGd d!d!ZdS)&N)Enum)configcoreutils)flattenget_dependencies ishashableistask reverse_dictsubstoposortc Cst|ttfs|g}t}t}i}ttt|}|rg}|D]L}t||dd}||||<|||<|D] } | |krj|| || qjq@|}q4||fS)a<Return new dask with only the tasks required to calculate keys. In other words, remove unnecessary tasks from dask. ``keys`` may be a single key or list of keys. Examples -------- >>> def inc(x): ... return x + 1 >>> def add(x, y): ... return x + y >>> d = {'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)} >>> dsk, dependencies = cull(d, 'out') >>> dsk # doctest: +ELLIPSIS {'out': (, 'x', 10), 'x': 1} >>> dependencies # doctest: +ELLIPSIS {'out': ['x'], 'x': []} Returns ------- dsk: culled dask graph dependencies: Dict mapping {key: [deps]}. Useful side effect to accelerate other optimizations, notably fuse. TZas_list) isinstancelistsetdictrraddappend) dskkeysseen dependenciesoutworknew_workkZdependencies_kdr5/tmp/pip-unpacked-wheel-dbjnr7gq/dask/optimization.pyculls$  rcCst|d}|tkrDdd|dddD}||dd|S|tkrt|ddkrt|ddtrdd|dddD}||ddd|f|dddSdSdS) zCreate new keys for fused tasksrcSsg|]}t|qSrr key_split.0xrrr Isz5default_fused_linear_keys_renamer..N-cSsg|]}t|qSrr r"rrrr%Ms)typestrrjointuplelenr)rtypnamesrrr!default_fused_linear_keys_renamerEs  *r0TcsB|dk r.t|ts.t|ts"|g}tt|}|dkrHfddD}i}t}D]x}||}t|dk}|D]Z} |dk r| |kr|| qr| |kr|| =|| qr|r|| qr| |krr||| <qrqVg} ttt| } |rd| \} }| |g} ||kr(| |}| |=| |q| | | krX| | } || =| | q0| | qdd| D}|dkrt} n|dkrd} n|} i}t}t}d}| D]} | dk r| | }|dk o|ko||k}| } | }| rF| }||| | ||| t|| |}|| |} q|| |r|||<||| <|| ||<|h|| <|| n||| <q D]\}}||kr|||<q|r:| D]J\}}||@D]6}||}||||t||||||<qؐq|dk r:||D]}||=||=q&||fS)aReturn new dask graph with linear sequence of tasks fused together. If specified, the keys in ``keys`` keyword argument are *not* fused. Supply ``dependencies`` from output of ``cull`` if available to avoid recomputing dependencies. **This function is mostly superseded by ``fuse``** Parameters ---------- dsk: dict keys: list dependencies: dict, optional {key: [list-of-keys]}. Must be a list to provide count of each key This optional input often comes from ``cull`` rename_keys: bool or func, optional Whether to rename fused keys with ``default_fused_linear_keys_renamer`` or not. Renaming fused keys can keep the graph more understandable and comprehensive, but it comes at the cost of additional processing. If False, then the top-most key will be used. For advanced usage, a func is also accepted, ``new_key = rename_keys(fused_key_list)``. Examples -------- >>> def inc(x): ... return x + 1 >>> def add(x, y): ... return x + y >>> d = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')} >>> dsk, dependencies = fuse(d) >>> dsk # doctest: +SKIP {'a-b-c': (inc, (inc, 1)), 'c': 'a-b-c'} >>> dsk, dependencies = fuse(d, rename_keys=False) >>> dsk # doctest: +ELLIPSIS {'c': (, (, 1))} >>> dsk, dependencies = fuse(d, keys=['b'], rename_keys=False) >>> dsk # doctest: +ELLIPSIS {'b': (, 1), 'c': (, 'b')} Returns ------- dsk: output graph with keys fused dependencies: dict mapping dependencies after fusion. Useful side effect to accelerate other downstream optimizations. Ncsi|]}|t|ddqSTr rr#rrrr szfuse_linear..r(cSsi|]\}}|t|qSrrr#rvrrrr5sTF)rrrrr-rrmapreverseditemspopitempoprreverser0updateremover )rrr rename_keys child2parent unfusibleparentdepshas_many_childrenchildchains parent2childchain key_renamerrvZfusedaliasesZ is_renamedZnew_keyvalkeyZold_keyrr4r fuse_linearTs0                              rPcCs8|dkrtSt|tr|St|ttfs0|g}t|SN)rrr)r$rrr _flat_sets rRc s2r,ttttr,ddDt|}dkrNfddD|rp|fddDtfdd|Dd}i}|D]L}|}||@D]*}||kr||} n|} t ||| }q|||<q| } D]<\}}|| kr||@D]} t || || }q || |<q| S) aReturn new dask with the given keys inlined with their values. Inlines all constants if ``inline_constants`` keyword is True. Note that the constant keys will remain in the graph, to remove them follow ``inline`` with ``cull``. Examples -------- >>> def inc(x): ... return x + 1 >>> def add(x, y): ... return x + y >>> d = {'x': 1, 'y': (inc, 'x'), 'z': (add, 'x', 'y')} >>> inline(d) # doctest: +ELLIPSIS {'x': 1, 'y': (, 1), 'z': (, 1, 'y')} >>> inline(d, keys='y') # doctest: +ELLIPSIS {'x': 1, 'y': (, 1), 'z': (, 1, (, 1))} >>> inline(d, keys='y', inline_constants=False) # doctest: +ELLIPSIS {'x': 1, 'y': (, 'x'), 'z': (, 'x', (, 'x'))} cSsi|]\}}|t|qSrr6r7rrrr5 szinline..Ncsi|]}|t|qSrr2r3r4rrr5sc3s6|].\}}t|r|ks*|st|s|VqdSrQ)rr r7rrrr s zinline..csi|]}|kr||qSrrr3r4rrr5s)r) rnextitervaluesrr;rRr?r r copy) rrinline_constantsrZ replaceorderZkeysubsrOrNdepreplaceZdsk2itemrrSrinlines:   r]FcssStt|dkr2fddD}t|fddfddD}|rt|||d|D] }|=qxS) aInline cheap functions into larger operations Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> double = lambda x: x * 2 >>> dsk = {'out': (add, 'i', 'd'), # doctest: +SKIP ... 'i': (inc, 'x'), ... 'd': (double, 'y'), ... 'x': 1, 'y': 1} >>> inline_functions(dsk, [], [inc]) # doctest: +SKIP {'out': (add, (inc, 'x'), 'd'), 'd': (double, 'y'), 'x': 1, 'y': 1} Protect output keys. In the example below ``i`` is not inlined because it is marked as an output key. >>> inline_functions(dsk, ['i', 'out'], [inc, double]) # doctest: +SKIP {'out': (add, 'i', (double, 'y')), 'i': (inc, 'x'), 'x': 1, 'y': 1} Ncsi|]}|t|qSrr2r3r4rrr5Tsz$inline_functions..cs,zt|WStk r&YdSXdS)NF) functions_ofissubset TypeError)r8)fast_functionsrr inlinableWsz#inline_functions..inlinablecs4g|],\}}t|r|r|kr|r|qSr)r r7) dependentsrboutputrrr%]s z$inline_functions..)rYr)rr r;r])rrdrarYrrrr)rcrrarbrdrinline_functions1s* recCst|dr|j}q|S)Nfunc)hasattrrf)rfrrrunwrap_partialls rhcCstt}|g}tth}|rpg}|D]H}t||kr t|r^|t|d||ddq ||q |}q|S)a1Set of functions contained within nested task Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> mul = lambda x, y: x * y >>> task = (add, (mul, 1, 2), (inc, 3)) # doctest: +SKIP >>> functions_of(task) # doctest: +SKIP set([add, mul, inc]) rr(N)rrr,r)r rrhextend)ZtaskfuncsrZsequence_typesrrrrr^rs   r^xc st|}t|}t|}r$d8fdd}|tkr~t|}dd|D}||t|}||d |}||S|t krt |dkrt |dtrt|}dd|D}||t|}||dd |}||f|d d Sd S) zCreate new keys for ``fuse`` tasks. The optional parameter `max_fused_key_length` is used to limit the maximum string length for each renamed key. If this parameter is set to `None`, there is no limit. cs>r:t|kr:t|ddd}|dd|}|S)Nr$r')r-hash)Zkey_nameZ name_hashmax_fused_key_lengthrr_enforce_max_key_limitsz:default_fused_keys_renamer.._enforce_max_key_limitcSsh|]}t|qSrr r3rrr sz-default_fused_keys_renamer..r'rcSsh|]}t|qSrr r3rrrrrsr(N) r:rUr)r*rr!discardsortedrr+r,r-r) rrpitZ first_keyr.rqZ first_namer/Zconcatenated_namerrordefault_fused_keys_renamers,     "   rvc@seZdZdZedddZdS)Defaultr)returncCsdS)Nz rselfrrr__repr__szDefault.__repr__N)__name__ __module__ __qualname__tokenr*r{rrrrrwsrwc > stddkr|fS|dk rDt|tsDt|ts8|g}tt|}|tkrbtd}|tk sbt|tkrtd}|tk st|tkrtd}|tk st|dkr|d}|tkrtd}|tk st|dkrd|t |d }|tkr td }|tk s t|dkrd}|r&|s.|fS|tkrPtd }|tk sPt|d kr`t } n(|dkrpd} nt |st d n|} | dk }|dkrfddD} nt |} i} | D]F\} } | D]*}|| kr| g| |<n| || qt| | | <qdd| D}|r*||8}D]4\} }t|tk r2t|tjtfs2|| q2|s|rtdd| Dr| fS}i}g}g}| j}|j}|j}|j}|j}|j}|j}|j}|j}|j}|r:|}||||kr| |d}q||||| |@|d}||kr|| |@} | rr|| |}|d}|| |@} qH||||||r|gndd d d d| ||fq*|| |}!|!|}"|!|"} t| }#|#d krx|\}$}%}&}'}(})}*}+t|+},|*|,d krdkr"nn|,d }*|"|+O}"t|"|,k}-|-sD|*d 7}*|)|*|'|kr|-sf|'|krt||$|%}.|!|$|!||$O}!||$=||$|r|&||&||<||$d|r|-r|||.|&|'|(|)|*|"fn |||.|&|'d |(|)d |*|"fn |.||<qnb|%||$<||$|r6|*t|d krFt|d }*|||||r\|gndd |(d |*|"fnqng}&d }'d}(d}/d})d}*t}+d}0||# d}1||# d=|1D]n\}2}2}2}3}4}5}6}7|3d kr|/d 7}/n|3|'kr|3}'|(|47}(|)|57})|*|67}*t|7|0krt|7}0|+|7O}+qt|+},|*t |#d t!d|,|07}*|*|,d krhdkrtnn|,d }*|"|+O}"t|"|,k}-|-s|*d 7}*|)|*|'|kr|/|kr|(|kr|'|kr|-s|'|kr|}.t}8|1D]V}9|9d}:t|.|:|9d }.||:=|8||:O}8||:|r||:d|&|9dq|!| 8}!|!|8O}!|rh|&||&||<|r|||.|&|'d |(|)d |*|"fn |.||<qn|1D]"}9|9d ||9d<||9dq|r6|(|kr|}(|*t|d krt|d }*|||||r|gndd |(d |*|"fnq| |d}q*q|rPt"||| ||| r|D]P\};}<| |<}=|=dk r^|=|kr^||;||=<|=||;<| |;| |=<|=h| |;<q^|| fS)am Fuse tasks that form reductions; more advanced than ``fuse_linear`` This trades parallelism opportunities for faster scheduling by making tasks less granular. It can replace ``fuse_linear`` in optimization passes. This optimization applies to all reductions--tasks that have at most one dependent--so it may be viewed as fusing "multiple input, single output" groups of tasks into a single task. There are many parameters to fine tune the behavior, which are described below. ``ave_width`` is the natural parameter with which to compare parallelism to granularity, so it should always be specified. Reasonable values for other parameters will be determined using ``ave_width`` if necessary. Parameters ---------- dsk: dict dask graph keys: list or set, optional Keys that must remain in the returned dask graph dependencies: dict, optional {key: [list-of-keys]}. Must be a list to provide count of each key This optional input often comes from ``cull`` ave_width: float (default 1) Upper limit for ``width = num_nodes / height``, a good measure of parallelizability. dask.config key: ``optimization.fuse.ave-width`` max_width: int (default infinite) Don't fuse if total width is greater than this. dask.config key: ``optimization.fuse.max-width`` max_height: int or None (default None) Don't fuse more than this many levels. Set to None to dynamically adjust to ``1.5 + ave_width * log(ave_width + 1)``. dask.config key: ``optimization.fuse.max-height`` max_depth_new_edges: int or None (default None) Don't fuse if new dependencies are added after this many levels. Set to None to dynamically adjust to ave_width * 1.5. dask.config key: ``optimization.fuse.max-depth-new-edges`` rename_keys: bool or func, optional (default True) Whether to rename the fused keys with ``default_fused_keys_renamer`` or not. Renaming fused keys can keep the graph more understandable and comprehensive, but it comes at the cost of additional processing. If False, then the top-most key will be used. For advanced usage, a function to create the new name is also accepted. dask.config key: ``optimization.fuse.rename-keys`` fuse_subgraphs : bool or None, optional (default None) Whether to fuse multiple tasks into ``SubgraphCallable`` objects. Set to None to let the default optimizer of individual dask collections decide. If no collection-specific default exists, None defaults to False. dask.config key: ``optimization.fuse.subgraphs`` Returns ------- dsk output graph with keys fused dependencies dict mapping dependencies after fusion. Useful side effect to accelerate other downstream optimizations. zoptimization.fuse.activeFNzoptimization.fuse.ave-widthzoptimization.fuse.max-heightz%optimization.fuse.max-depth-new-edgesg?zoptimization.fuse.max-widthr(zoptimization.fuse.subgraphszoptimization.fuse.rename-keysTz)rename_keys must be a boolean or callablecsi|]}|t|ddqSr1r2r3r4rrr5:szfuse..cSs h|]\}}t|dkr|qS)r()r-)r#rvalsrrrrrGs zfuse..css|]}tt|dkVqdS)r(N)r-r)r#r8rrrrTPszfuse..rr&)#rgetrrrr_defaultAssertionErrormathlogrvcallabler`rr;rr)r,numbersNumberr*rsallrWrXr=rr@rir-r intminmax_inplace_fuse_subgraphs)>rrrZ ave_width max_widthZ max_heightZmax_depth_new_edgesrAZfuse_subgraphsrKrEZrdepsrrr8Z reduciblerL fused_treesZ info_stackZchildren_stackZdeps_popZ reducible_addZ reducible_popZreducible_removeZfused_trees_popZinfo_stack_appendZinfo_stack_popZchildren_stack_appendZchildren_stack_extendZchildren_stack_poprDrGchildrenZ deps_parentedgesZ num_childrenZ child_keyZ child_taskZ child_keysheightwidthZ num_nodesZfudgeZchildren_edgesZnum_children_edgesZ no_new_edgesrNZnum_single_nodesZ max_num_edgesZ children_info_Z cur_heightZ cur_widthZ cur_num_nodesZ cur_fudgeZ cur_edgesZ children_depsZ child_infoZ cur_childZroot_keyZ fused_keysaliasrr4rfusesVH                                                          rcsi}t}D]x}||}t|dk} |D]Z} |dk rJ| |krJ|| q*| |krd|| =|| q*| rt|| q*| |kr*||| <q*qg} dd|D} |rD|\} }| |g} ||kr||}| |=| |q| | | kr | | } || =| | qd}| D].}|t|7}|dkr| | qqq| D]} fdd| D}| d}|| d}||<| ddD]}||=|=qt |}t |||f||<|rHg}| D].}||d}|r| |n ||q|||<qHdS) zKSubroutine of fuse. Mutates dsk, dependencies, and fused_trees inplacer(NcSsi|]\}}||qSrrr7rrrr5sz+_inplace_fuse_subgraphs..rcsi|]}||qSrrr3r4rrr5sr&F) rr-rr;r<r=rr>r r,SubgraphCallableri)rrrrrArBrCrDrErFrGrHrIrJZntasksrOZsubgraphoutkeyZ inkeys_setrinkeysZchain2Zsubchainrr4rrmsh                 rc@sNeZdZdZdZdddZddZdd Zd d Zd d Z ddZ ddZ dS)raDCreate a callable object from a dask graph. Parameters ---------- dsk : dict A dask graph outkey : hashable The output key from the graph inkeys : list A list of keys to be used as arguments to the callable. name : str, optional The name to use for the function. )rrrnameNcCs2||_||_||_|dkr(dt}||_dS)Nzsubgraph_callable-)rrruuidZuuid4r)rzrrrrrrr__init__s zSubgraphCallable.__init__cCs|jSrQ)rryrrrr{szSubgraphCallable.__repr__cCs<t|t|ko:|j|jko:|j|jko:t|jt|jkSrQ)r)rrrrrzotherrrr__eq__s  zSubgraphCallable.__eq__cCs ||k SrQrrrrr__ne__szSubgraphCallable.__ne__cGsJt|t|jks,tdt|jt|ft|j|jtt|j|S)NzExpected %d args, got %d) r-r ValueErrorrrrrrzip)rzargsrrr__call__szSubgraphCallable.__call__cCst|j|j|j|jffSrQ)rrrrrryrrr __reduce__szSubgraphCallable.__reduce__cCstt|jt|j|jfSrQ)rnr,r frozensetrrryrrr__hash__szSubgraphCallable.__hash__)N) r|r}r~__doc__ __slots__rr{rrrrrrrrrrs r)NNT)NTN)NFN)rk) rrrenumrZdaskrrrZ dask.corerrrr r r r rr0rPrRr]rerhr^rvrwrrrrrrrrrsD $ 3  C ; ' .G