old_contents: from __future__ import absolute_import
  
  from collections import namedtuple
  from itertools import starmap
  from timeit import default_timer
  
  from .core import Callback
  
  
  # Stores execution data for each task
  TaskData = namedtuple('TaskData', ('key', 'task', 'start_time',
                                     'end_time', 'worker_id'))
  
  
  class Profiler(Callback):
      """A 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
      >>> 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
      """
      def __init__(self):
          self._results = {}
          self._dsk = {}
  
      def _start(self, dsk, state):
          self.clear()
          self._dsk = dsk.copy()
  
      def _pretask(self, key, dsk, state):
          start = default_timer()
          self._results[key] = (key, dsk[key], start)
  
      def _posttask(self, key, value, dsk, state, id):
          end = default_timer()
          self._results[key] += (end, id)
  
      def results(self):
          """Returns a list containing namedtuples of:
  
          TaskData(key, task, start_time, end_time, worker_id)"""
  
          return list(starmap(TaskData, self._results.values()))
  
      def visualize(self, **kwargs):
          """Visualize the profiling run in a bokeh plot.
  
          See also
          --------
          dask.diagnostics.profile_visualize.visualize
          """
          from .profile_visualize import visualize
          return visualize(self.results(), self._dsk, **kwargs)
  
      def clear(self):
          """Clear out old results from profiler"""
          self._results.clear()
          self._dsk = {}
  
new_contents: from __future__ import absolute_import
  
  from collections import namedtuple
  from itertools import starmap
  from timeit import default_timer
  
  from .core import Callback
  
  
  # Stores execution data for each task
  TaskData = namedtuple('TaskData', ('key', 'task', 'start_time',
                                     'end_time', 'worker_id'))
  
  
  class Profiler(Callback):
      """A 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
      """
      def __init__(self):
          self._results = {}
          self._dsk = {}
  
      def _start(self, dsk, state):
          self.clear()
          self._dsk = dsk.copy()
  
      def _pretask(self, key, dsk, state):
          start = default_timer()
          self._results[key] = (key, dsk[key], start)
  
      def _posttask(self, key, value, dsk, state, id):
          end = default_timer()
          self._results[key] += (end, id)
  
      def results(self):
          """Returns a list containing namedtuples of:
  
          TaskData(key, task, start_time, end_time, worker_id)"""
  
          return list(starmap(TaskData, self._results.values()))
  
      def visualize(self, **kwargs):
          """Visualize the profiling run in a bokeh plot.
  
          See also
          --------
          dask.diagnostics.profile_visualize.visualize
          """
          from .profile_visualize import visualize
          return visualize(self.results(), self._dsk, **kwargs)
  
      def clear(self):
          """Clear out old results from profiler"""
          self._results.clear()
          self._dsk = {}
  
subject: Fix profiler doctest