Non throwing object read stream (#1750)

coryan · web-flow · commit 196e435e53ac · 2019-01-01T11:50:11.000-05:00
When possible, use `status()` to report errors in `ObjectReadStream`.
Unfortunately the standard library assumes that exceptions are enabled,
and a `std::basic_istreambuf&lt;&gt;` can only report errors by raising
exceptions. When exceptions are disabled this is, hmmm, difficult, so we
must rely on the application dutifully checking the `status()` flag.
diff --git a/google/cloud/storage/client.cc b/google/cloud/storage/client.cc
@@ -147,39 +147,40 @@ StatusOr<ObjectMetadata> Client::UploadStreamResumable(
 
 void Client::DownloadFileImpl(internal::ReadObjectRangeRequest const& request,
                               std::string const& file_name) {
+  // TODO(#1665) - use Status to report errors.
   std::unique_ptr<internal::ObjectReadStreambuf> streambuf =
       raw_client_->ReadObject(request).value();
+  // Open the download stream and immediately raise an exception on failure.
   ObjectReadStream stream(std::move(streambuf));
-  if (stream.eof() and not stream.IsOpen()) {
-    std::string msg = __func__;
-    msg += ": cannot open source object ";
-    msg += request.object_name();
-    msg += " in bucket ";
-    msg += request.bucket_name();
-    google::cloud::internal::RaiseRuntimeError(msg);
+
+  auto report_error = [&](char const* func, char const* what) {
+    std::ostringstream msg;
+    msg << func << "(" << request << ", " << file_name
+        << "): " << what << " - status=" << stream.status();
+    google::cloud::internal::RaiseRuntimeError(std::move(msg).str());
+  };
+  if (not stream.status().ok()) {
+    report_error(__func__, "cannot open download stream");
   }
+
+  // Open the destination file, and immediate raise an exception on failure.
   std::ofstream os(file_name);
   if (not os.is_open()) {
-    std::string msg = __func__;
-    msg += ": cannot open destination file ";
-    msg += file_name;
-    google::cloud::internal::RaiseRuntimeError(msg);
+    report_error(__func__, "cannot open destination file");
   }
+
   std::string buffer;
   buffer.resize(raw_client_->client_options().download_buffer_size(), '\0');
   do {
     stream.read(&buffer[0], buffer.size());
     os.write(buffer.data(), stream.gcount());
-    if (not stream.good()) {
-      break;
-    }
-  } while (os.good());
+  } while (os.good() and stream.good());
   os.close();
   if (not os.good()) {
-    std::string msg = __func__;
-    msg += ": failure closing destination file ";
-    msg += file_name;
-    google::cloud::internal::RaiseRuntimeError(msg);
+    report_error(__func__, "error closing destination file");
+  }
+  if (not stream.status().ok()) {
+    report_error(__func__, "error in download stream");
   }
 }
 
diff --git a/google/cloud/storage/client.h b/google/cloud/storage/client.h
@@ -786,6 +786,14 @@ class Client {
   /**
    * Reads the contents of an object.
    *
+   * Returns an object derived from `std::istream` which can be used to read the
+   * contents of the GCS blob. The application should check the `badbit` (e.g.
+   * by calling `stream.bad()`) on the returned object to detect if there was
+   * an error reading from the blob. If `badbit` is set, the application can
+   * check the `status()` variable to get details about the failure.
+   * Applications can also set the exception mask on the returned stream, in
+   * which case an exception is thrown if an error is detected.
+   *
    * @param bucket_name the name of the bucket that contains the object.
    * @param object_name the name of the object to be read.
    * @param options a list of optional query parameters and/or request headers.
diff --git a/google/cloud/storage/internal/curl_download_request.cc b/google/cloud/storage/internal/curl_download_request.cc
@@ -42,11 +42,14 @@ StatusOr<HttpResponse> CurlDownloadRequest::Close() {
   // callback, see the comments in the header file for more details.
   closing_ = true;
   // Block until that callback is made.
-  Wait([this] { return curl_closed_; });
+  auto status = Wait([this] { return curl_closed_; });
+  if (not status.ok()) {
+    return status;
+  }
 
   // Now remove the handle from the CURLM* interface and wait for the response.
   auto error = curl_multi_remove_handle(multi_.get(), handle_.handle_.get());
-  auto status = AsStatus(error, __func__);
+  status = AsStatus(error, __func__);
   if (not status.ok()) {
     return status;
   }
@@ -61,15 +64,18 @@ StatusOr<HttpResponse> CurlDownloadRequest::Close() {
 
 StatusOr<HttpResponse> CurlDownloadRequest::GetMore(std::string& buffer) {
   handle_.FlushDebug(__func__);
-  Wait([this] {
+  auto status = Wait([this] {
     return curl_closed_ or buffer_.size() >= initial_buffer_size_;
   });
+  if (not status.ok()) {
+    return status;
+  }
   GCP_LOG(DEBUG) << __func__ << "(), curl.size=" << buffer_.size()
                  << ", closing=" << closing_ << ", closed=" << curl_closed_;
   if (curl_closed_) {
     // Remove the handle from the CURLM* interface and wait for the response.
     auto error = curl_multi_remove_handle(multi_.get(), handle_.handle_.get());
-    Status status = AsStatus(error, __func__);
+    status = AsStatus(error, __func__);
     if (not status.ok()) {
       return status;
     }
@@ -89,9 +95,9 @@ StatusOr<HttpResponse> CurlDownloadRequest::GetMore(std::string& buffer) {
   buffer_.swap(buffer);
   buffer_.clear();
   buffer_.reserve(initial_buffer_size_);
-  Status pause = handle_.EasyPause(CURLPAUSE_RECV_CONT);
-  if (not pause.ok()) {
-    return pause;
+  status = handle_.EasyPause(CURLPAUSE_RECV_CONT);
+  if (not status.ok()) {
+    return status;
   }
   GCP_LOG(DEBUG) << __func__ << "(), size=" << buffer.size()
                  << ", closing=" << closing_ << ", closed=" << curl_closed_
diff --git a/google/cloud/storage/internal/curl_streambuf.cc b/google/cloud/storage/internal/curl_streambuf.cc
@@ -35,17 +35,20 @@ CurlReadStreambuf::CurlReadStreambuf(
 
 bool CurlReadStreambuf::IsOpen() const { return download_.IsOpen(); }
 
-HttpResponse CurlReadStreambuf::Close() {
-  // TODO(#1736) - return StatusOr<> from here.
-  return download_.Close().value();
+void CurlReadStreambuf::Close() {
+  auto response = download_.Close();
+  if (not response.ok()) {
+    status_ = std::move(response).status();
+    ReportError(status_);
+  }
 }
 
 CurlReadStreambuf::int_type CurlReadStreambuf::underflow() {
   if (not IsOpen()) {
-    current_ios_buffer_.clear();
-    current_ios_buffer_.push_back('\0');
-    char* data = &current_ios_buffer_[0];
-    setg(data, data + 1, data + 1);
+    // The stream is closed, reading from a closed stream can happen if there is
+    // no object to read from, or the object is empty. In that case just setup
+    // an empty (but valid) region and verify the checksums.
+    SetEmptyRegion();
     hash_validator_result_ = HashValidator::FinishAndCheck(
         __func__ + std::string(" mismatched hashes reading from closed stream"),
         std::move(*hash_validator_));
@@ -55,13 +58,15 @@ CurlReadStreambuf::int_type CurlReadStreambuf::underflow() {
   current_ios_buffer_.reserve(target_buffer_size_);
   StatusOr<HttpResponse> response = download_.GetMore(current_ios_buffer_);
   if (not response.ok()) {
-    return traits_type::eof();
+    return ReportError(std::move(response).status());
   }
   for (auto const& kv : response->headers) {
     hash_validator_->ProcessHeader(kv.first, kv.second);
+    headers_.emplace(kv.first, kv.second);
   }
   if (response->status_code >= 300) {
-    return traits_type::eof();
+    return ReportError(
+        Status(response->status_code, std::move(response->payload)));
   }
 
   if (not current_ios_buffer_.empty()) {
@@ -70,15 +75,41 @@ CurlReadStreambuf::int_type CurlReadStreambuf::underflow() {
     setg(data, data, data + current_ios_buffer_.size());
     return traits_type::to_int_type(*data);
   }
-  current_ios_buffer_.push_back('\0');
-  char* data = &current_ios_buffer_[0];
-  setg(data, data + 1, data + 1);
+
+  // This is an actual EOF, there is no more data to download, create an
+  // empty (but valid) region:
+  SetEmptyRegion();
+  // Verify the checksums, and return the EOF character.
   hash_validator_result_ = HashValidator::FinishAndCheck(
       __func__ + std::string(" mismatched hashes at end of download"),
       std::move(*hash_validator_));
   return traits_type::eof();
 }
 
+CurlReadStreambuf::int_type CurlReadStreambuf::ReportError(Status status) {
+  // The only way to report errors from a std::basic_streambuf<> (which this
+  // class derives from) is to throw exceptions:
+  //   https://stackoverflow.com/questions/50716688/how-to-set-the-badbit-of-a-stream-by-a-customized-streambuf
+  // but we need to be able to report errors when the application has disabled
+  // exceptions via `-fno-exceptions` or a similar option. In that case we set
+  // `status_`, and report the error as an EOF. This is obviously not ideal,
+  // but it is the best we can do when the application disables the standard
+  // mechanism to signal errors.
+  status_ = std::move(status);
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+  internal::ThrowStatus(status_);
+#else
+  return traits_type::eof();
+#endif  // GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+}
+
+void CurlReadStreambuf::SetEmptyRegion() {
+  current_ios_buffer_.clear();
+  current_ios_buffer_.push_back('\0');
+  char* data = &current_ios_buffer_[0];
+  setg(data, data + 1, data + 1);
+}
+
 CurlWriteStreambuf::CurlWriteStreambuf(
     CurlUploadRequest&& upload, std::size_t max_buffer_size,
     std::unique_ptr<HashValidator> hash_validator)
diff --git a/google/cloud/storage/internal/curl_streambuf.h b/google/cloud/storage/internal/curl_streambuf.h
@@ -36,25 +36,35 @@ class CurlReadStreambuf : public ObjectReadStreambuf {
 
   ~CurlReadStreambuf() override = default;
 
-  HttpResponse Close() override;
+  void Close() override;
   bool IsOpen() const override;
+  Status const& status() const override { return status_; }
   std::string const& received_hash() const override {
     return hash_validator_result_.received;
   }
   std::string const& computed_hash() const override {
     return hash_validator_result_.computed;
   }
+  std::multimap<std::string, std::string> const& headers() const override {
+    return headers_;
+  }
 
  protected:
   int_type underflow() override;
 
+  int_type ReportError(Status status);
+
+  void SetEmptyRegion();
+
  private:
   CurlDownloadRequest download_;
   std::string current_ios_buffer_;
   std::size_t target_buffer_size_;
 
   std::unique_ptr<HashValidator> hash_validator_;
   HashValidator::Result hash_validator_result_;
+  Status status_;
+  std::multimap<std::string, std::string> headers_;
 };
 
 /**
diff --git a/google/cloud/storage/internal/object_streambuf.h b/google/cloud/storage/internal/object_streambuf.h
@@ -16,6 +16,7 @@
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_OBJECT_STREAMBUF_H_
 
 #include "google/cloud/storage/internal/http_response.h"
+#include "google/cloud/storage/status.h"
 #include <iostream>
 
 namespace google {
@@ -41,10 +42,12 @@ class ObjectReadStreambuf : public std::basic_streambuf<char> {
   ObjectReadStreambuf(ObjectReadStreambuf const&) = delete;
   ObjectReadStreambuf& operator=(ObjectReadStreambuf const&) = delete;
 
-  virtual HttpResponse Close() = 0;
+  virtual void Close() = 0;
   virtual bool IsOpen() const = 0;
+  virtual Status const& status() const = 0;
   virtual std::string const& received_hash() const = 0;
   virtual std::string const& computed_hash() const = 0;
+  virtual std::multimap<std::string, std::string> const& headers() const = 0;
 };
 
 /**
diff --git a/google/cloud/storage/object_stream.cc b/google/cloud/storage/object_stream.cc
@@ -44,12 +44,14 @@ ObjectReadStream::~ObjectReadStream() {
 #endif  // GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
 }
 
-internal::HttpResponse ObjectReadStream::Close() {
+void ObjectReadStream::Close() {
   if (not IsOpen()) {
-    google::cloud::internal::RaiseRuntimeError(
-        "Attempting to Close() closed ObjectReadStream");
+    return;
+  }
+  buf_->Close();
+  if (not status().ok()) {
+    setstate(std::ios_base::badbit);
   }
-  return buf_->Close();
 }
 
 ObjectWriteStream::~ObjectWriteStream() {
diff --git a/google/cloud/storage/object_stream.h b/google/cloud/storage/object_stream.h
@@ -63,11 +63,64 @@ class ObjectReadStream : public std::basic_istream<char> {
   /// Closes the stream (if necessary).
   ~ObjectReadStream() override;
 
-  bool IsOpen() const { return buf_.get() != nullptr and buf_->IsOpen(); }
-  internal::HttpResponse Close();
+  bool IsOpen() const { return (bool)buf_ and buf_->IsOpen(); }
+
+  /**
+   * Terminate the download, possibly before completing it.
+   */
+  void Close();
+
+  //@{
+  /**
+   * Report any download errors.
+   *
+   * Note that errors may go undetected until the download completes.
+   */
+  Status const& status() const& { return buf_->status(); }
+
+  /**
+   * The received CRC32C checksum and the MD5 hash values as reported by GCS.
+   *
+   * When the download is finalized (via `Close()` or the end of file) the GCS
+   * server reports the CRC32C checksum and, except for composite objects, the
+   * MD5 hash of the data. This class compares the locally computed and received
+   * hashes so applications can detect data download errors.
+   *
+   * The values are reported as comma separated `tag=value` pairs, e.g.
+   * `crc32c=AAAAAA==,md5=1B2M2Y8AsgTpgAmY7PhCfg==`. The format of this string
+   * is subject to change without notice, they are provided for informational
+   * purposes only.
+   *
+   * @see https://cloud.google.com/storage/docs/hashes-etags for more
+   *     information on checksums and hashes in GCS.
+   */
   std::string const& received_hash() const { return buf_->received_hash(); }
+
+  /**
+   * The locally computed checksum and hashes, as a string.
+   *
+   * This object computes the CRC32C checksum and MD5 hash of the downloaded
+   * data. Note that there are several cases where these values may be empty or
+   * irrelevant, for example:
+   *   - When reading only a portion of a blob the hash of that portion is
+   *     irrelevant, note that GCS only reports the hashes for the full blob.
+   *   - The application may disable the CRC32C and/or the MD5 hash computation.
+   *
+   * The string has the same format as the value returned by `received_hash()`.
+   * Note that the format of this string is also subject to change without
+   * notice.
+   *
+   * @see https://cloud.google.com/storage/docs/hashes-etags for more
+   *     information on checksums and hashes in GCS.
+   */
   std::string const& computed_hash() const { return buf_->computed_hash(); }
 
+  /// The headers returned by the service, for debugging only.
+  std::multimap<std::string, std::string> const& headers() const {
+    return buf_->headers();
+  }
+  //@}
+
  private:
   std::unique_ptr<internal::ObjectReadStreambuf> buf_;
 };
diff --git a/google/cloud/storage/tests/object_integration_test.cc b/google/cloud/storage/tests/object_integration_test.cc
@@ -327,8 +327,12 @@ TEST_F(ObjectIntegrationTest, ReadNotFound) {
 
   // Create a iostream to read the object back.
   auto stream = client.ReadObject(bucket_name, object_name);
-  EXPECT_TRUE(stream.eof());
+  EXPECT_FALSE(stream.status().ok());
   EXPECT_FALSE(stream.IsOpen());
+  EXPECT_EQ(404, stream.status().status_code()) << "status=" << stream.status();
+#if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+  EXPECT_TRUE(stream.bad());
+#endif  // GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
 }
 
 TEST_F(ObjectIntegrationTest, Copy) {
@@ -1158,12 +1162,11 @@ TEST_F(ObjectIntegrationTest, StreamingWriteFailure) {
 
   // This operation should fail because the object already exists.
 #if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
-  EXPECT_THROW(
-      try { os.Close(); } catch (std::runtime_error const& ex) {
-        EXPECT_THAT(ex.what(), HasSubstr("[412]"));
-        throw;
-      },
-      std::runtime_error);
+  EXPECT_THROW(try { os.Close(); } catch (std::runtime_error const& ex) {
+    EXPECT_THAT(ex.what(), HasSubstr("[412]"));
+    throw;
+  },
+               std::runtime_error);
 #else
   EXPECT_DEATH_IF_SUPPORTED(os.Close(), "exceptions are disabled");
 #endif  // GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
diff --git a/google/cloud/storage/tests/object_media_integration_test.cc b/google/cloud/storage/tests/object_media_integration_test.cc

Original file line number	Diff line number	Diff line change
`@@ -44,12 +44,14 @@ ObjectReadStream::~ObjectReadStream() {`
`44`	`44`	`#endif // GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS`
`45`	`45`	`}`
`46`	`46`
`47`		`-internal::HttpResponse ObjectReadStream::Close() {`
	`47`	`+void ObjectReadStream::Close() {`
`48`	`48`	`if (not IsOpen()) {`
`49`		`- google::cloud::internal::RaiseRuntimeError(`
`50`		`- "Attempting to Close() closed ObjectReadStream");`
	`49`	`+ return;`
	`50`	`+ }`
	`51`	`+ buf_->Close();`
	`52`	`+ if (not status().ok()) {`
	`53`	`+ setstate(std::ios_base::badbit);`
`51`	`54`	`}`
`52`		`- return buf_->Close();`
`53`	`55`	`}`
`54`	`56`
`55`	`57`	`ObjectWriteStream::~ObjectWriteStream() {`