U /e!@sUddlmZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z ddl mZddlmZddlmZddlZddlmZddlmZmZmZdd lmZmZdd lmZdd lm Z d d Z!ee"e#j$e!eej%ej&dZ'ej(Z)ddZ*Gddde+Z,iZ-de.d<dddddZ/zddl0Z1e1j23ddZ4Wn(e5k rxddZ4d-ddZYnXddZ6d Z7d!d"Z8d.d$d%d&d'd(Z9d)d*Z:d/d+d,Z;dS)0) annotationsN)HashableMappingSequence)ProcessPoolExecutor)partial)warn)config)MultiprocessingPoolExecutor get_asyncreraise)cullfuse) CPU_COUNT) ensure_dictcCst|j|jffSN)getattr __objclass____name__)mr8/tmp/pip-unpacked-wheel-dbjnr7gq/dask/multiprocessing.py_reduce_method_descriptorsr)protocolcCs tjSr)multiprocessingcurrent_processidentrrrr_process_get_id#src@s0eZdZdZddZddZddZdd Zd S) RemoteExceptionzXRemote Exception Contains the exception and traceback from a remotely run task cCs||_||_dSr) exception traceback)selfrr rrr__init__:szRemoteException.__init__cCst|jd|jS)Nz Traceback --------- )strrr r!rrr__str__>szRemoteException.__str__cCs(tttt|t|jt|jSr)sortedsetdirtypelist__dict__rr$rrr__dir__AszRemoteException.__dir__cCs4zt||WStk r.t|j|YSXdSr)object__getattribute__AttributeErrorrr)r!keyrrr __getattr__DszRemoteException.__getattr__N)r __module__ __qualname____doc__r"r%r,r1rrrrr4s rz&dict[type[Exception], type[Exception]] exceptions Exception)excreturncCsxt|tkr"tt|}|||Sz8t|jjtt|fdt|i}|tt|<|||WStk rr|YSXdS)z6Metaclass that wraps exception type in RemoteExceptionZexception_typeN)r)r5 __class__rr TypeError)r7tbtyprrrremote_exceptionNs       r=cCs|Srrr;rrr_pack_tracebackesr?cCsdt|S)N)joinr format_tbr>rrrr?jscCst||}|dSr)r=)r7r;rrrr ms r c Cspt\}}}t|}z|||f}WnDtk rj}z&t\}}}t|}|||f}W5d}~XYnX|Sr)sysexc_infor? BaseException)edumpsexc_type exc_value exc_tracebackr;resultrrrpack_exceptionrsrLzThe 'multiprocessing.context' configuration option will be ignored on Python 2 and on Windows, because they each only support a single context. cCs:tdd}tjdkr,|dkr(ttttSt|SdS)z+Return the current multiprocessing context.zmultiprocessing.contextspawnwin32N) r getrCplatformr_CONTEXT_UNSUPPORTED UserWarningr get_context)Z context_namerrrrSs    rSTrzSequence[Hashable] | Hashable)dskkeysc  KsT|ptdd}|ptdd}|p.tdd}|pBtddpBt}|dkrtjddkrfd tjd<t} tt|d }t|| |d }d } n*|dk rt d t |t j j rt|}d} t|}t||\} } |rt| || \}} n| }|ptddpt}|ptddpt}z,t|j|j||ft||tt|d| }W5| rN|X|S)aMultiprocessed get function appropriate for Bags Parameters ---------- dsk : dict dask graph keys : object or list Desired results from graph num_workers : int Number of worker processes (defaults to number of cores) func_dumps : function Function to use for function serialization (defaults to cloudpickle.dumps) func_loads : function Function to use for function deserialization (defaults to cloudpickle.loads) optimize_graph : bool If True [default], `fuse` is applied to the graph before computation. pool : Executor or Pool Some sort of `Executor` or `Pool` to use initializer: function Ignored if ``pool`` has been set. Function to initialize a worker process before running any tasks in it. chunksize: int, optional Size of chunks to use when dispatching work. Defaults to 5 as some batching is helpful. If -1, will be computed to evenly divide ready work across workers. chunksizepoolNzmultiprocessing.initializer num_workersZPYTHONHASHSEED)N0Z6640Zuser_initializer)Z mp_context initializerTzThe ``initializer`` argument is ignored when ``pool`` is provided. The user should configure ``pool`` with the needed ``initializer`` on creation.F func_loads func_dumps)Zget_idrGloadsrLZraise_exceptionrV)r rOrosenvironrSrinitialize_worker_processrr isinstancerrXZPoolr rr r_loads_dumpsshutdownr ZsubmitZ _max_workersrrLr )rTrUrYr]r^Zoptimize_graphrXr\rVkwargscontextcleanupZdsk2Z dependenciesZdsk3r_rGrKrrrrOs`&     rOcCs"tjd}|dk r|jdS)NZnumpy)rCmodulesrOrandomseed)nprrrdefault_initializers rncCst|dk r|dS)zE Initialize a worker process before running any tasks in it. N)rnr[rrrrbsrb)N)NNNTNNN)N)< __future__rcopyregrZmultiprocessing.poolr`picklerCr collections.abcrrrconcurrent.futuresr functoolsrwarningsrZ cloudpickleZdaskr Z dask.localr r r Zdask.optimizationr rZ dask.systemrZ dask.utilsrrr)r'unionrGHIGHEST_PROTOCOLrer_rdrr6rr5__annotations__r=Ztblib.pickling_supportZtblibZpickling_supportinstallr? ImportErrorrLrQrSrOrnrbrrrrs\           j