Merge pull request #6737 from ctiller/delayed-write

Delay beginning most writes until we enter poll()

5 files changed, 35 insertions, 21 deletions

diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 354a59cedd..0f87ae3e44 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -1166,6 +1166,9 @@ TEST_P(ProxyEnd2endTest, HugeResponse) {
   request.mutable_param()->set_response_message_length(kResponseSize);
 
   ClientContext context;
+  std::chrono::system_clock::time_point deadline =
+      std::chrono::system_clock::now() + std::chrono::seconds(20);
+  context.set_deadline(deadline);
   Status s = stub_->Echo(&context, request, &response);
   EXPECT_EQ(kResponseSize, response.message().size());
   EXPECT_TRUE(s.ok());
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 1507d1e3d6..5dd0bd8533 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -206,21 +206,30 @@ class AsyncClient : public ClientImpl<StubType, RequestType> {
     void* got_tag;
     bool ok;
 
-    if (cli_cqs_[thread_idx]->Next(&got_tag, &ok)) {
-      // Got a regular event, so process it
-      ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
-      if (!ctx->RunNextState(ok, histogram)) {
-        // The RPC and callback are done, so clone the ctx
-        // and kickstart the new one
-        auto clone = ctx->StartNewClone();
-        clone->Start(cli_cqs_[thread_idx].get());
-        // delete the old version
-        delete ctx;
+    switch (cli_cqs_[thread_idx]->AsyncNext(
+        &got_tag, &ok,
+        std::chrono::system_clock::now() + std::chrono::milliseconds(10))) {
+      case CompletionQueue::SHUTDOWN:
+        return false;
+      case CompletionQueue::GOT_EVENT: {
+        // Got a regular event, so process it
+        ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
+        if (!ctx->RunNextState(ok, histogram)) {
+          // The RPC and callback are done, so clone the ctx
+          // and kickstart the new one
+          auto clone = ctx->StartNewClone();
+          clone->Start(cli_cqs_[thread_idx].get());
+          // delete the old version
+          delete ctx;
+        }
+        return true;
       }
-      return true;
-    } else {  // queue is shutting down
-      return false;
+      case CompletionQueue::TIMEOUT:
+        // TODO(ctiller): do something here to track how frequently we pass
+        // through this codepath.
+        return true;
     }
+    GPR_UNREACHABLE_CODE(return false);
   }
 
  protected:
diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py
index 34b8151441..4ff4e44b8b 100755
--- a/test/cpp/qps/gen_build_yaml.py
+++ b/test/cpp/qps/gen_build_yaml.py
@@ -45,9 +45,10 @@ import performance.scenario_config as scenario_config
 
 def _scenario_json_string(scenario_json):
   # tweak parameters to get fast test times
-  scenario_json['warmup_seconds'] = 1
+  scenario_json['warmup_seconds'] = 0
   scenario_json['benchmark_seconds'] = 1
-  return json.dumps(scenario_config.remove_nonproto_fields(scenario_json))
+  scenarios_json = {'scenarios': [scenario_config.remove_nonproto_fields(scenario_json)]}
+  return json.dumps(scenarios_json)
 
 def threads_of_type(scenario_json, path):
   d = scenario_json
@@ -72,8 +73,7 @@ print yaml.dump({
     {
       'name': 'json_run_localhost',
       'shortname': 'json_run_localhost:%s' % scenario_json['name'],
-      'args': ['--scenario_json',
-               pipes.quote(_scenario_json_string(scenario_json))],
+      'args': ['--scenarios_json', _scenario_json_string(scenario_json)],
       'ci_platforms': ['linux', 'mac', 'posix', 'windows'],
       'platforms': ['linux', 'mac', 'posix', 'windows'],
       'flaky': False,
@@ -81,7 +81,8 @@ print yaml.dump({
       'boringssl': True,
       'defaults': 'boringssl',
       'cpu_cost': guess_cpu(scenario_json),
-      'exclude_configs': []
+      'exclude_configs': [],
+      'timeout_seconds': 3*60
     }
     for scenario_json in scenario_config.CXXLanguage().scenarios()
   ]
diff --git a/test/cpp/qps/json_run_localhost.cc b/test/cpp/qps/json_run_localhost.cc
index 6545dc2917..74e40fbf1a 100644
--- a/test/cpp/qps/json_run_localhost.cc
+++ b/test/cpp/qps/json_run_localhost.cc
@@ -75,7 +75,7 @@ int main(int argc, char** argv) {
   for (int i = 1; i < argc; i++) {
     args.push_back(argv[i]);
   }
-  SubProcess(args).Join();
+  GPR_ASSERT(SubProcess(args).Join() == 0);
 
   for (auto it = jobs.begin(); it != jobs.end(); ++it) {
     (*it)->Interrupt();
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index c9954d0d02..73ca19148b 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -102,7 +102,7 @@ class AsyncQpsServerTest : public Server {
     auto process_rpc_bound =
         std::bind(process_rpc, config.payload_config(), _1, _2);
 
-    for (int i = 0; i < 10000 / num_threads; i++) {
+    for (int i = 0; i < 15000; i++) {
       for (int j = 0; j < num_threads; j++) {
         if (request_unary_function) {
           auto request_unary =
@@ -132,7 +132,8 @@ class AsyncQpsServerTest : public Server {
     for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
       (*ss)->set_shutdown();
     }
-    server_->Shutdown();
+    server_->Shutdown(std::chrono::system_clock::now() +
+                      std::chrono::seconds(3));
     for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
       thr->join();
     }