Merge github.com:grpc/grpc into bm_trickle

.pylintrc

@@ -0,0 +1,38 @@
+[MESSAGES CONTROL]
+
+#TODO: Enable missing-docstring
+#TODO: Enable too-few-public-methods
+#TODO: Enable too-many-arguments
+#TODO: Enable no-init
+#TODO: Enable duplicate-code
+#TODO: Enable invalid-name
+#TODO: Enable suppressed-message
+#TODO: Enable locally-disabled
+#TODO: Enable protected-access
+#TODO: Enable no-name-in-module
+#TODO: Enable unused-argument
+#TODO: Enable fixme
+#TODO: Enable wrong-import-order
+#TODO: Enable no-value-for-parameter
+#TODO: Enable cyclic-import
+#TODO: Enable unused-variable
+#TODO: Enable redefined-outer-name
+#TODO: Enable unused-import
+#TODO: Enable too-many-instance-attributes
+#TODO: Enable broad-except
+#TODO: Enable too-many-locals
+#TODO: Enable too-many-lines
+#TODO: Enable redefined-variable-type
+#TODO: Enable next-method-called
+#TODO: Enable import-error
+#TODO: Enable useless-else-on-loop
+#TODO: Enable too-many-return-statements
+#TODO: Enable too-many-nested-blocks
+#TODO: Enable super-init-not-called
+#TODO: Enable simplifiable-if-statement
+#TODO: Enable no-self-use
+#TODO: Enable no-member
+#TODO: Enable logging-format-interpolation
+#TODO: Enable dangerous-default-value
+
+disable=missing-docstring,too-few-public-methods,too-many-arguments,no-init,duplicate-code,invalid-name,suppressed-message,locally-disabled,protected-access,no-name-in-module,unused-argument,fixme,wrong-import-order,no-value-for-parameter,cyclic-import,unused-variable,redefined-outer-name,unused-import,too-many-instance-attributes,broad-except,too-many-locals,too-many-lines,redefined-variable-type,next-method-called,import-error,useless-else-on-loop,too-many-return-statements,too-many-nested-blocks,super-init-not-called,simplifiable-if-statement,no-self-use,no-member,logging-format-interpolation,dangerous-default-value

BUILD

@@ -1076,6 +1076,7 @@ grpc_cc_library(
     ],
     hdrs = [
         "third_party/objective_c/Cronet/bidirectional_stream_c.h",
+        "src/core/ext/transport/cronet/transport/cronet_transport.h",
     ],
     language = "c",
     public_hdrs = [

build.yaml

@@ -726,6 +726,7 @@ filegroups:
   - include/grpc/grpc_security.h
   - include/grpc/grpc_security_constants.h
   headers:
+  - src/core/ext/transport/cronet/transport/cronet_transport.h
   - third_party/objective_c/Cronet/bidirectional_stream_c.h
   src:
   - src/core/ext/transport/cronet/client/secure/cronet_channel_create.c

gRPC-Core.podspec

@@ -884,7 +884,8 @@ Pod::Spec.new do |s|
 
   s.subspec 'Cronet-Interface' do |ss|
     ss.header_mappings_dir = 'include/grpc'
-    ss.source_files = 'include/grpc/grpc_cronet.h'
+    ss.source_files = 'include/grpc/grpc_cronet.h',
+                      'src/core/ext/transport/cronet/transport/cronet_transport.h'
   end
 
   s.subspec 'Cronet-Implementation' do |ss|

include/grpc/impl/codegen/grpc_types.h

@@ -230,6 +230,10 @@ typedef struct {
 #define GRPC_ARG_LB_POLICY_NAME "grpc.lb_policy_name"
 /** The grpc_socket_mutator instance that set the socket options. A pointer. */
 #define GRPC_ARG_SOCKET_MUTATOR "grpc.socket_mutator"
+/** If non-zero, Cronet transport will coalesce packets to fewer frames when
+ * possible. */
+#define GRPC_ARG_USE_CRONET_PACKET_COALESCING \
+  "grpc.use_cronet_packet_coalescing"
 /** \} */
 
 /** Result of a grpc call. If the caller satisfies the prerequisites of a

src/core/ext/transport/cronet/client/secure/cronet_channel_create.c

@@ -39,6 +39,7 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
+#include "src/core/ext/transport/cronet/transport/cronet_transport.h"
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/transport/transport_impl.h"
 
@@ -54,16 +55,14 @@ extern grpc_transport_vtable grpc_cronet_vtable;
 GRPCAPI grpc_channel *grpc_cronet_secure_channel_create(
     void *engine, const char *target, const grpc_channel_args *args,
     void *reserved) {
-  cronet_transport *ct = gpr_malloc(sizeof(cronet_transport));
-  ct->base.vtable = &grpc_cronet_vtable;
-  ct->engine = engine;
-  ct->host = gpr_malloc(strlen(target) + 1);
-  strcpy(ct->host, target);
   gpr_log(GPR_DEBUG,
           "grpc_create_cronet_transport: stream_engine = %p, target=%s", engine,
-          ct->host);
+          target);
+
+  grpc_transport *ct =
+      grpc_create_cronet_transport(engine, target, args, reserved);
 
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   return grpc_channel_create(&exec_ctx, target, args,
-                             GRPC_CLIENT_DIRECT_CHANNEL, (grpc_transport *)ct);
+                             GRPC_CLIENT_DIRECT_CHANNEL, ct);
 }

src/core/ext/transport/cronet/transport/cronet_api_dummy.c

@@ -80,4 +80,16 @@ void bidirectional_stream_cancel(bidirectional_stream* stream) {
   GPR_ASSERT(0);
 }
 
+void bidirectional_stream_disable_auto_flush(bidirectional_stream* stream,
+                                             bool disable_auto_flush) {
+  GPR_ASSERT(0);
+}
+
+void bidirectional_stream_delay_request_headers_until_flush(
+    bidirectional_stream* stream, bool delay_headers_until_flush) {
+  GPR_ASSERT(0);
+}
+
+void bidirectional_stream_flush(bidirectional_stream* stream) { GPR_ASSERT(0); }
+
 #endif /* GRPC_COMPILE_WITH_CRONET */

src/core/ext/transport/cronet/transport/cronet_transport.c

@@ -88,7 +88,7 @@ enum e_op_id {
 
 /* Cronet callbacks. See cronet_c_for_grpc.h for documentation for each. */
 
-static void on_request_headers_sent(bidirectional_stream *);
+static void on_stream_ready(bidirectional_stream *);
 static void on_response_headers_received(
     bidirectional_stream *, const bidirectional_stream_header_array *,
     const char *);
@@ -100,7 +100,7 @@ static void on_succeeded(bidirectional_stream *);
 static void on_failed(bidirectional_stream *, int);
 static void on_canceled(bidirectional_stream *);
 static bidirectional_stream_callback cronet_callbacks = {
-    on_request_headers_sent,
+    on_stream_ready,
     on_response_headers_received,
     on_read_completed,
     on_write_completed,
@@ -114,6 +114,7 @@ struct grpc_cronet_transport {
   grpc_transport base; /* must be first element in this structure */
   stream_engine *engine;
   char *host;
+  bool use_packet_coalescing;
 };
 typedef struct grpc_cronet_transport grpc_cronet_transport;
 
@@ -152,6 +153,9 @@ struct op_state {
   bool state_callback_received[OP_NUM_OPS];
   bool fail_state;
   bool flush_read;
+  bool flush_cronet_when_ready;
+  bool pending_write_for_trailer;
+  bool unprocessed_send_message;
   grpc_error *cancel_error;
   /* data structure for storing data coming from server */
   struct read_state rs;
@@ -175,7 +179,7 @@ struct op_storage {
 struct stream_obj {
   struct op_and_state *oas;
   grpc_transport_stream_op *curr_op;
-  grpc_cronet_transport curr_ct;
+  grpc_cronet_transport *curr_ct;
   grpc_stream *curr_gs;
   bidirectional_stream *cbs;
   bidirectional_stream_header_array header_array;
@@ -274,6 +278,9 @@ static void add_to_storage(struct stream_obj *s, grpc_transport_stream_op *op) {
   new_op->next = storage->head;
   storage->head = new_op;
   storage->num_pending_ops++;
+  if (op->send_message) {
+    s->state.unprocessed_send_message = true;
+  }
   CRONET_LOG(GPR_DEBUG, "adding new op %p. %d in the queue.", new_op,
              storage->num_pending_ops);
   gpr_mu_unlock(&s->mu);
@@ -406,9 +413,10 @@ static void on_succeeded(bidirectional_stream *stream) {
 /*
   Cronet callback
 */
-static void on_request_headers_sent(bidirectional_stream *stream) {
-  CRONET_LOG(GPR_DEBUG, "W: on_request_headers_sent(%p)", stream);
+static void on_stream_ready(bidirectional_stream *stream) {
+  CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream);
   stream_obj *s = (stream_obj *)stream->annotation;
+  grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   gpr_mu_lock(&s->mu);
   s->state.state_op_done[OP_SEND_INITIAL_METADATA] = true;
   s->state.state_callback_received[OP_SEND_INITIAL_METADATA] = true;
@@ -417,6 +425,14 @@ static void on_request_headers_sent(bidirectional_stream *stream) {
     gpr_free(s->header_array.headers);
     s->header_array.headers = NULL;
   }
+  /* Send the initial metadata on wire if there is no SEND_MESSAGE or
+   * SEND_TRAILING_METADATA ops pending */
+  if (t->use_packet_coalescing) {
+    if (s->state.flush_cronet_when_ready) {
+      CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
+      bidirectional_stream_flush(stream);
+    }
+  }
   gpr_mu_unlock(&s->mu);
   execute_from_storage(s);
 }
@@ -528,6 +544,7 @@ static void on_response_trailers_received(
   CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream,
              trailers);
   stream_obj *s = (stream_obj *)stream->annotation;
+  grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   gpr_mu_lock(&s->mu);
   memset(&s->state.rs.trailing_metadata, 0,
          sizeof(s->state.rs.trailing_metadata));
@@ -558,6 +575,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);
+    if (t->use_packet_coalescing) {
+      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);
@@ -607,7 +628,7 @@ static void convert_metadata_to_cronet_headers(
     curr = curr->next;
     num_headers_available++;
   }
-  /* Allocate enough memory. It is freed in the on_request_headers_sent callback
+  /* Allocate enough memory. It is freed in the on_stream_ready callback
    */
   bidirectional_stream_header *headers =
       (bidirectional_stream_header *)gpr_malloc(
@@ -687,8 +708,10 @@ static bool header_has_authority(grpc_linked_mdelem *head) {
   executed. This is the heart of the state machine.
 */
 static bool op_can_be_run(grpc_transport_stream_op *curr_op,
-                          struct op_state *stream_state,
-                          struct op_state *op_state, enum e_op_id op_id) {
+                          struct stream_obj *s, struct op_state *op_state,
+                          enum e_op_id op_id) {
+  struct op_state *stream_state = &s->state;
+  grpc_cronet_transport *t = s->curr_ct;
   bool result = true;
   /* When call is canceled, every op can be run, except under following
   conditions
@@ -755,12 +778,14 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
     else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
       result = false;
     /* we haven't sent message yet */
-    else if (curr_op->send_message &&
+    else if (stream_state->unprocessed_send_message &&
              !stream_state->state_op_done[OP_SEND_MESSAGE])
       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])
+             !stream_state->state_callback_received[OP_SEND_MESSAGE] &&
+             !(t->use_packet_coalescing &&
+               stream_state->pending_write_for_trailer))
       result = false;
   } else if (op_id == OP_CANCEL_ERROR) {
     /* already executed */
@@ -833,24 +858,28 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
                                           struct op_and_state *oas) {
   grpc_transport_stream_op *stream_op = &oas->op;
   struct stream_obj *s = oas->s;
+  grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   struct op_state *stream_state = &s->state;
   enum e_op_result result = NO_ACTION_POSSIBLE;
   if (stream_op->send_initial_metadata &&
-      op_can_be_run(stream_op, stream_state, &oas->state,
-                    OP_SEND_INITIAL_METADATA)) {
+      op_can_be_run(stream_op, s, &oas->state, OP_SEND_INITIAL_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p OP_SEND_INITIAL_METADATA", oas);
     /* Start new cronet stream. It is destroyed in on_succeeded, on_canceled,
      * on_failed */
     GPR_ASSERT(s->cbs == NULL);
     GPR_ASSERT(!stream_state->state_op_done[OP_SEND_INITIAL_METADATA]);
-    s->cbs = bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
-                                         &cronet_callbacks);
+    s->cbs =
+        bidirectional_stream_create(t->engine, s->curr_gs, &cronet_callbacks);
     CRONET_LOG(GPR_DEBUG, "%p = bidirectional_stream_create()", s->cbs);
+    if (t->use_packet_coalescing) {
+      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;
     convert_metadata_to_cronet_headers(
-        stream_op->send_initial_metadata->list.head, s->curr_ct.host, &url,
+        stream_op->send_initial_metadata->list.head, t->host, &url,
         &s->header_array.headers, &s->header_array.count, &method);
     s->header_array.capacity = s->header_array.count;
     CRONET_LOG(GPR_DEBUG, "bidirectional_stream_start(%p, %s)", s->cbs, url);
@@ -862,30 +891,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
       gpr_free((void *)s->header_array.headers[header_index].value);
     }
     stream_state->state_op_done[OP_SEND_INITIAL_METADATA] = true;
-    result = ACTION_TAKEN_WITH_CALLBACK;
-  } else if (stream_op->recv_initial_metadata &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_RECV_INITIAL_METADATA)) {
-    CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_INITIAL_METADATA", oas);
-    if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
-      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                         GRPC_ERROR_NONE);
-    } else if (stream_state->state_callback_received[OP_FAILED]) {
-      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                         GRPC_ERROR_NONE);
-    } else {
-      grpc_chttp2_incoming_metadata_buffer_publish(
-          exec_ctx, &oas->s->state.rs.initial_metadata,
-          stream_op->recv_initial_metadata);
-      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                         GRPC_ERROR_NONE);
+    if (t->use_packet_coalescing) {
+      if (!stream_op->send_message && !stream_op->send_trailing_metadata) {
+        s->state.flush_cronet_when_ready = true;
+      }
     }
-    stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
-    result = ACTION_TAKEN_NO_CALLBACK;
+    result = ACTION_TAKEN_WITH_CALLBACK;
   } else if (stream_op->send_message &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_SEND_MESSAGE)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_SEND_MESSAGE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_MESSAGE", oas);
+    stream_state->unprocessed_send_message = false;
     if (stream_state->state_callback_received[OP_FAILED]) {
       result = NO_ACTION_POSSIBLE;
       CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
@@ -916,16 +931,63 @@ 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);
-        result = ACTION_TAKEN_WITH_CALLBACK;
+        if (t->use_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;
+          } else {
+            stream_state->pending_write_for_trailer = true;
+            result = ACTION_TAKEN_NO_CALLBACK;
+          }
+        } else {
+          result = ACTION_TAKEN_WITH_CALLBACK;
+        }
       } else {
         result = NO_ACTION_POSSIBLE;
       }
     }
     stream_state->state_op_done[OP_SEND_MESSAGE] = true;
     oas->state.state_op_done[OP_SEND_MESSAGE] = true;
+  } else if (stream_op->send_trailing_metadata &&
+             op_can_be_run(stream_op, s, &oas->state,
+                           OP_SEND_TRAILING_METADATA)) {
+    CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_TRAILING_METADATA", oas);
+    if (stream_state->state_callback_received[OP_FAILED]) {
+      result = NO_ACTION_POSSIBLE;
+      CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
+    } else {
+      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
+      stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
+      bidirectional_stream_write(s->cbs, "", 0, true);
+      if (t->use_packet_coalescing) {
+        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;
+  } else if (stream_op->recv_initial_metadata &&
+             op_can_be_run(stream_op, s, &oas->state,
+                           OP_RECV_INITIAL_METADATA)) {
+    CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_INITIAL_METADATA", oas);
+    if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
+      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                         GRPC_ERROR_NONE);
+    } else if (stream_state->state_callback_received[OP_FAILED]) {
+      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                         GRPC_ERROR_NONE);
+    } else {
+      grpc_chttp2_incoming_metadata_buffer_publish(
+          exec_ctx, &oas->s->state.rs.initial_metadata,
+          stream_op->recv_initial_metadata);
+      grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                         GRPC_ERROR_NONE);
+    }
+    stream_state->state_op_done[OP_RECV_INITIAL_METADATA] = true;
+    result = ACTION_TAKEN_NO_CALLBACK;
   } else if (stream_op->recv_message &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_RECV_MESSAGE)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_RECV_MESSAGE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_MESSAGE", oas);
     if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
       CRONET_LOG(GPR_DEBUG, "Stream is cancelled.");
@@ -980,6 +1042,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
                              GRPC_ERROR_NONE);
           stream_state->state_op_done[OP_RECV_MESSAGE] = true;
           oas->state.state_op_done[OP_RECV_MESSAGE] = true;
+
+          /* Extra read to trigger on_succeed */
+          stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
+          stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
+          stream_state->rs.received_bytes = 0;
+          CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs);
+          stream_state->state_op_done[OP_READ_REQ_MADE] =
+              true; /* Indicates that at least one read request has been made */
+          bidirectional_stream_read(s->cbs, stream_state->rs.read_buffer,
+                                    stream_state->rs.remaining_bytes);
           result = ACTION_TAKEN_NO_CALLBACK;
         }
       } else if (stream_state->rs.remaining_bytes == 0) {
@@ -1027,7 +1099,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
       result = ACTION_TAKEN_NO_CALLBACK;
     }
   } else if (stream_op->recv_trailing_metadata &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
+             op_can_be_run(stream_op, s, &oas->state,
                            OP_RECV_TRAILING_METADATA)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_RECV_TRAILING_METADATA", oas);
     if (oas->s->state.rs.trailing_metadata_valid) {
@@ -1038,23 +1110,8 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
     }
     stream_state->state_op_done[OP_RECV_TRAILING_METADATA] = true;
     result = ACTION_TAKEN_NO_CALLBACK;
-  } else if (stream_op->send_trailing_metadata &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_SEND_TRAILING_METADATA)) {
-    CRONET_LOG(GPR_DEBUG, "running: %p  OP_SEND_TRAILING_METADATA", oas);
-    if (stream_state->state_callback_received[OP_FAILED]) {
-      result = NO_ACTION_POSSIBLE;
-      CRONET_LOG(GPR_DEBUG, "Stream is either cancelled or failed.");
-    } else {
-      CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, 0)", s->cbs);
-      stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
-      bidirectional_stream_write(s->cbs, "", 0, true);
-      result = ACTION_TAKEN_WITH_CALLBACK;
-    }
-    stream_state->state_op_done[OP_SEND_TRAILING_METADATA] = true;
   } else if (stream_op->cancel_error &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_CANCEL_ERROR)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_CANCEL_ERROR)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_CANCEL_ERROR", oas);
     CRONET_LOG(GPR_DEBUG, "W: bidirectional_stream_cancel(%p)", s->cbs);
     if (s->cbs) {
@@ -1068,8 +1125,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
       stream_state->cancel_error = GRPC_ERROR_REF(stream_op->cancel_error);
     }
   } else if (stream_op->on_complete &&
-             op_can_be_run(stream_op, stream_state, &oas->state,
-                           OP_ON_COMPLETE)) {
+             op_can_be_run(stream_op, s, &oas->state, OP_ON_COMPLETE)) {
     CRONET_LOG(GPR_DEBUG, "running: %p  OP_ON_COMPLETE", oas);
     if (stream_state->state_op_done[OP_CANCEL_ERROR]) {
       grpc_closure_sched(exec_ctx, stream_op->on_complete,
@@ -1133,6 +1189,12 @@ 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;
+  s->state.flush_cronet_when_ready = s->state.pending_write_for_trailer = false;
+  s->state.unprocessed_send_message = false;
+
+  s->curr_gs = gs;
+  s->curr_ct = (grpc_cronet_transport *)gt;
+
   gpr_mu_init(&s->mu);
   return 0;
 }
@@ -1148,8 +1210,6 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                               grpc_stream *gs, grpc_transport_stream_op *op) {
   CRONET_LOG(GPR_DEBUG, "perform_stream_op");
   stream_obj *s = (stream_obj *)gs;
-  s->curr_gs = gs;
-  memcpy(&s->curr_ct, gt, sizeof(grpc_cronet_transport));
   add_to_storage(s, op);
   if (op->send_initial_metadata &&
       header_has_authority(op->send_initial_metadata->list.head)) {
@@ -1197,14 +1257,58 @@ static grpc_endpoint *get_endpoint(grpc_exec_ctx *exec_ctx,
 static void perform_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                        grpc_transport_op *op) {}
 
-const grpc_transport_vtable grpc_cronet_vtable = {sizeof(stream_obj),
-                                                  "cronet_http",
-                                                  init_stream,
-                                                  set_pollset_do_nothing,
-                                                  set_pollset_set_do_nothing,
-                                                  perform_stream_op,
-                                                  perform_op,
-                                                  destroy_stream,
-                                                  destroy_transport,
-                                                  get_peer,
-                                                  get_endpoint};
+static const grpc_transport_vtable grpc_cronet_vtable = {
+    sizeof(stream_obj),
+    "cronet_http",
+    init_stream,
+    set_pollset_do_nothing,
+    set_pollset_set_do_nothing,
+    perform_stream_op,
+    perform_op,
+    destroy_stream,
+    destroy_transport,
+    get_peer,
+    get_endpoint};
+
+grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
+                                             const grpc_channel_args *args,
+                                             void *reserved) {
+  grpc_cronet_transport *ct = gpr_malloc(sizeof(grpc_cronet_transport));
+  if (!ct) {
+    goto error;
+  }
+  ct->base.vtable = &grpc_cronet_vtable;
+  ct->engine = engine;
+  ct->host = gpr_malloc(strlen(target) + 1);
+  if (!ct->host) {
+    goto error;
+  }
+  strcpy(ct->host, target);
+
+  ct->use_packet_coalescing = true;
+  if (args) {
+    for (size_t i = 0; i < args->num_args; i++) {
+      if (0 ==
+          strcmp(args->args[i].key, GRPC_ARG_USE_CRONET_PACKET_COALESCING)) {
+        if (args->args[i].type != GRPC_ARG_INTEGER) {
+          gpr_log(GPR_ERROR, "%s ignored: it must be an integer",
+                  GRPC_ARG_USE_CRONET_PACKET_COALESCING);
+        } else {
+          ct->use_packet_coalescing = (args->args[i].value.integer != 0);
+        }
+      }
+    }
+  }
+
+  return &ct->base;
+
+error:
+  if (ct) {
+    if (ct->host) {
+      gpr_free(ct->host);
+    }
+    gpr_free(ct);
+  }
+
+  return NULL;
+}

src/core/ext/transport/cronet/transport/cronet_transport.h

@@ -0,0 +1,43 @@
+/*
+ *
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
+#define GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H
+
+#include "src/core/lib/transport/transport.h"
+
+grpc_transport *grpc_create_cronet_transport(void *engine, const char *target,
+                                             const grpc_channel_args *args,
+                                             void *reserved);
+
+#endif /* GRPC_CORE_EXT_TRANSPORT_CRONET_TRANSPORT_CRONET_TRANSPORT_H */

src/objective-c/tests/CronetUnitTests/CronetUnitTests.m

@@ -32,13 +32,13 @@
  */
 
 #import <XCTest/XCTest.h>
-#import <sys/socket.h>
 #import <netinet/in.h>
+#import <sys/socket.h>
 
 #import <Cronet/Cronet.h>
-#import <grpc/support/host_port.h>
-#import <grpc/grpc_cronet.h>
 #import <grpc/grpc.h>
+#import <grpc/grpc_cronet.h>
+#import <grpc/support/host_port.h>
 #import "test/core/end2end/cq_verifier.h"
 #import "test/core/util/port.h"
 
@@ -49,16 +49,19 @@
 #import "src/core/lib/support/env.h"
 #import "src/core/lib/support/string.h"
 #import "src/core/lib/support/tmpfile.h"
+#import "test/core/end2end/data/ssl_test_data.h"
 #import "test/core/util/test_config.h"
 
+#import <BoringSSL/openssl/ssl.h>
+
 static void drain_cq(grpc_completion_queue *cq) {
   grpc_event ev;
   do {
-    ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5), NULL);
+    ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5),
+                                    NULL);
   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
 }
 
-
 @interface CronetUnitTests : XCTestCase
 
 @end
@@ -68,47 +71,99 @@ static void drain_cq(grpc_completion_queue *cq) {
 + (void)setUp {
   [super setUp];
 
-/***  FILE *roots_file;
-  size_t roots_size = strlen(test_root_cert);*/
-
   char *argv[] = {"CoreCronetEnd2EndTests"};
   grpc_test_init(1, argv);
 
   grpc_init();
 
   [Cronet setHttp2Enabled:YES];
+  [Cronet setSslKeyLogFileName:@"Documents/key"];
+  [Cronet enableTestCertVerifierForTesting];
   NSURL *url = [[[NSFileManager defaultManager]
-                 URLsForDirectory:NSDocumentDirectory
-                 inDomains:NSUserDomainMask] lastObject];
+      URLsForDirectory:NSDocumentDirectory
+             inDomains:NSUserDomainMask] lastObject];
   NSLog(@"Documents directory: %@", url);
   [Cronet start];
   [Cronet startNetLogToFile:@"Documents/cronet_netlog.json" logBytes:YES];
+
+  init_ssl();
 }
 
 + (void)tearDown {
   grpc_shutdown();
+  cleanup_ssl();
 
   [super tearDown];
 }
 
+void init_ssl(void) {
+  SSL_load_error_strings();
+  OpenSSL_add_ssl_algorithms();
+}
+
+void cleanup_ssl(void) { EVP_cleanup(); }
+
+int alpn_cb(SSL *ssl, const unsigned char **out, unsigned char *outlen,
+            const unsigned char *in, unsigned int inlen, void *arg) {
+  // Always select "h2" as the ALPN protocol to be used
+  *out = (const unsigned char *)"h2";
+  *outlen = 2;
+  return SSL_TLSEXT_ERR_OK;
+}
+
+void init_ctx(SSL_CTX *ctx) {
+  // Install server certificate
+  BIO *pem = BIO_new_mem_buf((void *)test_server1_cert,
+                             (int)strlen(test_server1_cert));
+  X509 *cert = PEM_read_bio_X509_AUX(pem, NULL, NULL, "");
+  SSL_CTX_use_certificate(ctx, cert);
+  X509_free(cert);
+  BIO_free(pem);
+
+  // Install server private key
+  pem =
+      BIO_new_mem_buf((void *)test_server1_key, (int)strlen(test_server1_key));
+  EVP_PKEY *key = PEM_read_bio_PrivateKey(pem, NULL, NULL, "");
+  SSL_CTX_use_PrivateKey(ctx, key);
+  EVP_PKEY_free(key);
+  BIO_free(pem);
+
+  // Select cipher suite
+  SSL_CTX_set_cipher_list(ctx, "ECDHE-RSA-AES128-GCM-SHA256");
+
+  // Select ALPN protocol
+  SSL_CTX_set_alpn_select_cb(ctx, alpn_cb, NULL);
+}
+
+unsigned int parse_h2_length(const char *field) {
+  return ((unsigned int)(unsigned char)(field[0])) * 65536 +
+         ((unsigned int)(unsigned char)(field[1])) * 256 +
+         ((unsigned int)(unsigned char)(field[2]));
+}
+
 - (void)testInternalError {
   grpc_call *c;
   grpc_slice request_payload_slice =
-  grpc_slice_from_copied_string("hello world");
+      grpc_slice_from_copied_string("hello world");
   grpc_byte_buffer *request_payload =
-  grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
-  grpc_metadata meta_c[2] = {
-    {"key1", "val1", 4, 0, {{NULL, NULL, NULL, NULL}}},
-    {"key2", "val2", 4, 0, {{NULL, NULL, NULL, NULL}}}};
+  grpc_metadata meta_c[2] = {{grpc_slice_from_static_string("key1"),
+                              grpc_slice_from_static_string("val1"),
+                              0,
+                              {{NULL, NULL, NULL, NULL}}},
+                             {grpc_slice_from_static_string("key2"),
+                              grpc_slice_from_static_string("val2"),
+                              0,
+                              {{NULL, NULL, NULL, NULL}}}};
 
   int port = grpc_pick_unused_port_or_die();
   char *addr;
   gpr_join_host_port(&addr, "127.0.0.1", port);
   grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
-  cronet_engine *cronetEngine = [Cronet getGlobalEngine];
-  grpc_channel *client = grpc_cronet_secure_channel_create(cronetEngine, addr,
-                                                           NULL, NULL);
+  stream_engine *cronetEngine = [Cronet getGlobalEngine];
+  grpc_channel *client =
+      grpc_cronet_secure_channel_create(cronetEngine, addr, NULL, NULL);
 
   cq_verifier *cqv = cq_verifier_create(cq);
   grpc_op ops[6];
@@ -120,12 +175,11 @@ static void drain_cq(grpc_completion_queue *cq) {
   grpc_call_details call_details;
   grpc_status_code status;
   grpc_call_error error;
-  char *details = NULL;
-  size_t details_capacity = 0;
+  grpc_slice details;
 
-  c = grpc_channel_create_call(
-                               client, NULL, GRPC_PROPAGATE_DEFAULTS, cq, "/foo",
-                               NULL, deadline, NULL);
+  c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
+                               grpc_slice_from_static_string("/foo"), NULL,
+                               deadline, NULL);
   GPR_ASSERT(c);
 
   grpc_metadata_array_init(&initial_metadata_recv);
@@ -164,35 +218,40 @@ static void drain_cq(grpc_completion_queue *cq) {
   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
   op->data.recv_status_on_client.status = &status;
   op->data.recv_status_on_client.status_details = &details;
-  op->data.recv_status_on_client.status_details_capacity = &details_capacity;
   op->flags = 0;
   op->reserved = NULL;
   op++;
-  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void*)1, NULL);
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void *)1, NULL);
   GPR_ASSERT(GRPC_CALL_OK == error);
 
-  dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
-    int sl = socket(AF_INET, SOCK_STREAM, 0);
-    GPR_ASSERT(sl >= 0);
-    struct sockaddr_in s_addr;
-    memset(&s_addr, 0, sizeof(s_addr));
-    s_addr.sin_family = AF_INET;
-    s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
-    s_addr.sin_port = htons(port);
-    bind(sl, (struct sockaddr*)&s_addr, sizeof(s_addr));
-    listen(sl, 5);
-    int s = accept(sl, NULL, NULL);
-    sleep(1);
-    close(s);
-    close(sl);
-  });
-
-  CQ_EXPECT_COMPLETION(cqv, (void*)1, 1);
+  dispatch_async(
+      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
+        int sl = socket(AF_INET, SOCK_STREAM, 0);
+        GPR_ASSERT(sl >= 0);
+
+        // Make and TCP endpoint to accept the connection
+        struct sockaddr_in s_addr;
+        memset(&s_addr, 0, sizeof(s_addr));
+        s_addr.sin_family = AF_INET;
+        s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
+        s_addr.sin_port = htons(port);
+        GPR_ASSERT(0 == bind(sl, (struct sockaddr *)&s_addr, sizeof(s_addr)));
+        GPR_ASSERT(0 == listen(sl, 5));
+        int s = accept(sl, NULL, NULL);
+        GPR_ASSERT(s >= 0);
+
+        // Close the connection after 1 second to trigger Cronet's on_failed()
+        sleep(1);
+        close(s);
+        close(sl);
+      });
+
+  CQ_EXPECT_COMPLETION(cqv, (void *)1, 1);
   cq_verify(cqv);
 
   GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
 
-  gpr_free(details);
+  grpc_slice_unref(details);
   grpc_metadata_array_destroy(&initial_metadata_recv);
   grpc_metadata_array_destroy(&trailing_metadata_recv);
   grpc_metadata_array_destroy(&request_metadata_recv);
@@ -204,11 +263,195 @@ static void drain_cq(grpc_completion_queue *cq) {
 
   grpc_byte_buffer_destroy(request_payload);
   grpc_byte_buffer_destroy(response_payload_recv);
-  
+
+  grpc_channel_destroy(client);
+  grpc_completion_queue_shutdown(cq);
+  drain_cq(cq);
+  grpc_completion_queue_destroy(cq);
+}
+
+- (void)packetCoalescing:(BOOL)useCoalescing {
+  grpc_arg arg;
+  arg.key = GRPC_ARG_USE_CRONET_PACKET_COALESCING;
+  arg.type = GRPC_ARG_INTEGER;
+  arg.value.integer = useCoalescing ? 1 : 0;
+  grpc_channel_args *args = grpc_channel_args_copy_and_add(NULL, &arg, 1);
+  grpc_call *c;
+  grpc_slice request_payload_slice =
+      grpc_slice_from_copied_string("hello world");
+  grpc_byte_buffer *request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
+  grpc_metadata meta_c[2] = {{grpc_slice_from_static_string("key1"),
+                              grpc_slice_from_static_string("val1"),
+                              0,
+                              {{NULL, NULL, NULL, NULL}}},
+                             {grpc_slice_from_static_string("key2"),
+                              grpc_slice_from_static_string("val2"),
+                              0,
+                              {{NULL, NULL, NULL, NULL}}}};
+
+  int port = grpc_pick_unused_port_or_die();
+  char *addr;
+  gpr_join_host_port(&addr, "127.0.0.1", port);
+  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  stream_engine *cronetEngine = [Cronet getGlobalEngine];
+  grpc_channel *client =
+      grpc_cronet_secure_channel_create(cronetEngine, addr, args, NULL);
+
+  cq_verifier *cqv = cq_verifier_create(cq);
+  grpc_op ops[6];
+  grpc_op *op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_byte_buffer *response_payload_recv = NULL;
+  grpc_call_details call_details;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+
+  c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
+                               grpc_slice_from_static_string("/foo"), NULL,
+                               deadline, NULL);
+  GPR_ASSERT(c);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 2;
+  op->data.send_initial_metadata.metadata = meta_c;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), (void *)1, NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"Coalescing"];
+
+  dispatch_async(
+      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
+        int sl = socket(AF_INET, SOCK_STREAM, 0);
+        GPR_ASSERT(sl >= 0);
+        struct sockaddr_in s_addr;
+        memset(&s_addr, 0, sizeof(s_addr));
+        s_addr.sin_family = AF_INET;
+        s_addr.sin_addr.s_addr = htonl(INADDR_ANY);
+        s_addr.sin_port = htons(port);
+        GPR_ASSERT(0 == bind(sl, (struct sockaddr *)&s_addr, sizeof(s_addr)));
+        GPR_ASSERT(0 == listen(sl, 5));
+        int s = accept(sl, NULL, NULL);
+        GPR_ASSERT(s >= 0);
+        struct timeval tv;
+        tv.tv_sec = 2;
+        tv.tv_usec = 0;
+        setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+
+        // Make an TLS endpoint to receive Cronet's transmission
+        SSL_CTX *ctx = SSL_CTX_new(TLSv1_2_server_method());
+        init_ctx(ctx);
+        SSL *ssl = SSL_new(ctx);
+        SSL_set_fd(ssl, s);
+        SSL_accept(ssl);
+
+        const char magic[] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
+
+        char buf[4096];
+        long len;
+        BOOL coalesced = NO;
+        while ((len = SSL_read(ssl, buf, sizeof(buf))) > 0) {
+          gpr_log(GPR_DEBUG, "Read len: %ld", len);
+
+          // Analyze the HTTP/2 frames in the same TLS PDU to identify if
+          // coalescing is successful
+          unsigned int p = 0;
+          while (p < len) {
+            if (len - p >= 24 && 0 == memcmp(&buf[p], magic, 24)) {
+              p += 24;
+              continue;
+            }
+
+            if (buf[p + 3] == 0 &&                   // Type is DATA
+                parse_h2_length(&buf[p]) == 0x10 &&  // Length is correct
+                (buf[p + 4] & 1) != 0 &&             // EOS bit is set
+                0 == memcmp("hello world", &buf[p + 14],
+                            11)) {  // Message is correct
+              coalesced = YES;
+              break;
+            }
+            p += (parse_h2_length(&buf[p]) + 9);
+          }
+          if (coalesced) {
+            break;
+          }
+        }
+
+        XCTAssert(coalesced == useCoalescing);
+        SSL_free(ssl);
+        SSL_CTX_free(ctx);
+        close(s);
+        close(sl);
+        [expectation fulfill];
+      });
+
+  CQ_EXPECT_COMPLETION(cqv, (void *)1, 1);
+  cq_verify(cqv);
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+
+  grpc_call_destroy(c);
+
+  cq_verifier_destroy(cqv);
+
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload_recv);
+
   grpc_channel_destroy(client);
   grpc_completion_queue_shutdown(cq);
   drain_cq(cq);
   grpc_completion_queue_destroy(cq);
+  
+  [self waitForExpectationsWithTimeout:4 handler:nil];
+}
+
+- (void)testPacketCoalescing {
+  [self packetCoalescing:YES];
+  [self packetCoalescing:NO];
 }
 
 @end

src/objective-c/tests/Tests.xcodeproj/project.pbxproj

@@ -1474,6 +1474,7 @@
 					"$(inherited)",
 					"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
 					"GRPC_COMPILE_WITH_CRONET=1",
+					"GRPC_CRONET_WITH_PACKET_COALESCING=1",
 				);
 				INFOPLIST_FILE = InteropTestsRemoteWithCronet/Info.plist;
 				IPHONEOS_DEPLOYMENT_TARGET = 9.3;

src/objective-c/tests/Tests.xcodeproj/xcshareddata/xcschemes/AllTests.xcscheme

@@ -59,6 +59,16 @@
                ReferencedContainer = "container:Tests.xcodeproj">
             </BuildableReference>
          </TestableReference>
+         <TestableReference
+            skipped = "NO">
+            <BuildableReference
+               BuildableIdentifier = "primary"
+               BlueprintIdentifier = "5EAD6D231E27047400002378"
+               BuildableName = "CronetUnitTests.xctest"
+               BlueprintName = "CronetUnitTests"
+               ReferencedContainer = "container:Tests.xcodeproj">
+            </BuildableReference>
+         </TestableReference>
       </Testables>
       <MacroExpansion>
          <BuildableReference
@@ -100,6 +110,15 @@
       savedToolIdentifier = ""
       useCustomWorkingDirectory = "NO"
       debugDocumentVersioning = "YES">
+      <MacroExpansion>
+         <BuildableReference
+            BuildableIdentifier = "primary"
+            BlueprintIdentifier = "63423F431B150A5F006CF63C"
+            BuildableName = "AllTests.xctest"
+            BlueprintName = "AllTests"
+            ReferencedContainer = "container:Tests.xcodeproj">
+         </BuildableReference>
+      </MacroExpansion>
    </ProfileAction>
    <AnalyzeAction
       buildConfiguration = "Debug">

templates/gRPC-Core.podspec.template

@@ -161,7 +161,8 @@
 
     s.subspec 'Cronet-Interface' do |ss|
       ss.header_mappings_dir = 'include/grpc'
-      ss.source_files = 'include/grpc/grpc_cronet.h'
+      ss.source_files = 'include/grpc/grpc_cronet.h',
+                        'src/core/ext/transport/cronet/transport/cronet_transport.h'
     end
 
     s.subspec 'Cronet-Implementation' do |ss|

test/core/end2end/gen_build_yaml.py

@@ -91,6 +91,7 @@ LOWCPU = 0.1
 
 # maps test names to options
 END2END_TESTS = {
+    'authority_not_supported': default_test_options,
     'bad_hostname': default_test_options,
     'binary_metadata': default_test_options,
     'resource_quota_server': default_test_options._replace(large_writes=True,
@@ -142,7 +143,6 @@ END2END_TESTS = {
     'simple_request': default_test_options,
     'streaming_error_response': default_test_options,
     'trailing_metadata': default_test_options,
-    'authority_not_supported': default_test_options,
     'write_buffering': default_test_options,
     'write_buffering_at_end': default_test_options,
 }

tools/codegen/core/gen_nano_proto.sh

@@ -83,7 +83,7 @@ popd
 
 # this should be the same version as the submodule we compile against
 # ideally we'd update this as a template to ensure that
-pip install protobuf==3.0.0
+pip install protobuf==3.2.0
 
 pushd "$(dirname $INPUT_PROTO)" > /dev/null
 

tools/distrib/pylint_code.sh

@@ -0,0 +1,48 @@
+#!/bin/bash
+# Copyright 2017, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+set -ex
+
+# change to root directory
+cd $(dirname $0)/../..
+
+DIRS=src/python/grpcio/grpc
+
+VIRTUALENV=python_pylint_venv
+
+virtualenv $VIRTUALENV
+PYTHON=`realpath $VIRTUALENV/bin/python`
+$PYTHON -m pip install pylint==1.6.5
+
+for dir in $DIRS; do
+  $PYTHON -m pylint --rcfile=.pylintrc -rn $dir || exit $?
+done
+
+exit 0

tools/run_tests/generated/sources_and_headers.json

@@ -8032,6 +8032,7 @@
       "include/grpc/grpc_cronet.h", 
       "include/grpc/grpc_security.h", 
       "include/grpc/grpc_security_constants.h", 
+      "src/core/ext/transport/cronet/transport/cronet_transport.h", 
       "third_party/objective_c/Cronet/bidirectional_stream_c.h"
     ], 
     "is_filegroup": true, 
@@ -8043,7 +8044,8 @@
       "include/grpc/grpc_security_constants.h", 
       "src/core/ext/transport/cronet/client/secure/cronet_channel_create.c", 
       "src/core/ext/transport/cronet/transport/cronet_api_dummy.c", 
-      "src/core/ext/transport/cronet/transport/cronet_transport.c"
+      "src/core/ext/transport/cronet/transport/cronet_transport.c", 
+      "src/core/ext/transport/cronet/transport/cronet_transport.h"
     ], 
     "third_party": false, 
     "type": "filegroup"

tools/run_tests/run_microbenchmark.py

@@ -128,20 +128,21 @@ def collect_perf(bm_name, args):
        'CONFIG=mutrace', '-j', '%d' % multiprocessing.cpu_count()])
   for line in subprocess.check_output(['bins/mutrace/%s' % bm_name,
                                        '--benchmark_list_tests']).splitlines():
-    subprocess.check_call(['sudo', 'perf', 'record', '-o', 'perf.data',
+    subprocess.check_call(['perf', 'record', '-o', '%s-perf.data' % fnize(line),
                            '-g', '-c', '1000',
                            'bins/mutrace/%s' % bm_name,
                            '--benchmark_filter=^%s$' % line,
-                           '--benchmark_min_time=20'])
-    with open('bm.perf', 'w') as f:
-      f.write(subprocess.check_output(['sudo', 'perf', 'script']))
-    with open('bm.folded', 'w') as f:
-      f.write(subprocess.check_output([
-          '%s/stackcollapse-perf.pl' % flamegraph_dir, 'bm.perf']))
-    link(line, '%s.svg' % fnize(line))
-    with open('reports/%s.svg' % fnize(line), 'w') as f:
-      f.write(subprocess.check_output([
-          '%s/flamegraph.pl' % flamegraph_dir, 'bm.folded']))
+                           '--benchmark_min_time=10'])
+    env = os.environ.copy()
+    env.update({
+      'PERF_BASE_NAME': fnize(line),
+      'OUTPUT_DIR': 'reports',
+      'OUTPUT_FILENAME': fnize(line),
+    })
+    subprocess.check_call(['tools/run_tests/performance/process_local_perf_flamegraphs.sh'],
+                          env=env)
+    subprocess.check_call(['rm', '%s-perf.data' % fnize(line)])
+    subprocess.check_call(['rm', '%s-out.perf' % fnize(line)])
 
 def collect_summary(bm_name, args):
   heading('Summary: %s' % bm_name)

tools/run_tests/sanity/sanity_tests.yaml

@@ -12,6 +12,7 @@
 - script: tools/distrib/check_trailing_newlines.sh
 - script: tools/distrib/check_nanopb_output.sh
 - script: tools/distrib/check_include_guards.py
+- script: tools/distrib/pylint_code.sh
 - script: tools/distrib/yapf_code.sh
 - script: tools/distrib/python/check_grpcio_tools.py