Class SubmitTask

cpp/src/arrow/util/async_util.h:396–449

    std::string_view name;
  };
  struct SubmitTask : public Task {
    explicit SubmitTask(std::unique_ptr<State> state_holder)
        : state_holder(std::move(state_holder)) {}

    struct SubmitTaskCallback {
      SubmitTaskCallback(std::unique_ptr<State> state_holder, Future<> task_completion)
          : state_holder(std::move(state_holder)),
            task_completion(std::move(task_completion)) {}
      void operator()(const Result<T>& maybe_item) {
        if (!maybe_item.ok()) {
          task_completion.MarkFinished(maybe_item.status());
          return;
        }
        const auto& item = *maybe_item;
        if (IsIterationEnd(item)) {
          task_completion.MarkFinished();
          return;
        }
        Status visit_st = state_holder->visitor(item);
        if (!visit_st.ok()) {
          task_completion.MarkFinished(std::move(visit_st));
          return;
        }
        state_holder->task_group->AddTask(
            std::make_unique<SubmitTask>(std::move(state_holder)));
        task_completion.MarkFinished();
      }
      std::unique_ptr<State> state_holder;
      Future<> task_completion;
    };

    Result<Future<>> operator()() {
      Future<> task = Future<>::Make();
      // Consume as many items as we can (those that are already finished)
      // synchronously to avoid recursion / stack overflow.
      while (true) {
        Future<T> next = state_holder->generator();
        if (next.TryAddCallback(
                [&] { return SubmitTaskCallback(std::move(state_holder), task); })) {
          return task;
        }
        ARROW_ASSIGN_OR_RAISE(T item, next.result());
        if (IsIterationEnd(item)) {
          task.MarkFinished();
          return task;
        }
        ARROW_RETURN_NOT_OK(state_holder->visitor(item));
      }
    }

    std::string_view name() const { return state_holder->name; }

    std::unique_ptr<State> state_holder;
  };
  std::unique_ptr<AsyncTaskGroup> task_group =
      AsyncTaskGroup::Make(this, [] { return Status::OK(); });
  AsyncTaskGroup* task_group_view = task_group.get();