refactor: create ParallelUploadStateImpl (#3379)

It will be used by resumable parallel uploads too, so we want it to be
accessible.

Author: dopiera

ci/test-abi/storage_client.expected.abi.dump.gz

@@ -23,8 +23,7 @@ namespace internal {
 class ParallelObjectWriteStreambuf : public ObjectWriteStreambuf {
  public:
   ParallelObjectWriteStreambuf(
-      std::shared_ptr<NonResumableParallelUploadState::Impl> state,
-      std::size_t stream_idx,
+      std::shared_ptr<ParallelUploadStateImpl> state, std::size_t stream_idx,
       std::unique_ptr<ResumableUploadSession> upload_session,
       std::size_t max_buffer_size,
       std::unique_ptr<HashValidator> hash_validator)
@@ -40,20 +39,26 @@ class ParallelObjectWriteStreambuf : public ObjectWriteStreambuf {
   }
 
  private:
-  std::shared_ptr<NonResumableParallelUploadState::Impl> state_;
+  std::shared_ptr<ParallelUploadStateImpl> state_;
   std::size_t stream_idx_;
 };
 
-NonResumableParallelUploadState::Impl::Impl(
-    std::unique_ptr<ScopedDeleter> deleter, Composer composer)
+ParallelUploadStateImpl::ParallelUploadStateImpl(
+    std::string destination_object_name, std::shared_ptr<ScopedDeleter> deleter,
+    Composer composer)
     : deleter_(std::move(deleter)),
       composer_(std::move(composer)),
+      destination_object_name_(std::move(destination_object_name)),
       finished_{},
       num_unfinished_streams_{} {}
 
-NonResumableParallelUploadState::Impl::~Impl() { WaitForCompletion().wait(); }
+ParallelUploadStateImpl::~ParallelUploadStateImpl() {
+  std::cout << "Trying to destroy impl" << std::endl;
+  WaitForCompletion().wait();
+  std::cout << "Destroyed impl" << std::endl;
+}
 
-StatusOr<ObjectWriteStream> NonResumableParallelUploadState::Impl::CreateStream(
+StatusOr<ObjectWriteStream> ParallelUploadStateImpl::CreateStream(
     RawClient& raw_client, ResumableUploadRequest const& request) {
   auto session = raw_client.CreateResumableSession(request);
   std::unique_lock<std::mutex> lk(mu_);
@@ -69,7 +74,7 @@ StatusOr<ObjectWriteStream> NonResumableParallelUploadState::Impl::CreateStream(
           CreateHashValidator(request)));
 }
 
-Status NonResumableParallelUploadState::Impl::EagerCleanup() {
+Status ParallelUploadStateImpl::EagerCleanup() {
   std::unique_lock<std::mutex> lk(mu_);
   if (!finished_) {
     return Status(StatusCode::kFailedPrecondition,
@@ -84,7 +89,7 @@ Status NonResumableParallelUploadState::Impl::EagerCleanup() {
   return cleanup_status_;
 }
 
-void NonResumableParallelUploadState::Impl::Fail(Status status) {
+void ParallelUploadStateImpl::Fail(Status status) {
   std::unique_lock<std::mutex> lk(mu_);
   assert(!status.ok());
   if (!res_) {
@@ -93,7 +98,7 @@ void NonResumableParallelUploadState::Impl::Fail(Status status) {
   }
 }
 
-void NonResumableParallelUploadState::Impl::StreamFinished(
+void ParallelUploadStateImpl::StreamFinished(
     std::size_t stream_idx, StatusOr<ResumableUploadResponse> const& response) {
   std::unique_lock<std::mutex> lk(mu_);
 
@@ -130,8 +135,8 @@ void NonResumableParallelUploadState::Impl::StreamFinished(
   }
 }
 
-future<StatusOr<ObjectMetadata>>
-NonResumableParallelUploadState::Impl::WaitForCompletion() const {
+future<StatusOr<ObjectMetadata>> ParallelUploadStateImpl::WaitForCompletion()
+    const {
   std::unique_lock<std::mutex> lk(mu_);
 
   if (finished_) {

google/cloud/storage/parallel_upload.cc

@@ -69,6 +69,58 @@ namespace internal {
 
 class ParallelObjectWriteStreambuf;
 
+// Type-erased function object to execute ComposeMany with most arguments
+// bound.
+using Composer = std::function<StatusOr<ObjectMetadata>(
+    std::vector<ComposeSourceObject> const&)>;
+
+// The `ObjectWriteStream`s have to hold references to the state of
+// the parallel upload so that they can update it when finished and trigger
+// shards composition, hence `ResumableParallelUploadState` has to be
+// destroyed after the `ObjectWriteStream`s.
+// `ResumableParallelUploadState` and `ObjectWriteStream`s are passed
+// around by values, so we don't control their lifetime. In order to
+// circumvent it, we move the state to something held by a `shared_ptr`.
+class ParallelUploadStateImpl
+    : public std::enable_shared_from_this<ParallelUploadStateImpl> {
+ public:
+  ParallelUploadStateImpl(std::string destination_object_name,
+                          std::shared_ptr<ScopedDeleter> deleter,
+                          Composer composer);
+  ~ParallelUploadStateImpl();
+
+  StatusOr<ObjectWriteStream> CreateStream(
+      RawClient& raw_client, ResumableUploadRequest const& request);
+
+  void StreamFinished(std::size_t stream_idx,
+                      StatusOr<ResumableUploadResponse> const& response);
+
+  future<StatusOr<ObjectMetadata>> WaitForCompletion() const;
+
+  Status EagerCleanup();
+
+  void Fail(Status status);
+
+ private:
+  mutable std::mutex mu_;
+  // Promises made via `WaitForCompletion()`
+  mutable std::vector<promise<StatusOr<ObjectMetadata>>> res_promises_;
+  // Type-erased object for deleting temporary objects.
+  std::shared_ptr<ScopedDeleter> deleter_;
+  // Type-erased function object to execute ComposeMany with most arguments
+  // bound.
+  std::function<StatusOr<ObjectMetadata>(std::vector<ComposeSourceObject>)>
+      composer_;
+  std::string destination_object_name_;
+  // Set when all streams are closed and composed but before cleanup.
+  bool finished_;
+  // Tracks how many streams are still written to.
+  std::size_t num_unfinished_streams_;
+  std::vector<ComposeSourceObject> to_compose_;
+  google::cloud::optional<StatusOr<ObjectMetadata>> res_;
+  Status cleanup_status_;
+};
+
 struct ComposeManyApplyHelper {
   template <typename... Options>
   StatusOr<ObjectMetadata> operator()(Options&&... options) const {
@@ -168,62 +220,15 @@ class NonResumableParallelUploadState {
   void Fail(Status status) { return impl_->Fail(std::move(status)); }
 
  private:
-  // Type-erased function object to execute ComposeMany with most arguments
-  // bound.
-  using Composer = std::function<StatusOr<ObjectMetadata>(
-      std::vector<ComposeSourceObject> const&)>;
-
-  // The `ObjectWriteStream`s have to hold references to the state of
-  // the parallel upload so that they can update it when finished and trigger
-  // shards composition, hence `NonResumableParallelUploadState` has to be
-  // destroyed after the `ObjectWriteStream`s.
-  // `NonResumableParallelUploadState` and `ObjectWriteStream`s are passed
-  // around by values, so we don't control their lifetime. In order to
-  // circumvent it, we move the state to something held by a `shared_ptr`.
-  class Impl : public std::enable_shared_from_this<Impl> {
-   public:
-    Impl(std::unique_ptr<ScopedDeleter> deleter, Composer composer);
-    ~Impl();
-
-    StatusOr<ObjectWriteStream> CreateStream(
-        RawClient& raw_client, ResumableUploadRequest const& request);
-
-    void StreamFinished(std::size_t stream_idx,
-                        StatusOr<ResumableUploadResponse> const& response);
-
-    future<StatusOr<ObjectMetadata>> WaitForCompletion() const;
-
-    Status EagerCleanup();
-
-    void Fail(Status status);
-
-   private:
-    mutable std::mutex mu_;
-    // Promises made via `WaitForCompletion()`
-    mutable std::vector<promise<StatusOr<ObjectMetadata>>> res_promises_;
-    // Type-erased object for deleting temporary objects.
-    std::unique_ptr<ScopedDeleter> deleter_;
-    // Type-erased function object to execute ComposeMany with most arguments
-    // bound.
-    std::function<StatusOr<ObjectMetadata>(std::vector<ComposeSourceObject>)>
-        composer_;
-    // Set when all streams are closed and composed but before cleanup.
-    bool finished_;
-    // Tracks how many streams are still written to.
-    std::size_t num_unfinished_streams_;
-    std::vector<ComposeSourceObject> to_compose_;
-    google::cloud::optional<StatusOr<ObjectMetadata>> res_;
-    Status cleanup_status_;
-  };
-
-  NonResumableParallelUploadState(std::shared_ptr<Impl> state,
-                                  std::vector<ObjectWriteStream> shards)
+  NonResumableParallelUploadState(
+      std::shared_ptr<ParallelUploadStateImpl> state,
+      std::vector<ObjectWriteStream> shards)
       : impl_(std::move(state)), shards_(std::move(shards)) {}
 
-  std::shared_ptr<Impl> impl_;
+  std::shared_ptr<ParallelUploadStateImpl> impl_;
   std::vector<ObjectWriteStream> shards_;
 
-  friend class ParallelObjectWriteStreambuf;
+  friend class NonResumableParallelObjectWriteStreambuf;
 };
 
 /**
@@ -306,8 +311,8 @@ NonResumableParallelUploadState::Create(Client client,
   }
   deleter->Add(*lock);
 
-  auto internal_state = std::make_shared<NonResumableParallelUploadState::Impl>(
-      std::move(deleter), std::move(composer));
+  auto internal_state = std::make_shared<ParallelUploadStateImpl>(
+      object_name, std::move(deleter), std::move(composer));
   std::vector<ObjectWriteStream> streams;
 
   auto upload_options = StaticTupleFilter<