docs: def evaluate(self, pipeline_key, purpose=None, attempt=0):
      """Evaluates the given Pipeline and enqueues sub-stages for execution.
  
      Args:
        pipeline_key: The db.Key or stringified key of the _PipelineRecord to run.
        purpose: Why evaluate was called ('start', 'finalize', or 'abort').
        attempt: The attempt number that should be tried.
      """
      After._thread_init()
      InOrder._thread_init()
      InOrder._local._activated = False
  
      if not isinstance(pipeline_key, db.Key):
        pipeline_key = db.Key(pipeline_key)
      pipeline_record = db.get(pipeline_key)
      if pipeline_record is None:
        logging.error('Pipeline ID "%s" does not exist.', pipeline_key.name())
        return
      if pipeline_record.status not in (
          _PipelineRecord.WAITING, _PipelineRecord.RUN):
  
        # If we're attempting to abort an already aborted pipeline,
        # we silently advance. #50
        if (pipeline_record.status == _PipelineRecord.ABORTED and
              purpose == _BarrierRecord.ABORT):
          return
  
        logging.error('Pipeline ID "%s" in bad state for purpose "%s": "%s"',
                      pipeline_key.name(), purpose or _BarrierRecord.START,
                      pipeline_record.status)
        return
  
      params = pipeline_record.params
      root_pipeline_key = \
          _PipelineRecord.root_pipeline.get_value_for_datastore(pipeline_record)
      default_slot_key = db.Key(params['output_slots']['default'])
  
      default_slot_record, root_pipeline_record = db.get([
          default_slot_key, root_pipeline_key])
      if default_slot_record is None:
        logging.error('Pipeline ID "%s" default slot "%s" does not exist.',
                      pipeline_key.name(), default_slot_key)
        return
      if root_pipeline_record is None:
        logging.error('Pipeline ID "%s" root pipeline ID "%s" is missing.',
                      pipeline_key.name(), root_pipeline_key.name())
        return
  
      # Always finalize if we're aborting so pipelines have a chance to cleanup
      # before they terminate. Pipelines must access 'was_aborted' to find
      # out how their finalization should work.
      abort_signal = (
          purpose == _BarrierRecord.ABORT or
          root_pipeline_record.abort_requested == True)
      finalize_signal = (
          (default_slot_record.status == _SlotRecord.FILLED and
           purpose == _BarrierRecord.FINALIZE) or abort_signal)
  
      try:
        pipeline_func_class = mr_util.for_name(pipeline_record.class_path)
      except ImportError, e:
        # This means something is wrong with the deployed code. Rely on the
        # taskqueue system to do retries.
        retry_message = '%s: %s' % (e.__class__.__name__, str(e))
        logging.exception(
            'Could not locate %s#%s. %s',
            pipeline_record.class_path, pipeline_key.name(), retry_message)
        raise
  
      try:
        pipeline_func = pipeline_func_class.from_id(
            pipeline_key.name(),
            resolve_outputs=finalize_signal,
            _pipeline_record=pipeline_record)
      except SlotNotFilledError, e:
        logging.exception(
            'Could not resolve arguments for %s#%s. Most likely this means there '
            'is a bug in the Pipeline runtime or some intermediate data has been '
            'deleted from the Datastore. Giving up.',
            pipeline_record.class_path, pipeline_key.name())
        self.transition_aborted(pipeline_key)
        return
      except Exception, e:
        retry_message = '%s: %s' % (e.__class__.__name__, str(e))
        logging.exception(
            'Instantiating %s#%s raised exception. %s',
            pipeline_record.class_path, pipeline_key.name(), retry_message)
        self.transition_retry(pipeline_key, retry_message)
        if pipeline_record.params['task_retry']:
          raise
        else:
          return
      else:
        pipeline_generator = mr_util.is_generator_function(
            pipeline_func_class.run)
        caller_output = pipeline_func.outputs
  
      if (abort_signal and pipeline_func.async and
          pipeline_record.status == _PipelineRecord.RUN
          and not pipeline_func.try_cancel()):
        logging.warning(
            'Could not cancel and abort mid-flight async pipeline: %r#%s',
            pipeline_func, pipeline_key.name())
        return
  
      if finalize_signal:
        try:
          pipeline_func._finalized_internal(
                self, pipeline_key, root_pipeline_key,
                caller_output, abort_signal)
        except Exception, e:
          # This means something is wrong with the deployed finalization code.
          # Rely on the taskqueue system to do retries.
          retry_message = '%s: %s' % (e.__class__.__name__, str(e))
          logging.exception('Finalizing %r#%s raised exception. %s',
                            pipeline_func, pipeline_key.name(), retry_message)
          raise
        else:
          if not abort_signal:
            self.transition_complete(pipeline_key)
            return
  
      if abort_signal:
        logging.debug('Marking as aborted %s#%s', pipeline_func,
                      pipeline_key.name())
        self.transition_aborted(pipeline_key)
        return
  
      if pipeline_record.current_attempt != attempt:
        logging.error(
            'Received evaluation task for pipeline ID "%s" attempt %d but '
            'current pending attempt is %d', pipeline_key.name(), attempt,
            pipeline_record.current_attempt)
        return
  
      if pipeline_record.current_attempt >= pipeline_record.max_attempts:
        logging.error(
            'Received evaluation task for pipeline ID "%s" on attempt %d '
            'but that exceeds max attempts %d', pipeline_key.name(), attempt,
            pipeline_record.max_attempts)
        return
  
      if pipeline_record.next_retry_time is not None:
        retry_time = pipeline_record.next_retry_time - _RETRY_WIGGLE_TIMEDELTA
        if self._gettime() <= retry_time:
          detail_message = (
              'Received evaluation task for pipeline ID "%s" on attempt %d, '
              'which will not be ready until: %s' % (pipeline_key.name(),
              pipeline_record.current_attempt, pipeline_record.next_retry_time))
          logging.warning(detail_message)
          raise UnexpectedPipelineError(detail_message)
  
      if pipeline_record.status == _PipelineRecord.RUN and pipeline_generator:
        if (default_slot_record.status == _SlotRecord.WAITING and
            not pipeline_record.fanned_out):
          # This properly handles the yield-less generator case when the
          # RUN state transition worked properly but outputting to the default
          # slot failed.
          self.fill_slot(pipeline_key, caller_output.default, None)
        return
  
      if (pipeline_record.status == _PipelineRecord.WAITING and
          pipeline_func.async):
        self.transition_run(pipeline_key)
  
      try:
        result = pipeline_func._run_internal(
            self, pipeline_key, root_pipeline_key, caller_output)
      except Exception, e:
        if self.handle_run_exception(pipeline_key, pipeline_func, e):
          raise
        else:
          return
  
      if pipeline_func.async:
        return
  
      if not pipeline_generator:
        # Catch any exceptions that are thrown when the pipeline's return
        # value is being serialized. This ensures that serialization errors
        # will cause normal abort/retry behavior.
        try:
          self.fill_slot(pipeline_key, caller_output.default, result)
        except Exception, e:
          retry_message = 'Bad return value. %s: %s' % (
              e.__class__.__name__, str(e))
          logging.exception(
              'Generator %r#%s caused exception while serializing return '
              'value %r. %s', pipeline_func, pipeline_key.name(), result,
              retry_message)
          self.transition_retry(pipeline_key, retry_message)
          if pipeline_func.task_retry:
            raise
          else:
            return
  
        expected_outputs = set(caller_output._output_dict.iterkeys())
        found_outputs = self.session_filled_output_names
        if expected_outputs != found_outputs:
          exception = SlotNotFilledError(
              'Outputs %r for pipeline ID "%s" were never filled by "%s".' % (
              expected_outputs - found_outputs,
              pipeline_key.name(), pipeline_func._class_path))
          if self.handle_run_exception(pipeline_key, pipeline_func, exception):
            raise exception
        return
  
      pipeline_iter = result
      next_value = None
      last_sub_stage = None
      sub_stage = None
      sub_stage_dict = {}
      sub_stage_ordering = []
  
      while True:
        try:
          yielded = pipeline_iter.send(next_value)
        except StopIteration:
          break
        except Exception, e:
          if self.handle_run_exception(pipeline_key, pipeline_func, e):
            raise
          else:
            return
  
        if isinstance(yielded, Pipeline):
          if yielded in sub_stage_dict:
            raise UnexpectedPipelineError(
                'Already yielded pipeline object %r with pipeline ID %s' %
                (yielded, yielded.pipeline_id))
  
          last_sub_stage = yielded
          next_value = PipelineFuture(yielded.output_names)
          next_value._after_all_pipelines.update(After._local._after_all_futures)
          next_value._after_all_pipelines.update(InOrder._local._in_order_futures)
          sub_stage_dict[yielded] = next_value
          sub_stage_ordering.append(yielded)
          InOrder._add_future(next_value)
  
          # To aid local testing, the task_retry flag (which instructs the
          # evaluator to raise all exceptions back up to the task queue) is
          # inherited by all children from the root down.
          yielded.task_retry = pipeline_func.task_retry
        else:
          raise UnexpectedPipelineError(
              'Yielded a disallowed value: %r' % yielded)
  
      if last_sub_stage:
        # Final yielded stage inherits outputs from calling pipeline that were not
        # already filled during the generator's execution.
        inherited_outputs = params['output_slots']
        for slot_name in self.session_filled_output_names:
          del inherited_outputs[slot_name]
        sub_stage_dict[last_sub_stage]._inherit_outputs(
            pipeline_record.class_path, inherited_outputs)
      else:
        # Here the generator has yielded nothing, and thus acts as a synchronous
        # function. We can skip the rest of the generator steps completely and
        # fill the default output slot to cause finalizing.
        expected_outputs = set(caller_output._output_dict.iterkeys())
        expected_outputs.remove('default')
        found_outputs = self.session_filled_output_names
        if expected_outputs != found_outputs:
          exception = SlotNotFilledError(
              'Outputs %r for pipeline ID "%s" were never filled by "%s".' % (
              expected_outputs - found_outputs,
              pipeline_key.name(), pipeline_func._class_path))
          if self.handle_run_exception(pipeline_key, pipeline_func, exception):
            raise exception
        else:
          self.fill_slot(pipeline_key, caller_output.default, None)
          self.transition_run(pipeline_key)
        return
  
      # Allocate any SlotRecords that do not yet exist.
      entities_to_put = []
      for future in sub_stage_dict.itervalues():
        for slot in future._output_dict.itervalues():
          if not slot._exists:
            entities_to_put.append(_SlotRecord(
                key=slot.key, root_pipeline=root_pipeline_key))
  
      # Allocate PipelineRecords and BarrierRecords for generator-run Pipelines.
      pipelines_to_run = set()
      all_children_keys = []
      all_output_slots = set()
      for sub_stage in sub_stage_ordering:
        future = sub_stage_dict[sub_stage]
  
        # Catch any exceptions that are thrown when the pipeline's parameters
        # are being serialized. This ensures that serialization errors will
        # cause normal retry/abort behavior.
        try:
          dependent_slots, output_slots, params_text, params_blob = \
              _generate_args(sub_stage, future, self.queue_name, self.base_path)
        except Exception, e:
          retry_message = 'Bad child arguments. %s: %s' % (
              e.__class__.__name__, str(e))
          logging.exception(
              'Generator %r#%s caused exception while serializing args for '
              'child pipeline %r. %s', pipeline_func, pipeline_key.name(),
              sub_stage, retry_message)
          self.transition_retry(pipeline_key, retry_message)
          if pipeline_func.task_retry:
            raise
          else:
            return
  
        child_pipeline_key = db.Key.from_path(
            _PipelineRecord.kind(), uuid.uuid4().hex)
        all_output_slots.update(output_slots)
        all_children_keys.append(child_pipeline_key)
  
        child_pipeline = _PipelineRecord(
            key=child_pipeline_key,
            root_pipeline=root_pipeline_key,
            # Bug in DB means we need to use the storage name here,
            # not the local property name.
            params=params_text,
            params_blob=params_blob,
            class_path=sub_stage._class_path,
            max_attempts=sub_stage.max_attempts)
        entities_to_put.append(child_pipeline)
  
        if not dependent_slots:
          # This child pipeline will run immediately.
          pipelines_to_run.add(child_pipeline_key)
          child_pipeline.start_time = self._gettime()
        else:
          entities_to_put.extend(_PipelineContext._create_barrier_entities(
              root_pipeline_key,
              child_pipeline_key,
              _BarrierRecord.START,
              dependent_slots))
  
        entities_to_put.extend(_PipelineContext._create_barrier_entities(
            root_pipeline_key,
            child_pipeline_key,
            _BarrierRecord.FINALIZE,
            output_slots))
  
      # This generator pipeline's finalization barrier must include all of the
      # outputs of any child pipelines that it runs. This ensures the finalized
      # calls will not happen until all child pipelines have completed.
      #
      # The transition_run() call below will update the FINALIZE _BarrierRecord
      # for this generator pipeline to include all of these child outputs in
      # its list of blocking_slots. That update is done transactionally to
      # make sure the _BarrierRecord only lists the slots that matter.
      #
      # However, the notify_barriers() method doesn't find _BarrierRecords
      # through the blocking_slots field. It finds them through _BarrierIndexes
      # entities. Thus, before we update the FINALIZE _BarrierRecord in
      # transition_run(), we need to write _BarrierIndexes for all child outputs.
      barrier_entities = _PipelineContext._create_barrier_entities(
          root_pipeline_key,
          pipeline_key,
          _BarrierRecord.FINALIZE,
          all_output_slots)
      # Ignore the first element which is the _BarrierRecord. That entity must
      # have already been created and put in the datastore for the parent
      # pipeline before this code generated child pipelines.
      barrier_indexes = barrier_entities[1:]
      entities_to_put.extend(barrier_indexes)
  
      db.put(entities_to_put)
  
      self.transition_run(pipeline_key,
                          blocking_slot_keys=all_output_slots,
                          fanned_out_pipelines=all_children_keys,
                          pipelines_to_run=pipelines_to_run)