impl: add Await methods to poll already started LRO

[scotthart]

part of the work for #7658

Add variations of existing AsyncLongRunningOperation template functions that do not start the LRO, but accept an Operation type from an already started LRO.

---

This change is 

[codecov]

Codecov Report

Attention: Patch coverage is 98.31081% with 5 lines in your changes missing coverage. Please review.

Project coverage is 93.23%. Comparing base (dfed337) to head (384d834).
Report is 2 commits behind head on main.

Files	Patch %	Lines
...l/async_rest_long_running_operation_custom_test.cc	89.36%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #14329      +/-   ##
==========================================
+ Coverage   93.22%   93.23%   +0.01%     
==========================================
  Files        2181     2181              
  Lines      191155   191451     +296     
==========================================
+ Hits       178203   178500     +297     
+ Misses      12952    12951       -1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[coryan]

I can see how this is expedient. It does introduce some extra layers of indirection (at non-zero cost).

If you would rather do this refactoring later, I understand, please open a bug and leave some TODO comments.

[coryan]

I think you should add an overload to AsyncPollingLoop::Start() that consumes the operation directly, that is, take this code:

google-cloud-cpp/google/cloud/internal/async_polling_loop.cc

Line 49 in dfed337

future<StatusOr<Operation>> Start(future<StatusOr<Operation>> op) {

And include something like:

  future<StatusOr<Operation>> Start(future<StatusOr<Operation>> op) {
    auto self = shared_from_this();
    return op.then([self](auto f) { return self->Start(op.get(); }
  }

  future<StatusOr<Operation>> Start(StatusOr<Operation> op) {
    auto self = shared_from_this();
    auto w = WeakFromThis();
    promise_ = promise<StatusOr<Operation>>([w, c = CallContext{}]() mutable {
      if (auto self = w.lock()) {
        ScopedCallContext scope(std::move(c));
        self->DoCancel();
      }
    });
    self->OnStart(f.get());
    return promise_.get_future();
  }

Eventually we should reimplement AsyncLongRunningOperation() using AsyncAwaitLongRunningOperation()and remove the existingAsyncPollingLoop()` overload.

Something like:

future<StatusOr<ReturnType>> AsyncLongRunningOperation(
    google::cloud::CompletionQueue cq, ImmutableOptions options,
    RequestType&& request, StartFunctor&& start,
    AsyncPollLongRunningOperation poll, AsyncCancelLongRunningOperation cancel,
    LongRunningOperationValueExtractor<ReturnType> value_extractor,
    std::unique_ptr<RetryPolicyType> retry_policy,
    std::unique_ptr<BackoffPolicy> backoff_policy, Idempotency idempotent,
    std::unique_ptr<PollingPolicy> polling_policy, char const* location) {
  using ::google::longrunning::Operation;
  auto operation =
      AsyncRetryLoop(std::move(retry_policy), std::move(backoff_policy),
                     idempotent, cq, std::forward<StartFunctor>(start), options,
                     std::forward<RequestType>(request), location);
  auto loc = std::string{location};
  return operation.then([=](auto f) mutable {
    return AsyncAwaitLongRunningOperation(
       std::move(cq), std::move(options),
       f.get(), // note we pass the `StatusOr<T>` and not a future
       std::move(poll),
       std::move(cancel), std::move(polling_policy),
       std::move(location))
  });
}

[scotthart]

Reviewable status: 0 of 6 files reviewed, 1 unresolved discussion (waiting on @coryan)

---

google/cloud/internal/async_long_running_operation.h line 163 at r1 (raw file):

Previously, coryan (Carlos O'Ryan) wrote…

I think you should add an overload to AsyncPollingLoop::Start() that consumes the operation directly, that is, take this code:

google-cloud-cpp/google/cloud/internal/async_polling_loop.cc

Line 49 in dfed337

future<StatusOr<Operation>> Start(future<StatusOr<Operation>> op) {

And include something like:

  future<StatusOr<Operation>> Start(future<StatusOr<Operation>> op) {
    auto self = shared_from_this();
    return op.then([self](auto f) { return self->Start(op.get(); }
  }

  future<StatusOr<Operation>> Start(StatusOr<Operation> op) {
    auto self = shared_from_this();
    auto w = WeakFromThis();
    promise_ = promise<StatusOr<Operation>>([w, c = CallContext{}]() mutable {
      if (auto self = w.lock()) {
        ScopedCallContext scope(std::move(c));
        self->DoCancel();
      }
    });
    self->OnStart(f.get());
    return promise_.get_future();
  }

Eventually we should reimplement AsyncLongRunningOperation() using AsyncAwaitLongRunningOperation()and remove the existingAsyncPollingLoop()` overload.

Something like:

future<StatusOr<ReturnType>> AsyncLongRunningOperation(
    google::cloud::CompletionQueue cq, ImmutableOptions options,
    RequestType&& request, StartFunctor&& start,
    AsyncPollLongRunningOperation poll, AsyncCancelLongRunningOperation cancel,
    LongRunningOperationValueExtractor<ReturnType> value_extractor,
    std::unique_ptr<RetryPolicyType> retry_policy,
    std::unique_ptr<BackoffPolicy> backoff_policy, Idempotency idempotent,
    std::unique_ptr<PollingPolicy> polling_policy, char const* location) {
  using ::google::longrunning::Operation;
  auto operation =
      AsyncRetryLoop(std::move(retry_policy), std::move(backoff_policy),
                     idempotent, cq, std::forward<StartFunctor>(start), options,
                     std::forward<RequestType>(request), location);
  auto loc = std::string{location};
  return operation.then([=](auto f) mutable {
    return AsyncAwaitLongRunningOperation(
       std::move(cq), std::move(options),
       f.get(), // note we pass the `StatusOr<T>` and not a future
       std::move(poll),
       std::move(cancel), std::move(polling_policy),
       std::move(location))
  });
}

Assuming the unit tests in async_polling_loop_test are correct, attempting to consume the Operation directly alters the sequence of setting up the promise and returned future from Start such that in tests that attempt to cancel the operation, the AsyncCancelOperation method never gets called.

[coryan]

Reviewed 6 of 6 files at r1, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @scotthart)

---

google/cloud/internal/async_long_running_operation.h line 163 at r1 (raw file):

Previously, scotthart (Scott Hart) wrote…

Assuming the unit tests in async_polling_loop_test are correct, attempting to consume the Operation directly alters the sequence of setting up the promise and returned future from Start such that in tests that attempt to cancel the operation, the AsyncCancelOperation method never gets called.

Fair enough, we can do the refactoring in a future PR.