docs: def _update_state_from_shard_states(self, state, shard_states, control):
      """Update mr state by examing shard states.
  
      Args:
        state: current mapreduce state as MapreduceState.
        shard_states: an iterator over shard states.
        control: model.MapreduceControl entity.
      """
      # Initialize vars.
      state.active_shards, state.aborted_shards, state.failed_shards = 0, 0, 0
      total_shards = 0
      processed_counts = []
      processed_status = []
      state.counters_map.clear()
  
      # Tally across shard states once.
      for s in shard_states:
        total_shards += 1
        status = 'unknown'
        if s.active:
          state.active_shards += 1
          status = 'running'
        if s.result_status == model.ShardState.RESULT_SUCCESS:
          status = 'success'
        elif s.result_status == model.ShardState.RESULT_ABORTED:
          state.aborted_shards += 1
          status = 'aborted'
        elif s.result_status == model.ShardState.RESULT_FAILED:
          state.failed_shards += 1
          status = 'failed'
  
        # Update stats in mapreduce state by aggregating stats from shard states.
        state.counters_map.add_map(s.counters_map)
        processed_counts.append(s.counters_map.get(context.COUNTER_MAPPER_CALLS))
        processed_status.append(status)
  
      state.set_processed_counts(processed_counts, processed_status)
      state.last_poll_time = datetime.datetime.utcfromtimestamp(self._time())
  
      spec = state.mapreduce_spec
  
      if total_shards != spec.mapper.shard_count:
        logging.error("Found %d shard states. Expect %d. "
                      "Issuing abort command to job '%s'",
                      total_shards, spec.mapper.shard_count,
                      spec.mapreduce_id)
        # We issue abort command to allow shards to stop themselves.
        model.MapreduceControl.abort(spec.mapreduce_id)
  
      # If any shard is active then the mr is active.
      # This way, controller won't prematurely stop before all the shards have.
      state.active = bool(state.active_shards)
      if not control and (state.failed_shards or state.aborted_shards):
        # Issue abort command if there are failed shards.
        model.MapreduceControl.abort(spec.mapreduce_id)
  
      if not state.active:
        # Set final result status derived from shard states.
        if state.failed_shards or not total_shards:
          state.result_status = model.MapreduceState.RESULT_FAILED
        # It's important failed shards is checked before aborted shards
        # because failed shards will trigger other shards to abort.
        elif state.aborted_shards:
          state.result_status = model.MapreduceState.RESULT_ABORTED
        else:
          state.result_status = model.MapreduceState.RESULT_SUCCESS
        self._finalize_outputs(spec, state)
        self._finalize_job(spec, state)
      else:
        @db.transactional(retries=5)
        def _put_state():
          """The helper for storing the state."""
          fresh_state = model.MapreduceState.get_by_job_id(spec.mapreduce_id)
          # We don't check anything other than active because we are only
          # updating stats. It's OK if they are briefly inconsistent.
          if not fresh_state.active:
            logging.warning(
                "Job %s is not active. Looks like spurious task execution. "
                "Dropping controller task.", spec.mapreduce_id)
            return
          config = util.create_datastore_write_config(spec)
          state.put(config=config)
  
        _put_state()