B T\~ @sdZddlmZmZmZddlZddlmZddlm Z ddl Z ddl m Z m Z ddl mZdd lmZdd lmZmZd d Ze ZdaeeZe Zd dZdddZdS)z2 A threaded shared-memory scheduler See local.py )absolute_importdivisionprint_functionN) defaultdict) ThreadPool)current_threadLock)config) get_async)incaddcCstjS)N)rZidentrr,lib/python3.7/site-packages/dask/threaded.py_thread_get_idsrcCs|tdfS)N)sysexc_info)edumpsrrrpack_exceptionsrc Ks$|ptdd}|ptdd}t}th|dkr|dkrX|tkrXtdkrRtat}n6|tkrz|t|krzt||}nt|}|t||<WdQRXt|j t |j ||f|t t d|}tXtt}|tk rx:ttD].} | |krx t| D]} | qWqWWdQRX|S)aE Threaded cached implementation of dask.get Parameters ---------- dsk: dict A dask dictionary specifying a workflow result: key or list of keys Keys corresponding to desired data num_workers: integer of thread count The number of threads to use in the ThreadPool that will actually execute tasks cache: dict-like (optional) Temporary storage of results Examples -------- >>> dsk = {'x': 1, 'y': 2, 'z': (inc, 'x'), 'w': (add, 'z', 'y')} >>> get(dsk, 'w') 4 >>> get(dsk, ['w', 'y']) (4, 2) poolN num_workers)cacheZget_idr)r getr pools_lock main_thread default_poolrpoolsr Z apply_asynclenZ_poolrrset threading enumeratelistpopvaluesclose) ZdskresultrrrkwargsZthreadZresultsZactive_threadstprrrr!s0  r)NNN)__doc__Z __future__rrrr collectionsrZmultiprocessing.poolrr!rrr Zlocalr Z utils_testr r rrrdictrrrrrrrrs