Merge pull request #4175 from yang-g/tcp_refactor

Refactor security connector and handshake

src/core/census/grpc_filter.c

@@ -60,9 +60,7 @@ typedef struct call_data {
   grpc_closure finish_recv;
 } call_data;
 
-typedef struct channel_data {
-  gpr_uint8 unused;
-} channel_data;
+typedef struct channel_data { gpr_uint8 unused; } channel_data;
 
 static void extract_and_annotate_method_tag(grpc_metadata_batch *md,
                                             call_data *calld,

src/core/channel/compress_filter.c

@@ -39,11 +39,11 @@
 #include <grpc/support/log.h>
 #include <grpc/support/slice_buffer.h>
 
-#include "src/core/channel/compress_filter.h"
 #include "src/core/channel/channel_args.h"
-#include "src/core/profiling/timers.h"
+#include "src/core/channel/compress_filter.h"
 #include "src/core/compression/algorithm_metadata.h"
 #include "src/core/compression/message_compress.h"
+#include "src/core/profiling/timers.h"
 #include "src/core/support/string.h"
 #include "src/core/transport/static_metadata.h"
 

src/core/channel/http_client_filter.c

@@ -31,12 +31,12 @@
  */
 
 #include "src/core/channel/http_client_filter.h"
-#include <string.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
-#include "src/core/support/string.h"
+#include <string.h>
 #include "src/core/profiling/timers.h"
+#include "src/core/support/string.h"
 #include "src/core/transport/static_metadata.h"
 
 typedef struct call_data {

src/core/channel/http_server_filter.c

@@ -33,9 +33,9 @@
 
 #include "src/core/channel/http_server_filter.h"
 
-#include <string.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
+#include <string.h>
 #include "src/core/profiling/timers.h"
 #include "src/core/transport/static_metadata.h"
 

src/core/httpcli/httpcli_security_connector.c

@@ -68,7 +68,7 @@ static void httpcli_ssl_do_handshake(grpc_exec_ctx *exec_ctx,
   tsi_result result = TSI_OK;
   tsi_handshaker *handshaker;
   if (c->handshaker_factory == NULL) {
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, nonsecure_endpoint, NULL);
+    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL);
     return;
   }
   result = tsi_ssl_handshaker_factory_create_handshaker(
@@ -76,7 +76,7 @@ static void httpcli_ssl_do_handshake(grpc_exec_ctx *exec_ctx,
   if (result != TSI_OK) {
     gpr_log(GPR_ERROR, "Handshaker creation failed with error %s.",
             tsi_result_to_string(result));
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, nonsecure_endpoint, NULL);
+    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL);
   } else {
     grpc_do_security_handshake(exec_ctx, handshaker, sc, nonsecure_endpoint, cb,
                                user_data);
@@ -149,7 +149,6 @@ typedef struct {
 
 static void on_secure_transport_setup_done(grpc_exec_ctx *exec_ctx, void *rp,
                                            grpc_security_status status,
-                                           grpc_endpoint *wrapped_endpoint,
                                            grpc_endpoint *secure_endpoint) {
   on_done_closure *c = rp;
   if (status != GRPC_SECURITY_OK) {

src/core/security/client_auth_filter.c

@@ -39,11 +39,11 @@
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 
-#include "src/core/support/string.h"
 #include "src/core/channel/channel_stack.h"
-#include "src/core/security/security_context.h"
-#include "src/core/security/security_connector.h"
 #include "src/core/security/credentials.h"
+#include "src/core/security/security_connector.h"
+#include "src/core/security/security_context.h"
+#include "src/core/support/string.h"
 #include "src/core/surface/call.h"
 #include "src/core/transport/static_metadata.h"
 

src/core/security/handshake.c

@@ -64,12 +64,39 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
 static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx, void *setup,
                                            int success);
 
+static void security_connector_remove_handshake(grpc_security_handshake *h) {
+  grpc_security_connector_handshake_list *node;
+  grpc_security_connector_handshake_list *tmp;
+  grpc_security_connector *sc = h->connector;
+  gpr_mu_lock(&sc->mu);
+  node = sc->handshaking_handshakes;
+  if (node && node->handshake == h) {
+    sc->handshaking_handshakes = node->next;
+    gpr_free(node);
+    gpr_mu_unlock(&sc->mu);
+    return;
+  }
+  while (node) {
+    if (node->next->handshake == h) {
+      tmp = node->next;
+      node->next = node->next->next;
+      gpr_free(tmp);
+      gpr_mu_unlock(&sc->mu);
+      return;
+    }
+    node = node->next;
+  }
+  gpr_mu_unlock(&sc->mu);
+}
+
 static void security_handshake_done(grpc_exec_ctx *exec_ctx,
                                     grpc_security_handshake *h,
                                     int is_success) {
+  if (!h->connector->is_client_side) {
+    security_connector_remove_handshake(h);
+  }
   if (is_success) {
-    h->cb(exec_ctx, h->user_data, GRPC_SECURITY_OK, h->wrapped_endpoint,
-          h->secure_endpoint);
+    h->cb(exec_ctx, h->user_data, GRPC_SECURITY_OK, h->secure_endpoint);
   } else {
     if (h->secure_endpoint != NULL) {
       grpc_endpoint_shutdown(exec_ctx, h->secure_endpoint);
@@ -77,8 +104,7 @@ static void security_handshake_done(grpc_exec_ctx *exec_ctx,
     } else {
       grpc_endpoint_destroy(exec_ctx, h->wrapped_endpoint);
     }
-    h->cb(exec_ctx, h->user_data, GRPC_SECURITY_ERROR, h->wrapped_endpoint,
-          NULL);
+    h->cb(exec_ctx, h->user_data, GRPC_SECURITY_ERROR, NULL);
   }
   if (h->handshaker != NULL) tsi_handshaker_destroy(h->handshaker);
   if (h->handshake_buffer != NULL) gpr_free(h->handshake_buffer);
@@ -268,6 +294,7 @@ void grpc_do_security_handshake(grpc_exec_ctx *exec_ctx,
                                 grpc_endpoint *nonsecure_endpoint,
                                 grpc_security_handshake_done_cb cb,
                                 void *user_data) {
+  grpc_security_connector_handshake_list *handshake_node;
   grpc_security_handshake *h = gpr_malloc(sizeof(grpc_security_handshake));
   memset(h, 0, sizeof(grpc_security_handshake));
   h->handshaker = handshaker;
@@ -284,5 +311,19 @@ void grpc_do_security_handshake(grpc_exec_ctx *exec_ctx,
   gpr_slice_buffer_init(&h->left_overs);
   gpr_slice_buffer_init(&h->outgoing);
   gpr_slice_buffer_init(&h->incoming);
+  if (!connector->is_client_side) {
+    handshake_node = gpr_malloc(sizeof(grpc_security_connector_handshake_list));
+    handshake_node->handshake = h;
+    gpr_mu_lock(&connector->mu);
+    handshake_node->next = connector->handshaking_handshakes;
+    connector->handshaking_handshakes = handshake_node;
+    gpr_mu_unlock(&connector->mu);
+  }
   send_handshake_bytes_to_peer(exec_ctx, h);
 }
+
+void grpc_security_handshake_shutdown(grpc_exec_ctx *exec_ctx,
+                                      void *handshake) {
+  grpc_security_handshake *h = handshake;
+  grpc_endpoint_shutdown(exec_ctx, h->wrapped_endpoint);
+}

src/core/security/handshake.h

@@ -45,4 +45,6 @@ void grpc_do_security_handshake(grpc_exec_ctx *exec_ctx,
                                 grpc_security_handshake_done_cb cb,
                                 void *user_data);
 
+void grpc_security_handshake_shutdown(grpc_exec_ctx *exec_ctx, void *handshake);
+
 #endif /* GRPC_INTERNAL_CORE_SECURITY_HANDSHAKE_H */

src/core/security/security_connector.c

@@ -102,13 +102,29 @@ const tsi_peer_property *tsi_peer_get_property_by_name(const tsi_peer *peer,
   return NULL;
 }
 
+void grpc_security_connector_shutdown(grpc_exec_ctx *exec_ctx,
+                                      grpc_security_connector *connector) {
+  grpc_security_connector_handshake_list *tmp;
+  if (!connector->is_client_side) {
+    gpr_mu_lock(&connector->mu);
+    while (connector->handshaking_handshakes) {
+      tmp = connector->handshaking_handshakes;
+      grpc_security_handshake_shutdown(
+          exec_ctx, connector->handshaking_handshakes->handshake);
+      connector->handshaking_handshakes = tmp->next;
+      gpr_free(tmp);
+    }
+    gpr_mu_unlock(&connector->mu);
+  }
+}
+
 void grpc_security_connector_do_handshake(grpc_exec_ctx *exec_ctx,
                                           grpc_security_connector *sc,
                                           grpc_endpoint *nonsecure_endpoint,
                                           grpc_security_handshake_done_cb cb,
                                           void *user_data) {
   if (sc == NULL || nonsecure_endpoint == NULL) {
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, nonsecure_endpoint, NULL);
+    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL);
   } else {
     sc->vtable->do_handshake(exec_ctx, sc, nonsecure_endpoint, cb, user_data);
   }
@@ -219,6 +235,7 @@ static void fake_channel_destroy(grpc_security_connector *sc) {
 
 static void fake_server_destroy(grpc_security_connector *sc) {
   GRPC_AUTH_CONTEXT_UNREF(sc->auth_context, "connector");
+  gpr_mu_destroy(&sc->mu);
   gpr_free(sc);
 }
 
@@ -319,6 +336,7 @@ grpc_security_connector *grpc_fake_server_security_connector_create(void) {
   c->is_client_side = 0;
   c->vtable = &fake_server_vtable;
   c->url_scheme = GRPC_FAKE_SECURITY_URL_SCHEME;
+  gpr_mu_init(&c->mu);
   return c;
 }
 
@@ -354,10 +372,12 @@ static void ssl_channel_destroy(grpc_security_connector *sc) {
 static void ssl_server_destroy(grpc_security_connector *sc) {
   grpc_ssl_server_security_connector *c =
       (grpc_ssl_server_security_connector *)sc;
+
   if (c->handshaker_factory != NULL) {
     tsi_ssl_handshaker_factory_destroy(c->handshaker_factory);
   }
   GRPC_AUTH_CONTEXT_UNREF(sc->auth_context, "connector");
+  gpr_mu_destroy(&sc->mu);
   gpr_free(sc);
 }
 
@@ -390,7 +410,7 @@ static void ssl_channel_do_handshake(grpc_exec_ctx *exec_ctx,
                                         : c->target_name,
       &handshaker);
   if (status != GRPC_SECURITY_OK) {
-    cb(exec_ctx, user_data, status, nonsecure_endpoint, NULL);
+    cb(exec_ctx, user_data, status, NULL);
   } else {
     grpc_do_security_handshake(exec_ctx, handshaker, sc, nonsecure_endpoint, cb,
                                user_data);
@@ -408,7 +428,7 @@ static void ssl_server_do_handshake(grpc_exec_ctx *exec_ctx,
   grpc_security_status status =
       ssl_create_handshaker(c->handshaker_factory, 0, NULL, &handshaker);
   if (status != GRPC_SECURITY_OK) {
-    cb(exec_ctx, user_data, status, nonsecure_endpoint, NULL);
+    cb(exec_ctx, user_data, status, NULL);
   } else {
     grpc_do_security_handshake(exec_ctx, handshaker, sc, nonsecure_endpoint, cb,
                                user_data);
@@ -691,6 +711,7 @@ grpc_security_status grpc_ssl_server_security_connector_create(
     *sc = NULL;
     goto error;
   }
+  gpr_mu_init(&c->base.mu);
   *sc = &c->base;
   gpr_free((void *)alpn_protocol_strings);
   gpr_free(alpn_protocol_string_lengths);

src/core/security/security_connector.h

@@ -67,7 +67,6 @@ typedef void (*grpc_security_check_cb)(grpc_exec_ctx *exec_ctx, void *user_data,
 typedef void (*grpc_security_handshake_done_cb)(grpc_exec_ctx *exec_ctx,
                                                 void *user_data,
                                                 grpc_security_status status,
-                                                grpc_endpoint *wrapped_endpoint,
                                                 grpc_endpoint *secure_endpoint);
 
 typedef struct {
@@ -80,13 +79,22 @@ typedef struct {
                                      void *user_data);
 } grpc_security_connector_vtable;
 
+typedef struct grpc_security_connector_handshake_list {
+  void *handshake;
+  struct grpc_security_connector_handshake_list *next;
+} grpc_security_connector_handshake_list;
+
 struct grpc_security_connector {
   const grpc_security_connector_vtable *vtable;
   gpr_refcount refcount;
   int is_client_side;
   const char *url_scheme;
   grpc_auth_context *auth_context; /* Populated after the peer is checked. */
-  const grpc_channel_args *channel_args; /* Server side only. */
+  /* Used on server side only. */
+  /* TODO(yangg) maybe create a grpc_server_security_connector with these */
+  gpr_mu mu;
+  grpc_security_connector_handshake_list *handshaking_handshakes;
+  const grpc_channel_args *channel_args;
 };
 
 /* Refcounting. */
@@ -127,6 +135,9 @@ grpc_security_status grpc_security_connector_check_peer(
     grpc_security_connector *sc, tsi_peer peer, grpc_security_check_cb cb,
     void *user_data);
 
+void grpc_security_connector_shutdown(grpc_exec_ctx *exec_ctx,
+                                      grpc_security_connector *connector);
+
 /* Util to encapsulate the connector in a channel arg. */
 grpc_arg grpc_security_connector_to_arg(grpc_security_connector *sc);
 

src/core/security/server_secure_chttp2.c

@@ -52,17 +52,11 @@
 #include <grpc/support/sync.h>
 #include <grpc/support/useful.h>
 
-typedef struct tcp_endpoint_list {
-  grpc_endpoint *tcp_endpoint;
-  struct tcp_endpoint_list *next;
-} tcp_endpoint_list;
-
 typedef struct grpc_server_secure_state {
   grpc_server *server;
   grpc_tcp_server *tcp;
   grpc_security_connector *sc;
   grpc_server_credentials *creds;
-  tcp_endpoint_list *handshaking_tcp_endpoints;
   int is_shutdown;
   gpr_mu mu;
   gpr_refcount refcount;
@@ -103,52 +97,28 @@ static void setup_transport(grpc_exec_ctx *exec_ctx, void *statep,
   grpc_channel_args_destroy(args_copy);
 }
 
-static int remove_tcp_from_list_locked(grpc_server_secure_state *state,
-                                       grpc_endpoint *tcp) {
-  tcp_endpoint_list *node = state->handshaking_tcp_endpoints;
-  tcp_endpoint_list *tmp = NULL;
-  if (node && node->tcp_endpoint == tcp) {
-    state->handshaking_tcp_endpoints = state->handshaking_tcp_endpoints->next;
-    gpr_free(node);
-    return 0;
-  }
-  while (node) {
-    if (node->next->tcp_endpoint == tcp) {
-      tmp = node->next;
-      node->next = node->next->next;
-      gpr_free(tmp);
-      return 0;
-    }
-    node = node->next;
-  }
-  return -1;
-}
-
 static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep,
                                      grpc_security_status status,
-                                     grpc_endpoint *wrapped_endpoint,
                                      grpc_endpoint *secure_endpoint) {
   grpc_server_secure_state *state = statep;
   grpc_transport *transport;
   if (status == GRPC_SECURITY_OK) {
-    gpr_mu_lock(&state->mu);
-    remove_tcp_from_list_locked(state, wrapped_endpoint);
-    if (!state->is_shutdown) {
-      transport = grpc_create_chttp2_transport(
-          exec_ctx, grpc_server_get_channel_args(state->server),
-          secure_endpoint, 0);
-      setup_transport(exec_ctx, state, transport);
-      grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL, 0);
-    } else {
-      /* We need to consume this here, because the server may already have gone
-       * away. */
-      grpc_endpoint_destroy(exec_ctx, secure_endpoint);
+    if (secure_endpoint) {
+      gpr_mu_lock(&state->mu);
+      if (!state->is_shutdown) {
+        transport = grpc_create_chttp2_transport(
+            exec_ctx, grpc_server_get_channel_args(state->server),
+            secure_endpoint, 0);
+        setup_transport(exec_ctx, state, transport);
+        grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL, 0);
+      } else {
+        /* We need to consume this here, because the server may already have
+         * gone away. */
+        grpc_endpoint_destroy(exec_ctx, secure_endpoint);
+      }
+      gpr_mu_unlock(&state->mu);
     }
-    gpr_mu_unlock(&state->mu);
   } else {
-    gpr_mu_lock(&state->mu);
-    remove_tcp_from_list_locked(state, wrapped_endpoint);
-    gpr_mu_unlock(&state->mu);
     gpr_log(GPR_ERROR, "Secure transport failed with error %d", status);
   }
   state_unref(state);
@@ -157,14 +127,7 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep,
 static void on_accept(grpc_exec_ctx *exec_ctx, void *statep,
                       grpc_endpoint *tcp) {
   grpc_server_secure_state *state = statep;
-  tcp_endpoint_list *node;
   state_ref(state);
-  node = gpr_malloc(sizeof(tcp_endpoint_list));
-  node->tcp_endpoint = tcp;
-  gpr_mu_lock(&state->mu);
-  node->next = state->handshaking_tcp_endpoints;
-  state->handshaking_tcp_endpoints = node;
-  gpr_mu_unlock(&state->mu);
   grpc_security_connector_do_handshake(exec_ctx, state->sc, tcp,
                                        on_secure_handshake_done, state);
 }
@@ -181,14 +144,7 @@ static void destroy_done(grpc_exec_ctx *exec_ctx, void *statep, int success) {
   grpc_server_secure_state *state = statep;
   state->destroy_callback->cb(exec_ctx, state->destroy_callback->cb_arg,
                               success);
-  gpr_mu_lock(&state->mu);
-  while (state->handshaking_tcp_endpoints != NULL) {
-    grpc_endpoint_shutdown(exec_ctx,
-                           state->handshaking_tcp_endpoints->tcp_endpoint);
-    remove_tcp_from_list_locked(state,
-                                state->handshaking_tcp_endpoints->tcp_endpoint);
-  }
-  gpr_mu_unlock(&state->mu);
+  grpc_security_connector_shutdown(exec_ctx, state->sc);
   state_unref(state);
 }
 
@@ -281,7 +237,6 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,
   state->sc = sc;
   state->creds = grpc_server_credentials_ref(creds);
 
-  state->handshaking_tcp_endpoints = NULL;
   state->is_shutdown = 0;
   gpr_mu_init(&state->mu);
   gpr_ref_init(&state->refcount, 1);

src/core/surface/channel_create.c

@@ -99,8 +99,8 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
       grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer,
                           &c->initial_string_sent);
     }
-    c->result->transport = grpc_create_chttp2_transport(
-        exec_ctx, c->args.channel_args, tcp, 1);
+    c->result->transport =
+        grpc_create_chttp2_transport(exec_ctx, c->args.channel_args, tcp, 1);
     grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, NULL,
                                         0);
     GPR_ASSERT(c->result->transport);

src/core/surface/lame_client.c

@@ -103,8 +103,7 @@ static void lame_start_transport_op(grpc_exec_ctx *exec_ctx,
 }
 
 static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-                           grpc_call_element_args *args) {
-}
+                           grpc_call_element_args *args) {}
 
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
                               grpc_call_element *elem) {}

src/core/surface/secure_channel_create.c

@@ -88,7 +88,6 @@ static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
 
 static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
                                      grpc_security_status status,
-                                     grpc_endpoint *wrapped_endpoint,
                                      grpc_endpoint *secure_endpoint) {
   connector *c = arg;
   grpc_closure *notify;
@@ -97,13 +96,11 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
     memset(c->result, 0, sizeof(*c->result));
     gpr_mu_unlock(&c->mu);
   } else if (status != GRPC_SECURITY_OK) {
-    GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint);
     gpr_log(GPR_ERROR, "Secure handshake failed with error %d.", status);
     memset(c->result, 0, sizeof(*c->result));
     c->connecting_endpoint = NULL;
     gpr_mu_unlock(&c->mu);
   } else {
-    GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint);
     c->connecting_endpoint = NULL;
     gpr_mu_unlock(&c->mu);
     c->result->transport = grpc_create_chttp2_transport(

src/core/transport/metadata.c

@@ -225,7 +225,8 @@ void grpc_mdctx_global_shutdown(void) {
     gc_mdtab(shard);
     /* TODO(ctiller): GPR_ASSERT(shard->count == 0); */
     if (shard->count != 0) {
-      gpr_log(GPR_DEBUG, "WARNING: %d metadata elements were leaked", shard->count);
+      gpr_log(GPR_DEBUG, "WARNING: %d metadata elements were leaked",
+              shard->count);
     }
     gpr_free(shard->elems);
   }
@@ -234,7 +235,8 @@ void grpc_mdctx_global_shutdown(void) {
     gpr_mu_destroy(&shard->mu);
     /* TODO(ctiller): GPR_ASSERT(shard->count == 0); */
     if (shard->count != 0) {
-      gpr_log(GPR_DEBUG, "WARNING: %d metadata strings were leaked", shard->count);
+      gpr_log(GPR_DEBUG, "WARNING: %d metadata strings were leaked",
+              shard->count);
     }
     gpr_free(shard->strs);
   }

src/cpp/client/secure_credentials.cc

@@ -31,10 +31,10 @@
  *
  */
 
-#include <grpc/support/log.h>
 #include <grpc++/channel.h>
 #include <grpc++/impl/grpc_library.h>
 #include <grpc++/support/channel_arguments.h>
+#include <grpc/support/log.h>
 #include "src/cpp/client/create_channel_internal.h"
 #include "src/cpp/client/secure_credentials.h"
 #include "src/cpp/common/secure_auth_context.h"

src/node/ext/call.cc

@@ -234,9 +234,7 @@ class SendMetadataOp : public Op {
 
 class SendMessageOp : public Op {
  public:
-  SendMessageOp() {
-    send_message = NULL;
-  }
+  SendMessageOp() { send_message = NULL; }
   ~SendMessageOp() {
     if (send_message != NULL) {
       grpc_byte_buffer_destroy(send_message);
@@ -271,6 +269,7 @@ class SendMessageOp : public Op {
   std::string GetTypeString() const {
     return "send_message";
   }
+
  private:
   grpc_byte_buffer *send_message;
 };

test/core/client_config/lb_policies_test.c

@@ -135,8 +135,9 @@ static void kill_server(const servers_fixture *f, size_t i) {
   gpr_log(GPR_INFO, "KILLING SERVER %d", i);
   GPR_ASSERT(f->servers[i] != NULL);
   grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000));
-  GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000),
-                                         NULL).type == GRPC_OP_COMPLETE);
+  GPR_ASSERT(
+      grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000), NULL)
+          .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->servers[i]);
   f->servers[i] = NULL;
 }
@@ -202,8 +203,8 @@ static void teardown_servers(servers_fixture *f) {
     if (f->servers[i] == NULL) continue;
     grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000));
     GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000),
-                                           n_millis_time(5000),
-                                           NULL).type == GRPC_OP_COMPLETE);
+                                           n_millis_time(5000), NULL)
+                   .type == GRPC_OP_COMPLETE);
     grpc_server_destroy(f->servers[i]);
   }
   grpc_completion_queue_shutdown(f->cq);
@@ -303,8 +304,8 @@ int *perform_request(servers_fixture *f, grpc_channel *client,
 
     s_idx = -1;
     while ((ev = grpc_completion_queue_next(
-                f->cq, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1), NULL)).type !=
-           GRPC_QUEUE_TIMEOUT) {
+                f->cq, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1), NULL))
+               .type != GRPC_QUEUE_TIMEOUT) {
       GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
       read_tag = ((int)(gpr_intptr)ev.tag);
       gpr_log(GPR_DEBUG, "EVENT: success:%d, type:%d, tag:%d iter:%d",

test/core/compression/message_compress_test.c

@@ -40,9 +40,9 @@
 #include <grpc/support/log.h>
 #include <grpc/support/useful.h>
 
-#include "test/core/util/test_config.h"
 #include "src/core/support/murmur_hash.h"
 #include "test/core/util/slice_splitter.h"
+#include "test/core/util/test_config.h"
 
 typedef enum { ONE_A = 0, ONE_KB_A, ONE_MB_A, TEST_VALUE_COUNT } test_value;
 

test/core/end2end/tests/hpack_size.c

@@ -262,9 +262,9 @@ static void drain_cq(grpc_completion_queue *cq) {
 static void shutdown_server(grpc_end2end_test_fixture *f) {
   if (!f->server) return;
   grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));
-  GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(1000),
-                                         GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5),
-                                         NULL).type == GRPC_OP_COMPLETE);
+  GPR_ASSERT(grpc_completion_queue_pluck(
+                 f->cq, tag(1000), GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5), NULL)
+                 .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(f->server);
   f->server = NULL;
 }

test/core/fling/server.c

@@ -50,9 +50,9 @@
 #include <grpc/support/log.h>
 #include <grpc/support/time.h>
 #include "src/core/profiling/timers.h"
-#include "test/core/util/port.h"
 #include "test/core/end2end/data/ssl_test_data.h"
 #include "test/core/util/grpc_profiler.h"
+#include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
 
 static grpc_completion_queue *cq;

test/core/transport/chttp2/hpack_table_test.c

@@ -36,10 +36,10 @@
 #include <string.h>
 #include <stdio.h>
 
+#include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
-#include <grpc/grpc.h>
 
 #include "src/core/support/string.h"
 #include "test/core/util/test_config.h"

test/proto/metrics.proto

@@ -43,7 +43,9 @@ message GaugeResponse {
   }
 }
 
-message GaugeRequest { string name = 1; }
+message GaugeRequest {
+  string name = 1;
+}
 
 message EmptyMessage {
 }