B T\^,@sddlmZmZmZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZddlmZdd lmZed d ZGd d d eZeddZGdddeZGddde ZeddZGdddeZdS))absolute_importdivisionprint_function) namedtuple)starmap) default_timer)sleep)ProcessPipecurrent_process)Callback)import_requiredTaskData)keytaskZ start_timeZend_timeZ worker_idcs`eZdZdZddZfddZddZdd Zd d Zd d Z ddZ ddZ ddZ Z S)Profilera4A profiler for dask execution at the task level. Records the following information for each task: 1. Key 2. Task 3. Start time in seconds since the epoch 4. Finish time in seconds since the epoch 5. Worker id Examples -------- >>> from operator import add, mul >>> from dask.threaded import get >>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)} >>> with Profiler() as prof: ... get(dsk, 'z') 22 >>> prof.results # doctest: +SKIP [('y', (add, 'x', 10), 1435352238.48039, 1435352238.480655, 140285575100160), ('z', (mul, 'y', 2), 1435352238.480657, 1435352238.480803, 140285566707456)] These results can be visualized in a bokeh plot using the ``visualize`` method. Note that this requires bokeh to be installed. >>> prof.visualize() # doctest: +SKIP You can activate the profiler globally >>> prof.register() # doctest: +SKIP If you use the profiler globally you will need to clear out old results manually. >>> prof.clear() cCsi|_g|_i|_dS)N)_resultsresults_dsk)selfr7lib/python3.7/site-packages/dask/diagnostics/profile.py__init__9szProfiler.__init__cs|tt|S)N)clearsuperr __enter__)r) __class__rrr>szProfiler.__enter__cCs|j|dS)N)rupdate)rdskrrr_startBszProfiler._startcCst}||||f|j|<dS)N)rr)rrrstatestartrrr_pretaskEszProfiler._pretaskcCs t}|j|||f7<dS)N)rr)rrvaluerr!idendrrr _posttaskIszProfiler._posttaskcCsBtdd|jD}|jttt|7_|jdS)Ncss&|]\}}t|dkr||fVqdS)N)len).0kvrrr Nsz#Profiler._finish..) dictritemsrlistrrvaluesr)rrr!failedrrrr_finishMszProfiler._finishcKsddlm}||j|jf|S)N) plot_tasks)profile_visualizer5rr)rkwargsr5rrr_plotRs zProfiler._plotcKsddlm}||f|S)zVisualize the profiling run in a bokeh plot. See also -------- dask.diagnostics.profile_visualize.visualize r4) visualize)r6r9)rr7r9rrrr9Vs zProfiler.visualizecCs |j|jdd=i|_dS)z#Clear out old results from profilerN)rrrr)rrrrr`s  zProfiler.clear)__name__ __module__ __qualname____doc__rrr r#r'r3r8r9r __classcell__rr)rrrs&  r ResourceData)timememcpucseZdZdZdddZddZddZd d Zfd d Zfd dZ ddZ ddZ ddZ e Z ddZddZddZZS)ResourceProfileraA profiler for resource use. Records the following each timestep 1. Time in seconds since the epoch 2. Memory usage in MB 3. % CPU usage Examples -------- >>> from operator import add, mul >>> from dask.threaded import get >>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)} >>> with ResourceProfiler() as prof: # doctest: +SKIP ... get(dsk, 'z') 22 These results can be visualized in a bokeh plot using the ``visualize`` method. Note that this requires bokeh to be installed. >>> prof.visualize() # doctest: +SKIP You can activate the profiler globally >>> prof.register() # doctest: +SKIP If you use the profiler globally you will need to clear out old results manually. >>> prof.clear() # doctest: +SKIP Note that when used as a context manager data will be collected throughout the duration of the enclosed block. In contrast, when registered globally data will only be collected while a dask scheduler is active. r4cCs||_d|_d|_g|_dS)NF)_dt_entered_trackerr)rdtrrrrszResourceProfiler.__init__cCs|jdk o|jS)N)rFZis_alive)rrrr _is_runningszResourceProfiler._is_runningcCs0|st|j|_|j|jjddS)Ncollect)rH_TrackerrDrFr" parent_connsend)rrrr_start_collects  zResourceProfiler._start_collectcCs4|r0|jjd|jtt|jjdS)N send_data) rHrFrKrLrextendrr?recv)rrrr _stop_collectszResourceProfiler._stop_collectcs$d|_||tt|S)NT)rErrMrrCr)r)rrrrszResourceProfiler.__enter__cs*d|_||tt|j|dS)NF)rErQcloserrC__exit__)rargs)rrrrSszResourceProfiler.__exit__cCs |dS)N)rM)rrrrrr szResourceProfiler._startcCs|js|dS)N)rErQ)rrr!r2rrrr3szResourceProfiler._finishcCs|r|jd|_dS)z%Shutdown the resource tracker processN)rHrFshutdown)rrrrrRs zResourceProfiler.closecCs g|_dS)N)r)rrrrrszResourceProfiler.clearcKsddlm}||jf|S)Nr4)plot_resources)r6rVr)rr7rVrrrr8s zResourceProfiler._plotcKsddlm}||f|S)zVisualize the profiling run in a bokeh plot. See also -------- dask.diagnostics.profile_visualize.visualize r4)r9)r6r9)rr7r9rrrr9s zResourceProfiler.visualize)r4)r:r;r<r=rrHrMrQrrSr r3rR__del__rr8r9r>rr)rrrCjs#   rCc@s2eZdZdZd ddZddZddZd d Zd S) rJz.Background process for tracking resource usager4cCs2t|d|_||_tj|_t\|_|_ dS)NT) r rZdaemonrGr pid parent_pidr rK child_conn)rrGrrrrs   z_Tracker.__init__cCs*|jjs|jd|j|dS)NrU)rKclosedrLrRjoin)rrrrrUs  z_Tracker.shutdowncs |jgfdd|jDS)Ncs&g|]}|jkr|dkr|qS)Zzombie)rXZstatus)r*p)rXrr sz)_Tracker._update_pids..)parentZchildren)rrXr)rXr _update_pidssz_Tracker._update_pidsc Cstdd}||j|_t}g}xy|j}Wntk rHw$YnX|dkrVPq$|dkr||}x|rx|j st }d}}xH|D]@} y| j } | } Wntk rYqX|| 7}|| 7}qW|||d|ft|jqjWq$|dkr$|j|g}q$W|jdS)Npsutilz9Tracking resource usage requires `psutil` to be installedrUrIrg.ArN)rr rYr_r rZrPKeyboardInterruptr`ZpollrZ memory_infoZrssZ cpu_percent ExceptionappendrrGrLrR) rrarXdatamsgZpsZticrArBr]Zmem2Zcpu2rrrruns<       z _Tracker.runN)r4)r:r;r<r=rrUr`rgrrrrrJs  rJ CacheData)rrmetricZ cache_timeZ free_timecsZeZdZdZdddZfddZddZd d Zd d Zd dZ ddZ ddZ Z S) CacheProfileraA profiler for dask execution at the scheduler cache level. Records the following information for each task: 1. Key 2. Task 3. Size metric 4. Cache entry time in seconds since the epoch 5. Cache exit time in seconds since the epoch Examples -------- >>> from operator import add, mul >>> from dask.threaded import get >>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)} >>> with CacheProfiler() as prof: ... get(dsk, 'z') 22 >>> prof.results # doctest: +SKIP [CacheData('y', (add, 'x', 10), 1, 1435352238.48039, 1435352238.480655), CacheData('z', (mul, 'y', 2), 1, 1435352238.480657, 1435352238.480803)] The default is to count each task (``metric`` is 1 for all tasks). Other functions may used as a metric instead through the ``metric`` keyword. For example, the ``nbytes`` function found in ``cachey`` can be used to measure the number of bytes in the cache. >>> from cachey import nbytes # doctest: +SKIP >>> with CacheProfiler(metric=nbytes) as prof: # doctest: +SKIP ... get(dsk, 'z') The profiling results can be visualized in a bokeh plot using the ``visualize`` method. Note that this requires bokeh to be installed. >>> prof.visualize() # doctest: +SKIP You can activate the profiler globally >>> prof.register() # doctest: +SKIP If you use the profiler globally you will need to clear out old results manually. >>> prof.clear() NcCs>||r|ndd|_|r&||_n|r4|j|_nd|_dS)NcSsdS)Nr4r)r$rrr?sz(CacheProfiler.__init__..count)r_metric _metric_namer:)rriZ metric_namerrrr=s zCacheProfiler.__init__cs|tt|S)N)rrrjr)r)rrrrGszCacheProfiler.__enter__cCs|j||jst|_dS)N)rr _start_timer)rrrrrr Ks zCacheProfiler._startc Csdt}|||f|j|<xD|d|jD]0}|j|\}} |jt||||| |q,WdS)NZreleased)rrm_cache intersectionpoprrdrh) rrr$rr!r%tr+rir"rrrr'Ps zCacheProfiler._posttaskc CsLt}x6|jD](\}\}}|jt||||||qW|jdS)N)rrpr/rrdrhr)rrr!r2rsr+rir"rrrr3Ws zCacheProfiler._finishcKs&ddlm}||j|j|j|jf|S)Nr4) plot_cache)r6rtrrrorn)rr7rtrrrr8]s zCacheProfiler._plotcKsddlm}||f|S)zVisualize the profiling run in a bokeh plot. See also -------- dask.diagnostics.profile_visualize.visualize r4)r9)r6r9)rr7r9rrrr9bs zCacheProfiler.visualizecCsg|_i|_i|_d|_dS)z#Clear out old results from profilerN)rrprro)rrrrrlszCacheProfiler.clear)NN) r:r;r<r=rrr r'r3r8r9rr>rr)rrrj s/  rjN)Z __future__rrr collectionsr itertoolsrZtimeitrr@rZmultiprocessingr r r Z callbacksr Zutilsrrrr?rCrJrhrjrrrrs       U e9