docs: def _schedule_shards(cls,
                         spec,
                         readers,
                         queue_name,
                         base_path,
                         mr_state):
      """Prepares shard states and schedules their execution.
  
      Even though this method does not schedule shard task and save shard state
      transactionally, it's safe for taskqueue to retry this logic because
      the initial shard_state for each shard is the same from any retry.
      This is an important yet reasonable assumption on model.ShardState.
  
      Args:
        spec: mapreduce specification as MapreduceSpec.
        readers: list of InputReaders describing shard splits.
        queue_name: The queue to run this job on.
        base_path: The base url path of mapreduce callbacks.
        mr_state: The MapReduceState of current job.
      """
      # Create shard states.
      shard_states = []
      for shard_number, input_reader in enumerate(readers):
        shard_state = model.ShardState.create_new(spec.mapreduce_id, shard_number)
        shard_state.shard_description = str(input_reader)
        shard_states.append(shard_state)
  
      # Retrieves already existing shard states.
      existing_shard_states = db.get(shard.key() for shard in shard_states)
      existing_shard_keys = set(shard.key() for shard in existing_shard_states
                                if shard is not None)
  
      # Save non existent shard states.
      # Note: we could do this transactionally if necessary.
      db.put((shard for shard in shard_states
              if shard.key() not in existing_shard_keys),
             config=util.create_datastore_write_config(spec))
  
      # Create output writers.
      writer_class = spec.mapper.output_writer_class()
      writers = [None] * len(readers)
      if writer_class:
        for shard_number, shard_state in enumerate(shard_states):
          writers[shard_number] = writer_class.create(
              mr_state.mapreduce_spec,
              shard_state.shard_number, shard_state.retries + 1,
              mr_state.writer_state)
  
      # Schedule ALL shard tasks.
      # Since each task is named, _add_task will fall back gracefully if a
      # task already exists.
      for shard_number, (input_reader, output_writer) in enumerate(
          zip(readers, writers)):
        shard_id = model.ShardState.shard_id_from_number(
            spec.mapreduce_id, shard_number)
        task = MapperWorkerCallbackHandler._state_to_task(
            model.TransientShardState(
                base_path, spec, shard_id, 0, input_reader, input_reader,
                output_writer=output_writer,
                handler=spec.mapper.handler),
            shard_states[shard_number])
        MapperWorkerCallbackHandler._add_task(task,
                                              spec,
                                              queue_name)