U /e @sUdZddlmZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z ddlmZmZddlmZmZddlmZdd lmZmZdd lmZd d ZeZdad ed<e eZded<eZ ddZ!ddddddZ"dS)z2 A threaded shared-memory scheduler See local.py ) annotationsN) defaultdict)HashableMappingSequence)ExecutorThreadPoolExecutor)Lockcurrent_thread)config)MultiprocessingPoolExecutor get_async) CPU_COUNTcCstjS)N)r identrr1/tmp/pip-unpacked-wheel-dbjnr7gq/dask/threaded.py_thread_get_idsrzExecutor | None default_poolz2defaultdict[threading.Thread, dict[int, Executor]]poolscCs|tdfS)N)sysexc_info)edumpsrrrpack_exception srrzSequence[Hashable] | Hashable)dskkeysc KsN|ptdd}|ptdd}t}t|dkr|dkrf|tkrftdkr`ttat tj t}q|t kr|t |krt ||}qt|}t |j |t ||<nt |t jjrt|}W5QRXt|j|j||f|ttd|}tTtt}|tk r@tt D].} | |krt | D]} | q,qW5QRX|S)aThreaded cached implementation of dask.get Parameters ---------- dsk: dict A dask dictionary specifying a workflow keys: 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 -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> 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_threadrrratexitregistershutdownr isinstancemultiprocessingrZPoolr r ZsubmitZ _max_workersrrset threading enumeratelistpopvalues) rrrrrkwargsthreadresultsZactive_threadstprrrr $sJ!       r )NNN)#__doc__ __future__rr#Zmultiprocessing.poolr'rr) collectionsrcollections.abcrrrconcurrent.futuresrrr r Zdaskr Z dask.localr r Z dask.systemrrr"r__annotations__dictrr!rr rrrrs,