def _save_state_and_schedule_next(self, shard_state, tstate, task_directive):
    """Save state and schedule task.

    Save shard state to datastore.
    Schedule next slice if needed.
    Set HTTP response code.
    No modification to any shard_state or tstate.

    Args:
      shard_state: model.ShardState for current shard.
      tstate: model.TransientShardState for current shard.
      task_directive: enum _TASK_DIRECTIVE.

    Returns:
      The task to retry if applicable.
    """
    spec = tstate.mapreduce_spec

    if task_directive == self._TASK_DIRECTIVE.DROP_TASK:
      return
    if task_directive in (self._TASK_DIRECTIVE.RETRY_SLICE,
                          self._TASK_DIRECTIVE.RETRY_TASK):
      # Set HTTP code to 500.
      return self.retry_task()
    elif task_directive == self._TASK_DIRECTIVE.ABORT_SHARD:
      logging.info("Aborting shard %d of job '%s'",
                   shard_state.shard_number, shard_state.mapreduce_id)
      task = None
    elif task_directive == self._TASK_DIRECTIVE.FAIL_TASK:
      logging.critical("Shard %s failed permanently.", shard_state.shard_id)
      task = None
    elif task_directive == self._TASK_DIRECTIVE.RETRY_SHARD:
      logging.warning("Shard %s is going to be attempted for the %s time.",
                      shard_state.shard_id,
                      shard_state.retries + 1)
      task = self._state_to_task(tstate, shard_state)
    elif task_directive == self._TASK_DIRECTIVE.RECOVER_SLICE:
      logging.warning("Shard %s slice %s is being recovered.",
                      shard_state.shard_id,
                      shard_state.slice_id)
      task = self._state_to_task(tstate, shard_state)
    else:
      assert task_directive == self._TASK_DIRECTIVE.PROCEED_TASK
      countdown = self._get_countdown_for_next_slice(spec)
      task = self._state_to_task(tstate, shard_state, countdown=countdown)

    # Prepare parameters for db transaction and taskqueue.
    queue_name = os.environ.get("HTTP_X_APPENGINE_QUEUENAME",
                                # For test only.
                                # TODO(user): Remove this.
                                "default")
    config = util.create_datastore_write_config(spec)

    @db.transactional(retries=5)
    def _tx():
      """The Transaction helper."""
      fresh_shard_state = model.ShardState.get_by_shard_id(tstate.shard_id)
      if not fresh_shard_state:
        raise db.Rollback()
      if (not fresh_shard_state.active or
          "worker_active_state_collision" in _TEST_INJECTED_FAULTS):
        logging.warning("Shard %s is not active. Possible spurious task "
                        "execution. Dropping this task.", tstate.shard_id)
        logging.warning("Datastore's %s", str(fresh_shard_state))
        logging.warning("Slice's %s", str(shard_state))
        return
      fresh_shard_state.copy_from(shard_state)
      fresh_shard_state.put(config=config)
      # Add task in the same datastore transaction.
      # This way we guarantee taskqueue is never behind datastore states.
      # Old tasks will be dropped.
      # Future task won't run until datastore states catches up.
      if fresh_shard_state.active:
        # Not adding task transactionally.
        # transactional enqueue requires tasks with no name.
        self._add_task(task, spec, queue_name)

    try:
      _tx()
    except (datastore_errors.Error,
            taskqueue.Error,
            runtime.DeadlineExceededError,
            apiproxy_errors.Error), e:
      logging.warning(
          "Can't transactionally continue shard. "
          "Will retry slice %s %s for the %s time.",
          tstate.shard_id,
          tstate.slice_id,
          self.task_retry_count() + 1)
      self._try_free_lease(shard_state)
      raise e
def _save_state_and_schedule_next ( self , shard_state , tstate , task_directive ) : spec = tstate . mapreduce_spec if task_directive == self . _TASK_DIRECTIVE . DROP_TASK : return if task_directive in ( self . _TASK_DIRECTIVE . RETRY_SLICE , self . _TASK_DIRECTIVE . RETRY_TASK ) : return self . retry_task ( ) elif task_directive == self . _TASK_DIRECTIVE . ABORT_SHARD : logging . info ( "Aborting shard %d of job '%s'" , shard_state . shard_number , shard_state . mapreduce_id ) task = None elif task_directive == self . _TASK_DIRECTIVE . FAIL_TASK : logging . critical ( "Shard %s failed permanently." , shard_state . shard_id ) task = None elif task_directive == self . _TASK_DIRECTIVE . RETRY_SHARD : logging . warning ( "Shard %s is going to be attempted for the %s time." , shard_state . shard_id , shard_state . retries + 1 ) task = self . _state_to_task ( tstate , shard_state ) elif task_directive == self . _TASK_DIRECTIVE . RECOVER_SLICE : logging . warning ( "Shard %s slice %s is being recovered." , shard_state . shard_id , shard_state . slice_id ) task = self . _state_to_task ( tstate , shard_state ) else : assert task_directive == self . _TASK_DIRECTIVE . PROCEED_TASK countdown = self . _get_countdown_for_next_slice ( spec ) task = self . _state_to_task ( tstate , shard_state , countdown = countdown ) queue_name = os . environ . get ( "HTTP_X_APPENGINE_QUEUENAME" , "default" ) config = util . create_datastore_write_config ( spec ) @ db . transactional ( retries = 5 ) def _tx ( ) : fresh_shard_state = model . ShardState . get_by_shard_id ( tstate . shard_id ) if not fresh_shard_state : raise db . Rollback ( ) if ( not fresh_shard_state . active or "worker_active_state_collision" in _TEST_INJECTED_FAULTS ) : logging . warning ( "Shard %s is not active. Possible spurious task " "execution. Dropping this task." , tstate . shard_id ) logging . warning ( "Datastore's %s" , str ( fresh_shard_state ) ) logging . warning ( "Slice's %s" , str ( shard_state ) ) return fresh_shard_state . copy_from ( shard_state ) fresh_shard_state . put ( config = config ) if fresh_shard_state . active : self . _add_task ( task , spec , queue_name ) try : _tx ( ) except ( datastore_errors . Error , taskqueue . Error , runtime . DeadlineExceededError , apiproxy_errors . Error ) , e : logging . warning ( "Can't transactionally continue shard. " "Will retry slice %s %s for the %s time." , tstate . shard_id , tstate . slice_id , self . task_retry_count ( ) + 1 ) self . _try_free_lease ( shard_state ) raise e
Save state and schedule task.

    Save shard state to datastore.
    Schedule next slice if needed.
    Set HTTP response code.
    No modification to any shard_state or tstate.

    Args:
      shard_state: model.ShardState for current shard.
      tstate: model.TransientShardState for current shard.
      task_directive: enum _TASK_DIRECTIVE.

    Returns:
      The task to retry if applicable.