[C++] FS: Would S3 OutputStream CloseAsync meets lifetime issue?

[mapleFU]

Describe the enhancement requested

During CloseAsync:

  Future<> CloseAsync() override {
    if (closed_) return Status::OK();

    if (current_part_) {
      // Upload last part
      RETURN_NOT_OK(CommitCurrentPart());
    }

    // S3 mandates at least one part, upload an empty one if necessary
    if (part_number_ == 1) {
      RETURN_NOT_OK(UploadPart("", 0));
    }

    // Wait for in-progress uploads to finish (if async writes are enabled)
    return FlushAsync().Then([this]() {
      ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());

      // At this point, all part uploads have finished successfully
      DCHECK_GT(part_number_, 1);
      DCHECK_EQ(upload_state_->completed_parts.size(),
                static_cast<size_t>(part_number_ - 1));

      S3Model::CompletedMultipartUpload completed_upload;
      completed_upload.SetParts(upload_state_->completed_parts);
      S3Model::CompleteMultipartUploadRequest req;
      req.SetBucket(ToAwsString(path_.bucket));
      req.SetKey(ToAwsString(path_.key));
      req.SetUploadId(upload_id_);
      req.SetMultipartUpload(std::move(completed_upload));

      auto outcome =
          client_lock.Move()->CompleteMultipartUploadWithErrorFixup(std::move(req));
      if (!outcome.IsSuccess()) {
        return ErrorToStatus(
            std::forward_as_tuple("When completing multiple part upload for key '",
                                  path_.key, "' in bucket '", path_.bucket, "': "),
            "CompleteMultipartUpload", outcome.GetError());
      }

      holder_ = nullptr;
      closed_ = true;
      return Status::OK();
    });
  }

Notice that in FlushAsync().Then, the lambda capture this. When:

1. user call FlushAsync()
2. destruct OutputStream without waiting for future
3. Call Then([this]()...

Would this raise lifetime problem? Or here is some lifetime constraint I didn't catch?

Component(s)

C++

[mapleFU]

@pitrou @lidavidm since I'm not so familiar with s3fs, would you mind take a look? Would this be a problem? Or here is some lifetime constraint I didn't catch?

[pitrou]

Hmm... perhaps OutputStream should inherit from std::enable_shared_from_this like InputStream does, so that we can capture an owning pointer in the closure?

[mapleFU]

This looks ok to me, but would it break the user who is using std::unique_ptr<arrow::io::OutputStream> ?

https://en.cppreference.com/w/cpp/memory/enable_shared_from_this/shared_from_this should we capture std::bad_weak_ptr like:

std::shared_ptr<OutputStream> lifetime;
try {
   lifetime = shared_from_this();
} catch (.. )  {
}

...
FlushAsync().Then([lifetime, this]) {}

?

[pitrou]

This looks ok to me, but would it break the user who is using std::unique_ptr<arrow::io::OutputStream> ?

In which context would they get a unique_ptr? The public APIs return a shared_ptr:

arrow/cpp/src/arrow/filesystem/s3fs.h

Lines 299 to 311 in 4fac528

	/// Create a sequential output stream for writing to a S3 object.
	///
	/// NOTE: Writes to the stream will be buffered. Depending on
	/// S3Options.background_writes, they can be synchronous or not.
	/// It is recommended to enable background_writes unless you prefer
	/// implementing your own background execution strategy.
	Result<std::shared_ptr<io::OutputStream>> OpenOutputStream(
	const std::string& path,
	const std::shared_ptr<const KeyValueMetadata>& metadata = {}) override;
	Result<std::shared_ptr<io::OutputStream>> OpenAppendStream(
	const std::string& path,
	const std::shared_ptr<const KeyValueMetadata>& metadata = {}) override;

[mapleFU]

@pitrou Should I just wrap s3fs's ObjectOutputStream with enable_shared_from_this, or wrap the arrow::io::OutputStream with enable_shared_from_this? Wrap the ObjectOutputStream looks good to me. But I wonder would arrow::io::OutputStream harm?

[pitrou]

We should wrap arrow::io::OutputStream, since that's probably going to be useful for other filesystems.