Merge master

5 files changed, 53 insertions, 77 deletions

diff --git a/test/core/tsi/transport_security_test_lib.cc b/test/core/tsi/transport_security_test_lib.cc
index 454b79c0cc..b98ab9ba05 100644
--- a/test/core/tsi/transport_security_test_lib.cc
+++ b/test/core/tsi/transport_security_test_lib.cc
@@ -354,25 +354,28 @@ static void do_handshaker_next(handshaker_args *args) {
   tsi_handshaker_result *handshaker_result = NULL;
   unsigned char *bytes_to_send = NULL;
   size_t bytes_to_send_size = 0;
+  tsi_result result = TSI_OK;
   /* Receive data from peer, if available. */
-  size_t buf_size = args->handshake_buffer_size;
-  receive_bytes_from_peer(args->fixture, &args->handshake_buffer, &buf_size,
-                          args->is_client);
-  if (buf_size > 0) {
-    args->transferred_data = true;
-  }
-  /* Peform handshaker next. */
-  tsi_result result = tsi_handshaker_next(
-      handshaker, args->handshake_buffer, buf_size,
-      (const unsigned char **)&bytes_to_send, &bytes_to_send_size,
-      &handshaker_result, &on_handshake_next_done_wrapper, args);
-  if (result != TSI_ASYNC) {
-    args->error = on_handshake_next_done(result, args, bytes_to_send,
-                                         bytes_to_send_size, handshaker_result);
-    if (args->error != GRPC_ERROR_NONE) {
-      return;
+  do {
+    size_t buf_size = args->handshake_buffer_size;
+    receive_bytes_from_peer(args->fixture, &args->handshake_buffer, &buf_size,
+                            args->is_client);
+    if (buf_size > 0) {
+      args->transferred_data = true;
     }
-  }
+    /* Peform handshaker next. */
+    result = tsi_handshaker_next(handshaker, args->handshake_buffer, buf_size,
+                                 (const unsigned char **)&bytes_to_send,
+                                 &bytes_to_send_size, &handshaker_result,
+                                 &on_handshake_next_done_wrapper, args);
+    if (result != TSI_ASYNC) {
+      args->error = on_handshake_next_done(
+          result, args, bytes_to_send, bytes_to_send_size, handshaker_result);
+      if (args->error != GRPC_ERROR_NONE) {
+        return;
+      }
+    }
+  } while (result == TSI_INCOMPLETE_DATA);
   notification_wait(fixture);
 }
 
diff --git a/test/core/tsi/transport_security_test_lib.h b/test/core/tsi/transport_security_test_lib.h
index 1a04c1f074..49a7a59c58 100644
--- a/test/core/tsi/transport_security_test_lib.h
+++ b/test/core/tsi/transport_security_test_lib.h
@@ -21,6 +21,8 @@
 
 #include "src/core/tsi/transport_security_interface.h"
 
+#include <grpc/support/sync.h>
+
 #define TSI_TEST_TINY_HANDSHAKE_BUFFER_SIZE 32
 #define TSI_TEST_SMALL_HANDSHAKE_BUFFER_SIZE 128
 #define TSI_TEST_SMALL_READ_BUFFER_ALLOCATED_SIZE 41
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index af3bdb25ac..cf1cc7e486 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -461,7 +461,6 @@ TEST_P(AsyncEnd2endTest, ReconnectChannel) {
   if (GetParam().inproc) {
     return;
   }
-  gpr_setenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS", "200");
   int poller_slowdown_factor = 1;
   // It needs 2 pollset_works to reconnect the channel with polling engine
   // "poll"
@@ -2013,6 +2012,9 @@ INSTANTIATE_TEST_CASE_P(AsyncEnd2endServerTryCancel,
 }  // namespace grpc
 
 int main(int argc, char** argv) {
+  // Change the backup poll interval from 5s to 200ms to speed up the
+  // ReconnectChannel test
+  gpr_setenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS", "200");
   grpc_test_init(argc, argv);
   gpr_tls_init(&g_is_async_end2end_test);
   ::testing::InitGoogleTest(&argc, argv);
diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc
index f990a7ed9d..90b2eddbbb 100644
--- a/test/cpp/end2end/thread_stress_test.cc
+++ b/test/cpp/end2end/thread_stress_test.cc
@@ -50,23 +50,6 @@ const int kNumRpcs = 1000;  // Number of RPCs per thread
 namespace grpc {
 namespace testing {
 
-namespace {
-
-// When echo_deadline is requested, deadline seen in the ServerContext is set in
-// the response in seconds.
-void MaybeEchoDeadline(ServerContext* context, const EchoRequest* request,
-                       EchoResponse* response) {
-  if (request->has_param() && request->param().echo_deadline()) {
-    gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
-    if (context->deadline() != system_clock::time_point::max()) {
-      Timepoint2Timespec(context->deadline(), &deadline);
-    }
-    response->mutable_param()->set_request_deadline(deadline.tv_sec);
-  }
-}
-
-}  // namespace
-
 class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
  public:
   TestServiceImpl() : signal_client_(false) {}
@@ -74,29 +57,6 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service {
   Status Echo(ServerContext* context, const EchoRequest* request,
               EchoResponse* response) override {
     response->set_message(request->message());
-    MaybeEchoDeadline(context, request, response);
-    if (request->has_param() && request->param().client_cancel_after_us()) {
-      {
-        std::unique_lock<std::mutex> lock(mu_);
-        signal_client_ = true;
-      }
-      while (!context->IsCancelled()) {
-        gpr_sleep_until(gpr_time_add(
-            gpr_now(GPR_CLOCK_REALTIME),
-            gpr_time_from_micros(request->param().client_cancel_after_us(),
-                                 GPR_TIMESPAN)));
-      }
-      return Status::CANCELLED;
-    } else if (request->has_param() &&
-               request->param().server_cancel_after_us()) {
-      gpr_sleep_until(gpr_time_add(
-          gpr_now(GPR_CLOCK_REALTIME),
-          gpr_time_from_micros(request->param().server_cancel_after_us(),
-                               GPR_TIMESPAN)));
-      return Status::CANCELLED;
-    } else {
-      EXPECT_FALSE(context->IsCancelled());
-    }
     return Status::OK;
   }
 
diff --git a/test/cpp/microbenchmarks/fullstack_streaming_ping_pong.h b/test/cpp/microbenchmarks/fullstack_streaming_ping_pong.h
index 6df044f344..06d18b890d 100644
--- a/test/cpp/microbenchmarks/fullstack_streaming_ping_pong.h
+++ b/test/cpp/microbenchmarks/fullstack_streaming_ping_pong.h
@@ -278,7 +278,7 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
 
       void* t;
       bool ok;
-      int need_tags;
+      int expect_tags;
 
       // Send 'max_ping_pongs' number of ping pong messages
       int ping_pong_cnt = 0;
@@ -289,7 +289,7 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
           request_rw->Write(send_request, tag(2));  // Start client send
         }
 
-        need_tags = (1 << 2) | (1 << 3) | (1 << 4) | (1 << 5);
+        int await_tags = (1 << 2);
 
         if (ping_pong_cnt == 0) {
           // wait for the server call structure (call_hook, etc.) to be
@@ -301,8 +301,8 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
             // In some cases tag:2 comes before tag:0 (write tag comes out
             // first), this while loop is to make sure get tag:0.
             int i = (int)(intptr_t)t;
-            GPR_ASSERT(need_tags & (1 << i));
-            need_tags &= ~(1 << i);
+            GPR_ASSERT(await_tags & (1 << i));
+            await_tags &= ~(1 << i);
             GPR_ASSERT(fixture->cq()->Next(&t, &ok));
           }
         }
@@ -310,7 +310,11 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
         response_rw.Read(&recv_request, tag(3));   // Start server recv
         request_rw->Read(&recv_response, tag(4));  // Start client recv
 
-        while (need_tags) {
+        await_tags |= (1 << 3) | (1 << 4);
+        expect_tags = await_tags;
+        await_tags |= (1 << 5);
+
+        while (await_tags != 0) {
           GPR_ASSERT(fixture->cq()->Next(&t, &ok));
           GPR_ASSERT(ok);
           int i = (int)(intptr_t)t;
@@ -321,34 +325,39 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
               if (write_and_finish == 1) {
                 response_rw.WriteAndFinish(send_response, WriteOptions(),
                                            Status::OK, tag(5));
+                expect_tags |= (1 << 5);
               } else {
                 response_rw.WriteLast(send_response, WriteOptions(), tag(5));
-                // WriteLast buffers the write, so neither server write op nor
-                // client read op will finish inside the while loop.
-                need_tags &= ~(1 << 4);
-                need_tags &= ~(1 << 5);
+                // WriteLast buffers the write, so it's possible neither server
+                // write op nor client read op will finish inside the while
+                // loop.
+                await_tags &= ~(1 << 4);
+                await_tags &= ~(1 << 5);
+                expect_tags |= (1 << 5);
               }
             } else {
               response_rw.Write(send_response, tag(5));
+              expect_tags |= (1 << 5);
             }
           }
 
-          GPR_ASSERT(need_tags & (1 << i));
-          need_tags &= ~(1 << i);
+          GPR_ASSERT(expect_tags & (1 << i));
+          expect_tags &= ~(1 << i);
+          await_tags &= ~(1 << i);
         }
 
         ping_pong_cnt++;
       }
 
       if (max_ping_pongs == 0) {
-        need_tags = (1 << 6) | (1 << 7) | (1 << 8);
+        expect_tags |= (1 << 6) | (1 << 7) | (1 << 8);
       } else {
         if (write_and_finish == 1) {
-          need_tags = (1 << 8);
+          expect_tags |= (1 << 8);
         } else {
           // server's buffered write and the client's read of the buffered write
           // tags should come up.
-          need_tags = (1 << 4) | (1 << 5) | (1 << 7) | (1 << 8);
+          expect_tags |= (1 << 7) | (1 << 8);
         }
       }
 
@@ -360,8 +369,8 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
         GPR_ASSERT(fixture->cq()->Next(&t, &ok));
         while ((int)(intptr_t)t != 0) {
           int i = (int)(intptr_t)t;
-          GPR_ASSERT(need_tags & (1 << i));
-          need_tags &= ~(1 << i);
+          GPR_ASSERT(expect_tags & (1 << i));
+          expect_tags &= ~(1 << i);
           GPR_ASSERT(fixture->cq()->Next(&t, &ok));
         }
         response_rw.Finish(Status::OK, tag(7));
@@ -374,11 +383,11 @@ static void BM_StreamingPingPongWithCoalescingApi(benchmark::State& state) {
       Status recv_status;
       request_rw->Finish(&recv_status, tag(8));
 
-      while (need_tags) {
+      while (expect_tags) {
         GPR_ASSERT(fixture->cq()->Next(&t, &ok));
         int i = (int)(intptr_t)t;
-        GPR_ASSERT(need_tags & (1 << i));
-        need_tags &= ~(1 << i);
+        GPR_ASSERT(expect_tags & (1 << i));
+        expect_tags &= ~(1 << i);
       }
 
       GPR_ASSERT(recv_status.ok());