U /e-1@sPddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZmZmZmZddlZddlmZdd lmZmZmZdd lmZdd lmZeeZ d(d dZ!GdddZ"dgZ#ed fddZ$e%e&e'e(fZ)dddddddZ*d)ddddddZ+e,fddZ-d d!Z.d"e/e0fdfd#d$Z1dd%d&d'Z2dS)*) annotationsN) defaultdict)partial)cycle)Any)concatdropgroupbymerge)SubgraphCallable)is_namedtuple_instanceparse_timedelta stringifyrpc)AllTc sddlmt}t}|dd|D}t}t}t|t|t|krtt} t} t} |D]^\} } | |krqpz,t t| |}| | | || | <Wqpt k r| | YqpXqp| r|| O}fdd| D}i|D]\}}z|IdH}WnXt k r:| |YnHtk rr}ztd|| |W5d}~XYnX|dq|fd d | DO}|q:fd d|D} || t|fS) aGather data directly from peers Parameters ---------- who_has: dict Dict mapping keys to sets of workers that may have that key rpc: callable Returns dict mapping key to value See Also -------- gather _gather r)get_data_from_workercSsi|]\}}|t|qS)set.0kvrr:/tmp/pip-unpacked-wheel-g426oqom/distributed/utils_comm.py ,sz'gather_from_workers..c s6i|].\}}|tj||ddd|dqS)F)who serializersZmax_connectionszget-data-from-)name)asyncioZ create_task)raddresskeys)rrrrrrr?s Nz9Got an unexpected error while collecting from workers: %sdatacsh|]\}}|kr|qSrrr)responserr [sz&gather_from_workers..csi|]}|t|qSr)list)rr)original_who_hasrrr^s)Zdistributed.workerrritemsdictlenrr$randomchoiceappend IndexErroraddOSError ValueErrorloggerinfoupdate)who_hasrcloserrZ bad_addressesZmissing_workersresultsZ all_bad_keysdrevZbad_keyskey addressesaddrZ coroutinesworkercrer)rr%r"rrrrgather_from_workerssT    r?c@s eZdZdZddZddZdS) WrappedKeyaInterface for a key in a dask graph. Subclasses must have .key attribute that refers to a key in a dask graph. Sometimes we want to associate metadata to keys in a dask graph. For example we might know that that key lives on a particular machine or can only be accessed in a certain way. Schedulers may have particular needs that can only be addressed by additional metadata. cCs ||_dSNr8)selfr8rrr__init__mszWrappedKey.__init__cCst|jd|jdS)Nz('z'))type__name__r8)rCrrr__repr__pszWrappedKey.__repr__N)rF __module__ __qualname____doc__rDrGrrrrr@bs r@c s0t|tstt|tstttdd|D}tt|\}}ttdt |t |}tdt |7<tt|||}t d|}dd|D}fdd|Dz&tfdd |DIdH} W5 D]} | IdHqXtd d| D} d dt d |D} || | fS) aScatter data directly to workers This distributes data in a round-robin fashion to a set of workers based on how many cores they have. nthreads should be a dictionary mapping worker identities to numbers of cores. See scatter for parameter docstring css|]\}}|g|VqdSrAr)rwZncrrr sz%scatter_to_workers..rcSs i|]\}}|dd|DqS)cSsi|]\}}}||qSrr)r_r8valuerrrrsz1scatter_to_workers...r)rr;rrrrrsz&scatter_to_workers..csi|]}||qSrr)rr:rrrrsNcs"g|]\}}|j|dqS))r!report)Z update_data)rrr)rOrpcsrr s z&scatter_to_workers..css|]}|dVqdS)nbytesNr)rorrrrLscSs i|]\}}|dd|DqS)cSsg|]\}}}|qSrr)rrKrMrrrrQsz1scatter_to_workers...rrrrrrs) isinstancer'AssertionErrorr$rr&zipr_round_robin_counterr(rr valuesZ close_rpcrr ) Znthreadsr!rrOworkersnamesZ worker_iterLr6r=outrRr3r)rOrrPrscatter_to_workersws*    r^rboolzset[WrappedKey])rS byte_keys found_keysreturnc st|}|tkr|s|St|dtkrttfdd|ddD}|d}tfdd|jD}d}r܈rtd dDntd dD}t|j|}t||j||j }|f||Stfd d|DSn"t |r&|fd d |DS|t krV|s:|Sfdd |D} || S|t kr|r~fdd|DS|Sn2t |tr|j} rt| } || S|SdS)zXInner implementation of `unpack_remotedata` that adds found wrapped keys to `found_keys`rc3s|]}t|VqdSrA_unpack_remotedata_innerrir`futuresrrrLsz+_unpack_remotedata_inner..rTNcsi|]\}}|t|qSrrcrrgrrrsz,_unpack_remotedata_inner..rcss|]}t|jVqdSrA)rr8rfrrrrLscss|] }|jVqdSrArBrirrrrLsc3s|]}t|VqdSrArcritemr`rarrrLscsg|]}t|qSrrcrkrmrrrQsz,_unpack_remotedata_inner..csg|]}t|qSrrcrkrmrrrQscsi|]\}}|t|qSrrcrrmrrrs)rEtupler rr2dskr&inkeysZoutkeyrr collection_typesr' issubclassr@r8rr-) rSr`ratypargsZscroZ future_keysrpZoutsrr)r`rarhrrds`            rdFztuple[Any, set])rSr`rbcCst}t||||fS)aUnpack WrappedKey objects from collection Returns original collection and set of all found WrappedKey objects Examples -------- >>> rd = WrappedKey('mykey') >>> unpack_remotedata(1) (1, set()) >>> unpack_remotedata(()) ((), set()) >>> unpack_remotedata(rd) ('mykey', {WrappedKey('mykey')}) >>> unpack_remotedata([1, rd]) ([1, 'mykey'], {WrappedKey('mykey')}) >>> unpack_remotedata({1: rd}) ({1: 'mykey'}, {WrappedKey('mykey')}) >>> unpack_remotedata({1: [rd]}) ({1: ['mykey']}, {WrappedKey('mykey')}) Use the ``byte_keys=True`` keyword to force string keys >>> rd = WrappedKey(('x', 1)) >>> unpack_remotedata(rd, byte_keys=True) ("('x', 1)", {WrappedKey('('x', 1)')}) )rrd)rSr`rarrrunpack_remotedatasrucst|}z t|r&|kr&|WSWntk r<YnX|tkr^|fdd|DS|tkr~fdd|DSt|r|fdd|DS|SdS)aMerge known data into tuple or dict Parameters ---------- o core data structures containing literals and keys d : dict mapping of keys to data Examples -------- >>> data = {'x': 1} >>> pack_data(('x', 'y'), data) (1, 'y') >>> pack_data({'a': 'x', 'b': 'y'}, data) # doctest: +SKIP {'a': 1, 'b': 'y'} >>> pack_data({'a': ['x'], 'b': 'y'}, data) # doctest: +SKIP {'a': [1], 'b': 'y'} csg|]}t|dqS) key_types pack_datarxr6rwrrrQ#szpack_data..cs i|]\}}|t|dqSrvrxrr|rrr%szpack_data..csg|]}t|dqSrvrxrzr|rrrQ'sN)rErU TypeErrorrqr'r&r )rSr6rwrsrr|rrysrycst|}|tkrH|rHt|drH|dftfdd|ddDS|tkrbfdd|DS|tkrfdd |DSz||WStk r|YSXdS) aXPerform substitutions on a tasks Parameters ---------- o Core data structures containing literals and keys d : dict Mapping of keys to values Examples -------- >>> dsk = {"a": (sum, ["x", 2])} >>> data = {"x": 1} >>> subs_multiple(dsk, data) # doctest: +SKIP {'a': (sum, [1, 2])} rc3s|]}t|VqdSrA subs_multiplerer6rrrL@sz subs_multiple..rTNcsg|]}t|qSrr~rerrrrQBsz!subs_multiple..csi|]\}}|t|qSrr~rrrrrDsz!subs_multiple..)rErncallabler$r'r&getr})rSr6rsrrrr,s(rg?c st|D]}z|IdHWS|k r}zt|p:t|}td|d|d|d|t|d|d|} |dkr| dt|9} t| IdHW5d}~XYqXq|IdHS) a Return the result of ``await coro()``, re-trying in case of exceptions The delay between attempts is ``delay_min * (2 ** i - 1)`` where ``i`` enumerates the attempt that just failed (starting at 0), but never larger than ``delay_max``. This yields no delay between the first and second attempt, then ``delay_min``, ``3 * delay_min``, etc. (The reason to re-try with no delay is that in most cases this is sufficient and will thus recover faster from a communication failure). Parameters ---------- coro The coroutine function to call and await count The maximum number of re-tries before giving up. 0 means no re-try; must be >= 0. delay_min The base factor for the delay (in seconds); this is the first non-zero delay between re-tries. delay_max The maximum delay (in seconds) between consecutive re-tries (without jitter) jitter_fraction The maximum jitter to add to the delay, as fraction of the total delay. No jitter is added if this value is <= 0. Using a non-zero value here avoids "herd effects" of many operations re-tried at the same time retry_on_exceptions A tuple of exception classes to retry. Other exceptions are not caught and re-tried, but propagate immediately. operation A human-readable description of the operation attempted; used only for logging failures Returns ------- Any Whatever `await coro()` returned Nz Retrying z after exception in attempt /z: rTr)rangestrr0r1minr)rsleep) corocount delay_min delay_maxZjitter_fractionZretry_on_exceptions operationZi_tryexdelayrrrretryLs+  $r)rcsXtjd}ttjddd}ttjddd}tt|f||||||dIdHS)zT Retry an operation using the configuration values for the retry parameters zdistributed.comm.retry.countz distributed.comm.retry.delay.mins)defaultz distributed.comm.retry.delay.max)rrrrN)daskconfigrr rr)rrrtkwargsZ retry_countZretry_delay_minZretry_delay_maxrrrretry_operations    r)TNN)F)3 __future__rrloggingr) collectionsr functoolsr itertoolsrtypingrZtlzrrr r Z dask.configrZdask.optimizationr Z dask.utilsr r rZdistributed.corerZdistributed.utilsr getLoggerrFr0r?r@rXr^rnr$r frozensetrqrdruobjectryrEnvironmentErrorIOErrorrrrrrrs8          K, B %% :