Make GCS throughput benchmark record progress.

google/cloud/storage/benchmarks/CMakeLists.txt

@@ -72,7 +72,8 @@ if (BUILD_TESTING)
     set(storage_benchmarks_unit_tests
         benchmark_parser_test.cc
         benchmark_make_random_test.cc
-        benchmark_parse_args_test.cc)
+        benchmark_parse_args_test.cc
+        benchmark_utils_test.cc)
 
     foreach (fname ${storage_benchmarks_unit_tests})
         string(REPLACE "/"

google/cloud/storage/benchmarks/benchmark_utils.cc

@@ -279,6 +279,23 @@ bool SimpleTimer::SupportPerThreadUsage() {
 #endif  // GOOGLE_CLOUD_CPP_HAVE_RUSAGE_THREAD
 }
 
+void ProgressReporter::Start() {
+  progress_.clear();
+  start_ = std::chrono::steady_clock::now();
+  progress_.emplace_back(TimePoint{0, std::chrono::microseconds(0)});
+}
+
+void ProgressReporter::Advance(size_t progress) {
+  progress_.emplace_back(
+      TimePoint{progress, std::chrono::duration_cast<std::chrono::microseconds>(
+                              std::chrono::steady_clock::now() - start_)});
+}
+
+std::vector<ProgressReporter::TimePoint> const&
+ProgressReporter::GetAccumulatedProgress() const {
+  return progress_;
+}
+
 }  // namespace storage_benchmarks
 }  // namespace cloud
 }  // namespace google

google/cloud/storage/benchmarks/benchmark_utils.h

@@ -119,6 +119,24 @@ class SimpleTimer {
   std::string annotations_;
 };
 
+/**
+ * A small class to accumulate time-points with progress.
+ */
+class ProgressReporter {
+ public:
+  struct TimePoint {
+    std::uint64_t bytes;
+    std::chrono::microseconds elapsed;
+  };
+  void Start();
+  void Advance(size_t progress);
+  std::vector<TimePoint> const& GetAccumulatedProgress() const;
+
+ private:
+  std::chrono::steady_clock::time_point start_;
+  std::vector<TimePoint> progress_;
+};
+
 }  // namespace storage_benchmarks
 }  // namespace cloud
 }  // namespace google

google/cloud/storage/benchmarks/benchmark_utils_test.cc

@@ -0,0 +1,44 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/storage/benchmarks/benchmark_utils.h"
+#include <gmock/gmock.h>
+#include <thread>
+
+namespace google {
+namespace cloud {
+namespace storage_benchmarks {
+namespace {
+
+TEST(ProgressReporterTest, Trivial) {
+  ProgressReporter rep;
+  rep.Start();
+  std::this_thread::sleep_for(std::chrono::milliseconds(2));
+  rep.Advance(5);
+  std::this_thread::sleep_for(std::chrono::milliseconds(3));
+  rep.Advance(7);
+  auto res = rep.GetAccumulatedProgress();
+  EXPECT_EQ(3U, res.size());
+  EXPECT_EQ(0, res[0].bytes);
+  EXPECT_EQ(0, res[0].elapsed.count());
+  EXPECT_EQ(5, res[1].bytes);
+  EXPECT_LE(2000, res[1].elapsed.count());
+  EXPECT_EQ(7, res[2].bytes);
+  EXPECT_LE(3000, res[2].elapsed.count());
+}
+
+}  // namespace
+}  // namespace storage_benchmarks
+}  // namespace cloud
+}  // namespace google

google/cloud/storage/benchmarks/storage_benchmarks_unit_tests.bzl

@@ -20,4 +20,5 @@ storage_benchmarks_unit_tests = [
     "benchmark_parser_test.cc",
     "benchmark_make_random_test.cc",
     "benchmark_parse_args_test.cc",
+    "benchmark_utils_test.cc",
 ]

google/cloud/storage/benchmarks/storage_throughput_vs_cpu_benchmark.cc

@@ -104,6 +104,7 @@ struct IterationResult {
   std::chrono::microseconds elapsed_time;
   std::chrono::microseconds cpu_time;
   google::cloud::StatusCode status;
+  std::vector<gcs_bm::ProgressReporter::TimePoint> progress;
 };
 using TestResults = std::vector<IterationResult>;
 
@@ -225,11 +226,28 @@ char const* ToString(OpType type) {
   return nullptr;  // silence g++ error.
 }
 
+std::ostream& operator<<(std::ostream& os,
+                         gcs_bm::ProgressReporter::TimePoint const& point) {
+  return os << point.bytes << '|' << point.elapsed.count();
+}
+
+std::ostream& operator<<(
+    std::ostream& os,
+    std::vector<gcs_bm::ProgressReporter::TimePoint> const& points) {
+  char const* sep = "";
+  for (auto const& point : points) {
+    os << sep << point;
+    sep = ";";
+  }
+  return os;
+}
+
 std::ostream& operator<<(std::ostream& os, IterationResult const& rhs) {
   return os << ToString(rhs.op) << ',' << rhs.object_size << ','
             << rhs.chunk_size << ',' << rhs.buffer_size << ','
             << rhs.elapsed_time.count() << ',' << rhs.cpu_time.count() << ','
-            << rhs.status << ',' << google::cloud::storage::version_string();
+            << rhs.status << ',' << rhs.progress << ','
+            << google::cloud::storage::version_string();
 }
 
 void PrintResults(TestResults const& results) {
@@ -281,7 +299,9 @@ TestResults RunThread(Options const& options, std::string const& bucket_name) {
     auto object_size = size_generator(generator);
     auto chunk_size = chunk_generator(generator);
 
+    gcs_bm::ProgressReporter progress;
     timer.Start();
+    progress.Start();
     auto writer =
         client.WriteObject(bucket_name, object_name,
                            gcs::DisableCrc32cChecksum(options.disable_crc32c),
@@ -292,33 +312,39 @@ TestResults RunThread(Options const& options, std::string const& bucket_name) {
         len = object_size - offset;
       }
       writer.write(contents.data() + offset, len);
+      progress.Advance(offset);
     }
     writer.Close();
+    progress.Advance(object_size);
     timer.Stop();
 
     auto object_metadata = writer.metadata();
-    results.emplace_back(IterationResult{OP_UPLOAD, object_size, chunk_size,
-                                         download_buffer_size,
-                                         timer.elapsed_time(), timer.cpu_time(),
-                                         object_metadata.status().code()});
+    results.emplace_back(IterationResult{
+        OP_UPLOAD, object_size, chunk_size, download_buffer_size,
+        timer.elapsed_time(), timer.cpu_time(), object_metadata.status().code(),
+        progress.GetAccumulatedProgress()});
 
     if (!object_metadata) {
       continue;
     }
 
     timer.Start();
+    progress.Start();
     auto reader =
         client.ReadObject(object_metadata->bucket(), object_metadata->name(),
                           gcs::Generation(object_metadata->generation()),
                           gcs::DisableCrc32cChecksum(options.disable_crc32c),
                           gcs::DisableMD5Hash(options.disable_md5));
+    progress.Advance(0);
     std::vector<char> buffer(chunk_size);
-    while (reader.read(buffer.data(), buffer.size())) {
+    for (size_t num_read = 0; reader.read(buffer.data(), buffer.size());
+         progress.Advance(num_read += reader.gcount())) {
     }
     timer.Stop();
     results.emplace_back(IterationResult{
         OP_DOWNLOAD, object_size, chunk_size, upload_buffer_size,
-        timer.elapsed_time(), timer.cpu_time(), reader.status().code()});
+        timer.elapsed_time(), timer.cpu_time(), reader.status().code(),
+        progress.GetAccumulatedProgress()});
 
     auto status =
         client.DeleteObject(object_metadata->bucket(), object_metadata->name(),