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)