Dynamically enable/disable packet coalecsing and test it

2 files changed, 40 insertions, 38 deletions

diff --git a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
index 477cf07f45..2e40020ae0 100644
--- a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
+++ b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.c
@@ -51,6 +51,8 @@ typedef struct cronet_transport {
 
 extern grpc_transport_vtable grpc_cronet_vtable;
 
+bool grpc_cronet_packet_coalescing_enabled = true;
+
 GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
     void *engine, const char *target, const grpc_channel_args *args,
     void *reserved) {
@@ -67,3 +69,7 @@ GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
   return grpc_channel_create(&exec_ctx, target, args,
                              GRPC_CLIENT_DIRECT_CHANNEL, (grpc_transport *)ct);
 }
+
+GRPCAPI void grpc_cronet_use_packet_coalescing(bool use_coalescing) {
+  grpc_cronet_packet_coalescing_enabled = use_coalescing;
+}
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index 447f3f31ec..5429eb32e3 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -61,6 +61,8 @@
 /* TODO (makdharma): Hook up into the wider tracing mechanism */
 int grpc_cronet_trace = 0;
 
+extern bool grpc_cronet_packet_coalescing_enabled;
+
 enum e_op_result {
   ACTION_TAKEN_WITH_CALLBACK,
   ACTION_TAKEN_NO_CALLBACK,
@@ -150,12 +152,13 @@ struct op_state {
   bool state_callback_received[OP_NUM_OPS];
   bool fail_state;
   bool flush_read;
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
   bool flush_cronet_when_ready;
   bool pending_write_for_trailer;
-#endif
   bool unprocessed_send_message;
   grpc_error *cancel_error;
+
+  /* Whether packet coalescing is enabled */
+  bool packet_coalescing_enabled;
   /* data structure for storing data coming from server */
   struct read_state rs;
   /* data structure for storing data going to the server */
@@ -425,12 +428,10 @@ static void on_stream_ready(bidirectional_stream *stream) {
   }
 /* Send the initial metadata on wire if there is no SEND_MESSAGE or
  * SEND_TRAILING_METADATA ops pending */
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-  if (s->state.flush_cronet_when_ready) {
+  if (s->state.packet_coalescing_enabled && s->state.flush_cronet_when_ready) {
     CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
     bidirectional_stream_flush(stream);
   }
-#endif
   gpr_mu_unlock(&s->mu);
   execute_from_storage(s);
 }
@@ -568,10 +569,10 @@ static void on_response_trailers_received(
     CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
     s->state.state_callback_received[OP_SEND_MESSAGE] = false;
     bidirectional_stream_write(s->cbs, "", 0, true);
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-    CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
-    bidirectional_stream_flush(s->cbs);
-#endif
+    if (s->state.packet_coalescing_enabled) {
+      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+      bidirectional_stream_flush(s->cbs);
+    }
     s->state.state_op_done[OP_SEND_TRAILING_METADATA] = true;
 
     gpr_mu_unlock(&s->mu);
@@ -768,11 +769,9 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
       result = false;
     /* we haven't got on_write_completed for the send yet */
     else if (stream_state->state_op_done[OP_SEND_MESSAGE] &&
-             !stream_state->state_callback_received[OP_SEND_MESSAGE]
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-             && !stream_state->pending_write_for_trailer
-#endif
-             )
+             !stream_state->state_callback_received[OP_SEND_MESSAGE] &&
+             !(stream_state->packet_coalescing_enabled &&
+               stream_state->pending_write_for_trailer))
       result = false;
   } else if (op_id == OP_CANCEL_ERROR) {
     /* already executed */
@@ -858,10 +857,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
     s->cbs = bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
                                          &cronet_callbacks);
     CRONET_LOG(GPR_DEBUG, "%p = bidirectional_stream_create()", s->cbs);
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-    bidirectional_stream_disable_auto_flush(s->cbs, true);
-    bidirectional_stream_delay_request_headers_until_flush(s->cbs, true);
-#endif
+    if (stream_state->packet_coalescing_enabled) {
+      bidirectional_stream_disable_auto_flush(s->cbs, true);
+      bidirectional_stream_delay_request_headers_until_flush(s->cbs, true);
+    }
     char *url = NULL;
     const char *method = "POST";
     s->header_array.headers = NULL;
@@ -872,11 +871,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
     CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url);
     bidirectional_stream_start(s->cbs, url, 0, method, &s->header_array, false);
     stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-    if (!stream_op->send_message && !stream_op->send_trailing_metadata) {
+    if (stream_state->packet_coalescing_enabled && !stream_op->send_message &&
+        !stream_op->send_trailing_metadata) {
       s->state.flush_cronet_when_ready = true;
     }
-#endif
     result = ACTION_TAKEN_WITH_CALLBACK;
   } else if (stream_op->send_message &&
              op_can_be_run(stream_op, stream_state, &oas->state,
@@ -913,19 +911,18 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
         stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
         bidirectional_stream_write(s->cbs, stream_state->ws.write_buffer,
                                    (int)write_buffer_size, false);
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-        if (!stream_op->send_trailing_metadata) {
-          CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)",
-                     s->cbs);
-          bidirectional_stream_flush(s->cbs);
-          result = ACTION_TAKEN_WITH_CALLBACK;
+        if (stream_state->packet_coalescing_enabled) {
+          if (!stream_op->send_trailing_metadata) {
+            CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+            bidirectional_stream_flush(s->cbs);
+            result = ACTION_TAKEN_WITH_CALLBACK;
+          } else {
+            stream_state->pending_write_for_trailer = true;
+            result = ACTION_TAKEN_NO_CALLBACK;
+          }
         } else {
-          stream_state->pending_write_for_trailer = true;
-          result = ACTION_TAKEN_NO_CALLBACK;
+          result = ACTION_TAKEN_WITH_CALLBACK;
         }
-#else
-        result = ACTION_TAKEN_WITH_CALLBACK;
-#endif
       } else {
         result = NO_ACTION_POSSIBLE;
       }
@@ -944,10 +941,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
                  s->cbs);
       stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
       bidirectional_stream_write(s->cbs, "", 0, true);
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
-      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
-      bidirectional_stream_flush(s->cbs);
-#endif
+      if (stream_state->packet_coalescing_enabled) {
+        CRONET_LOG(GPR_DEBUG, "bidirectional_stream_flush (%p)", s->cbs);
+        bidirectional_stream_flush(s->cbs);
+      }
       result = ACTION_TAKEN_WITH_CALLBACK;
     }
     stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
@@ -1176,10 +1173,9 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
          sizeof(s->state.state_callback_received));
   s->state.fail_state = s->state.flush_read = false;
   s->state.cancel_error = NULL;
-#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
   s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false;
-#endif
   s->state.unprocessed_send_message = false;
+  s->state.packet_coalescing_enabled = grpc_cronet_packet_coalescing_enabled;
   gpr_mu_init(&s->mu);
   return 0;
 }