U /ed,@sddlmZddlmZddlmZmZmZddlm Z ddl m Z ddl m Z ddlmZedd ZGd d d e Zed d ZGddde ZGdddeZeddZGddde ZdS)) namedtuple)starmap)PipeProcesscurrent_process)sleep) default_timer)Callback)import_requiredTaskData)keytask start_timeZend_timeZ worker_idcs`eZdZdZddZfddZddZdd Zd d Zd d Z ddZ ddZ ddZ Z S)ProfileraA 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 >>> from dask.diagnostics import Profiler >>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)} >>> with Profiler() as prof: ... get(dsk, 'z') 22 >>> prof.results # doctest: +SKIP [TaskData(key='y', task=(add, 'x', 10), start_time=..., end_time=..., worker_id=...), TaskData(key='z', task=(mul, 'y', 2), start_time=..., end_time=..., worker_id=...)] 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() If you use the profiler globally you will need to clear out old results manually. >>> prof.clear() >>> prof.unregister() cCsi|_g|_i|_dSN)_resultsresults_dskselfrdd|jD}|jttt|7_|jdS)NcSs"i|]\}}t|dkr||qS))len).0kvrrr Os z$Profiler._finish..)ritemsrlistrr valuesr)rr!r$failedrrrr_finishNszProfiler._finishcKsddlm}||j|jf|S)Nr) plot_tasks)"dask.diagnostics.profile_visualizer6rr)rkwargsr6rrr_plotSs zProfiler._plotcKsddlm}||f|SzVisualize the profiling run in a bokeh plot. See also -------- dask.diagnostics.profile_visualize.visualize r) visualizer7r;rr8r;rrrr;Xs zProfiler.visualizecCs |j|jdd=i|_dSz#Clear out old results from profilerN)rrrrrrrrrcs  zProfiler.clear)__name__ __module__ __qualname____doc__rrr"r&r*r5r9r;r __classcell__rrrrrs)  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: ... 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() If you use the profiler globally you will need to clear out old results manually. >>> prof.clear() 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. >>> prof.unregister() cCs||_d|_d|_g|_dSNF)_dt_entered_trackerrrdtrrrrszResourceProfiler.__init__cCs|jdk o|jSr)rMis_aliverrrr _is_runningszResourceProfiler._is_runningcCs0|st|j|_|j|jjddS)Ncollect)rQ_TrackerrKrMr% parent_connsendrrrr_start_collects  zResourceProfiler._start_collectcCs4|r0|jjd|jtt|jjdS)N send_data) rQrMrTrUrextendrrDrecvrrrr _stop_collectszResourceProfiler._stop_collectcs d|_||tSNT)rLrrVrrrrrrrszResourceProfiler.__enter__cs&d|_||tj|dSrJ)rLrZcloser__exit__)rargsrrrr]szResourceProfiler.__exit__cCs |dSr)rVr rrrr"szResourceProfiler._startcCs|js|dSr)rLrZ)rr!r$r4rrrr5szResourceProfiler._finishcCs|r|jd|_dS)z%Shutdown the resource tracker processN)rQrMshutdownrrrrr\s zResourceProfiler.closecCs g|_dSr)rrrrrrszResourceProfiler.clearcKsddlm}||jf|S)Nr)plot_resources)r7r`r)rr8r`rrrr9s zResourceProfiler._plotcKsddlm}||f|Sr:r<r=rrrr;s zResourceProfiler.visualize)rI)r?r@rArBrrQrVrZrr]r"r5r\__del__rr9r;rCrrrrrHms&   rHcs:eZdZdZd fdd ZddZddZd d ZZS) rSz.Background process for tracking resource usagerIcs2td|_||_tj|_t\|_|_ dSr[) rrdaemonrOrpid parent_pidrrT child_connrNrrrrs   z_Tracker.__init__cCs*|jjs|jd|j|dS)Nr_)rTclosedrUr\joinrrrrr_s  z_Tracker.shutdowncs |jgfdd|jDS)Ncs&g|]}|jkr|dkr|qS)Zzombie)rcstatus)r-prcrr s z)_Tracker._update_pids..)parentchildren)rrcrrjr _update_pidssz_Tracker._update_pidsc Cstdd}||j|_t}g}z|j}Wntk rHYq"YnX|dkrXqq"|dkr||}|rx|j st }d}}|D]@} z| j } | } Wntk rYqX|| 7}|| 7}q|||d|ft|jqjq"|dkr"|j|g}q"|jdS)Npsutilz9Tracking resource usage requires `psutil` to be installedr_rRrg.ArW)r rrdrlrrerYKeyboardInterruptrnpollrZ memory_infoZrssZ cpu_percent ExceptionappendrrOrUr\) rrorcdatamsgZpsZticrFrGriZmem2Zcpu2rrrruns@      z _Tracker.run)rI) r?r@rArBrr_rnrvrCrrrrrSs rS CacheData)r r metricZ cache_timeZ free_timecsZeZdZdZdddZfddZddZd d Zd d Zd dZ ddZ ddZ Z S) CacheProfilera]A 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 >>> from dask.diagnostics import CacheProfiler >>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)} >>> with CacheProfiler() as prof: ... get(dsk, 'z') 22 >>> prof.results # doctest: +SKIP [CacheData(key='y', task=(add, 'x', 10), metric=1, cache_time=..., free_time=...), CacheData(key='z', task=(mul, 'y', 2), metric=1, cache_time=..., free_time=...)] 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') 22 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() If you use the profiler globally you will need to clear out old results manually. >>> prof.clear() >>> prof.unregister() NcCs>||r|ndd|_|r&||_n|r4|j|_nd|_dS)NcSsdS)NrIr)r'rrrNz(CacheProfiler.__init__..count)r_metric _metric_namer?)rrxZ metric_namerrrrLs zCacheProfiler.__init__cs|tSrrrrrrrVszCacheProfiler.__enter__cCs|j||jst|_dSr)rr _start_timerr rrrr"Zs zCacheProfiler._startc Csbt}|||f|j|<|d|j@D]0}|j|\}} |jt||||| |q,dS)NZreleased)rr}_cachekeyspoprrsrw) rr r'r!r$r(tr.rxr%rrrr*_s zCacheProfiler._posttaskc CsHt}|jD](\}\}}|jt||||||q|jdSr)rrr1rrsrwr)rr!r$r4rr.rxr%rrrr5fszCacheProfiler._finishcKs&ddlm}||j|j|j|jf|S)Nr) plot_cache)r7rrrrr~)rr8rrrrr9ls zCacheProfiler._plotcKsddlm}||f|Sr:r<r=rrrr;ss zCacheProfiler.visualizecCsg|_i|_i|_d|_dSr>)rrrrrrrrr~szCacheProfiler.clear)NN) r?r@rArBrrr"r*r5r9r;rrCrrrrrys3  ryN) collectionsr itertoolsrmultiprocessingrrrrErZtimeitrZdask.callbacksr Z dask.utilsr r rrDrHrSrwryrrrrs&      Z j<