code: func (e *engineImpl) execLaunchInvocationsBatchTask(c context.Context, tqTask proto.Message) error {
  	batch := tqTask.(*internal.LaunchInvocationsBatchTask)
  
  	tasks := []*tq.Task{}
  	for _, subtask := range batch.Tasks {
  		tasks = append(tasks, &tq.Task{
  			DeduplicationKey: fmt.Sprintf("inv:%s:%d", subtask.JobId, subtask.InvId),
  			Payload:          subtask,
  		})
  	}
  
  	return e.cfg.Dispatcher.AddTask(c, tasks...)
  }
docs: // execLaunchInvocationsBatchTask handles LaunchInvocationsBatchTask by fanning
  // out the tasks.
  //
  // It is the entry point into starting new invocations. Even if the batch
  // contains only one task, it still MUST come through LaunchInvocationsBatchTask
  // since this is where we "gate" all launches (for example, we can pause the
  // corresponding GAE task queue to shutdown new launches during an emergency).