Merge pull request #10709 from ctiller/infinite_stream

Fix infinite streams in qps_test

1 files changed, 16 insertions, 8 deletions

diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc
index a020adde51..f8ce2cccbe 100644
--- a/test/cpp/qps/client_sync.cc
+++ b/test/cpp/qps/client_sync.cc
@@ -153,16 +153,22 @@ class SynchronousStreamingClient final : public SynchronousClient {
     StartThreads(num_threads_);
   }
   ~SynchronousStreamingClient() {
+    std::vector<std::thread> cleanup_threads;
     for (size_t i = 0; i < num_threads_; i++) {
-      auto stream = &stream_[i];
-      if (*stream) {
-        (*stream)->WritesDone();
-        Status s = (*stream)->Finish();
-        if (!s.ok()) {
-          gpr_log(GPR_ERROR, "Stream %" PRIuPTR " received an error %s", i,
-                  s.error_message().c_str());
+      cleanup_threads.emplace_back([this, i]() {
+        auto stream = &stream_[i];
+        if (*stream) {
+          (*stream)->WritesDone();
+          Status s = (*stream)->Finish();
+          if (!s.ok()) {
+            gpr_log(GPR_ERROR, "Stream %" PRIuPTR " received an error %s", i,
+                    s.error_message().c_str());
+          }
         }
-      }
+      });
+    }
+    for (size_t i = 0; i < num_threads_; i++) {
+      cleanup_threads[i].join();
     }
   }
 
@@ -179,6 +185,8 @@ class SynchronousStreamingClient final : public SynchronousClient {
       if ((messages_per_stream_ != 0) &&
           (++messages_issued_[thread_idx] < messages_per_stream_)) {
         return true;
+      } else if (messages_per_stream_ == 0) {
+        return true;
       } else {
         // Fall through to the below resetting code after finish
       }