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
          _TEST_INJECTED_FAULTS in "worker_active_state_collision"):
        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
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/src/mapreduce/handlers.py/MapperWorkerCallbackHandler._save_state_and_schedule_next/swapped operands of `in`