Packet coalescing transport layer and end2end test changes

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

@@ -77,9 +77,8 @@ int cronet_bidirectional_stream_write(cronet_bidirectional_stream* stream,
   return 0;
 }
 
-int cronet_bidirectional_stream_cancel(cronet_bidirectional_stream* stream) {
+void cronet_bidirectional_stream_cancel(cronet_bidirectional_stream* stream) {
   GPR_ASSERT(0);
-  return 0;
 }
 
 #endif /* GRPC_COMPILE_WITH_CRONET */

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

@@ -86,7 +86,7 @@ enum e_op_id {
 
 /* Cronet callbacks. See cronet_c_for_grpc.h for documentation for each. */
 
-static void on_request_headers_sent(cronet_bidirectional_stream *);
+static void on_stream_ready(cronet_bidirectional_stream *);
 static void on_response_headers_received(
     cronet_bidirectional_stream *,
     const cronet_bidirectional_stream_header_array *, const char *);
@@ -99,7 +99,7 @@ static void on_succeeded(cronet_bidirectional_stream *);
 static void on_failed(cronet_bidirectional_stream *, int);
 static void on_canceled(cronet_bidirectional_stream *);
 static cronet_bidirectional_stream_callback cronet_callbacks = {
-    on_request_headers_sent,
+    on_stream_ready,
     on_response_headers_received,
     on_read_completed,
     on_write_completed,
@@ -151,6 +151,11 @@ 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;
   /* data structure for storing data coming from server */
   struct read_state rs;
@@ -273,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);
@@ -405,8 +413,8 @@ static void on_succeeded(cronet_bidirectional_stream *stream) {
 /*
   Cronet callback
 */
-static void on_request_headers_sent(cronet_bidirectional_stream *stream) {
-  CRONET_LOG(GPR_DEBUG, "W: on_request_headers_sent(%p)", stream);
+static void on_stream_ready(cronet_bidirectional_stream *stream) {
+  CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream);
   stream_obj *s = (stream_obj *)stream->annotation;
   gpr_mu_lock(&s->mu);
   s->state.state_op_done[OP_SEND_INITIAL_METADATA] = true;
@@ -416,6 +424,14 @@ static void on_request_headers_sent(cronet_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 */
+#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
+  if (s->state.flush_cronet_when_ready) {
+    CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
+    cronet_bidirectional_stream_flush(stream);
+  }
+#endif
   gpr_mu_unlock(&s->mu);
   execute_from_storage(s);
 }
@@ -551,6 +567,10 @@ static void on_response_trailers_received(
     CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_write (%p, 0)", s->cbs);
     s->state.state_callback_received[OP_SEND_MESSAGE] = false;
     cronet_bidirectional_stream_write(s->cbs, "", 0, true);
+#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
+    CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
+    cronet_bidirectional_stream_flush(s->cbs);
+#endif
     s->state.state_op_done[OP_SEND_TRAILING_METADATA] = true;
 
     gpr_mu_unlock(&s->mu);
@@ -598,7 +618,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
    */
   cronet_bidirectional_stream_header *headers =
       (cronet_bidirectional_stream_header *)gpr_malloc(
@@ -740,12 +760,16 @@ 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]
+#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
+             && !stream_state->pending_write_for_trailer
+#endif
+             )
       result = false;
   } else if (op_id == OP_CANCEL_ERROR) {
     /* already executed */
@@ -831,6 +855,10 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
     s->cbs = cronet_bidirectional_stream_create(s->curr_ct.engine, s->curr_gs,
                                                 &cronet_callbacks);
     CRONET_LOG(GPR_DEBUG, "%p = cronet_bidirectional_stream_create()", s->cbs);
+#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
+    cronet_bidirectional_stream_disable_auto_flush(s->cbs, true);
+    cronet_bidirectional_stream_delay_request_headers_until_flush(s->cbs, true);
+#endif
     char *url = NULL;
     const char *method = "POST";
     s->header_array.headers = NULL;
@@ -843,30 +871,17 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
     cronet_bidirectional_stream_start(s->cbs, url, 0, method, &s->header_array,
                                       false);
     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_exec_ctx_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                          GRPC_ERROR_CANCELLED, NULL);
-    } else if (stream_state->state_callback_received[OP_FAILED]) {
-      grpc_exec_ctx_sched(
-          exec_ctx, stream_op->recv_initial_metadata_ready,
-          make_error_with_desc(GRPC_STATUS_UNAVAILABLE, "Unavailable."), NULL);
-    } else {
-      grpc_chttp2_incoming_metadata_buffer_publish(
-          &oas->s->state.rs.initial_metadata, stream_op->recv_initial_metadata);
-      grpc_exec_ctx_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
-                          GRPC_ERROR_NONE, NULL);
+#ifdef GRPC_CRONET_WITH_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;
+#endif
+    result = ACTION_TAKEN_WITH_CALLBACK;
   } else if (stream_op->send_message &&
              op_can_be_run(stream_op, stream_state, &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.");
@@ -897,13 +912,63 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
         stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
         cronet_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, "cronet_bidirectional_stream_flush (%p)",
+                     s->cbs);
+          cronet_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;
+#endif
       } 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, 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, "cronet_bidirectional_stream_write (%p, 0)",
+                 s->cbs);
+      stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
+      cronet_bidirectional_stream_write(s->cbs, "", 0, true);
+#ifdef GRPC_CRONET_WITH_PACKET_COALESCING
+      CRONET_LOG(GPR_DEBUG, "cronet_bidirectional_stream_flush (%p)", s->cbs);
+      cronet_bidirectional_stream_flush(s->cbs);
+#endif
+      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, 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_exec_ctx_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                          GRPC_ERROR_CANCELLED, NULL);
+    } else if (stream_state->state_callback_received[OP_FAILED]) {
+      grpc_exec_ctx_sched(
+          exec_ctx, stream_op->recv_initial_metadata_ready,
+          make_error_with_desc(GRPC_STATUS_UNAVAILABLE, "Unavailable."), NULL);
+    } else {
+      grpc_chttp2_incoming_metadata_buffer_publish(
+          &oas->s->state.rs.initial_metadata, stream_op->recv_initial_metadata);
+      grpc_exec_ctx_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+                          GRPC_ERROR_NONE, NULL);
+    }
+    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)) {
@@ -962,6 +1027,16 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
                               GRPC_ERROR_NONE, NULL);
           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, "cronet_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 */
+          cronet_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) {
@@ -1020,21 +1095,6 @@ 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, "cronet_bidirectional_stream_write (%p, 0)",
-                 s->cbs);
-      stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
-      cronet_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)) {
@@ -1117,6 +1177,10 @@ 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;
   gpr_mu_init(&s->mu);
   return 0;
 }

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

@@ -346,6 +346,10 @@ static char *roots_filename;
   [self testIndividualCase:"no_op"];
 }
 
+- (void)testPacketCoalescing {
+  [self testIndividualCase:"packet_coalescing"];
+}
+
 - (void)testPayload {
   [self testIndividualCase:"payload"];
 }

src/objective-c/tests/Podfile

@@ -92,6 +92,7 @@ post_install do |installer|
         # GPR_UNREACHABLE_CODE causes "Control may reach end of non-void
         # function" warning
         config.build_settings['GCC_WARN_ABOUT_RETURN_TYPE'] = 'NO'
+        config.build_settings['GCC_PREPROCESSOR_DEFINITIONS'] = '$(inherited) COCOAPODS=1 GRPC_CRONET_WITH_PACKET_COALESCING=1'
       end
     end
 

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

@@ -1296,6 +1296,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;

test/core/end2end/end2end_nosec_tests.c

@@ -103,6 +103,8 @@ extern void no_logging(grpc_end2end_test_config config);
 extern void no_logging_pre_init(void);
 extern void no_op(grpc_end2end_test_config config);
 extern void no_op_pre_init(void);
+extern void packet_coalescing(grpc_end2end_test_config config);
+extern void packet_coalescing_pre_init(void);
 extern void payload(grpc_end2end_test_config config);
 extern void payload_pre_init(void);
 extern void ping(grpc_end2end_test_config config);
@@ -135,6 +137,8 @@ extern void streaming_error_response(grpc_end2end_test_config config);
 extern void streaming_error_response_pre_init(void);
 extern void trailing_metadata(grpc_end2end_test_config config);
 extern void trailing_metadata_pre_init(void);
+extern void packet_coalescing(grpc_end2end_test_config config);
+extern void packet_coalescing_pre_init(void);
 
 void grpc_end2end_tests_pre_init(void) {
   GPR_ASSERT(!g_pre_init_called);
@@ -169,6 +173,7 @@ void grpc_end2end_tests_pre_init(void) {
   network_status_change_pre_init();
   no_logging_pre_init();
   no_op_pre_init();
+  packet_coalescing_pre_init();
   payload_pre_init();
   ping_pre_init();
   ping_pong_streaming_pre_init();
@@ -224,6 +229,7 @@ void grpc_end2end_tests(int argc, char **argv,
     network_status_change(config);
     no_logging(config);
     no_op(config);
+    packet_coalescing(config);
     payload(config);
     ping(config);
     ping_pong_streaming(config);
@@ -364,6 +370,10 @@ void grpc_end2end_tests(int argc, char **argv,
       no_op(config);
       continue;
     }
+    if (0 == strcmp("packet_coalescing", argv[i])) {
+      packet_coalescing(config);
+      continue;
+    }
     if (0 == strcmp("payload", argv[i])) {
       payload(config);
       continue;

test/core/end2end/end2end_tests.c

@@ -105,6 +105,8 @@ extern void no_logging(grpc_end2end_test_config config);
 extern void no_logging_pre_init(void);
 extern void no_op(grpc_end2end_test_config config);
 extern void no_op_pre_init(void);
+extern void packet_coalescing(grpc_end2end_test_config config);
+extern void packet_coalescing_pre_init(void);
 extern void payload(grpc_end2end_test_config config);
 extern void payload_pre_init(void);
 extern void ping(grpc_end2end_test_config config);
@@ -172,6 +174,7 @@ void grpc_end2end_tests_pre_init(void) {
   network_status_change_pre_init();
   no_logging_pre_init();
   no_op_pre_init();
+  packet_coalescing_pre_init();
   payload_pre_init();
   ping_pre_init();
   ping_pong_streaming_pre_init();
@@ -228,6 +231,7 @@ void grpc_end2end_tests(int argc, char **argv,
     network_status_change(config);
     no_logging(config);
     no_op(config);
+    packet_coalescing(config);
     payload(config);
     ping(config);
     ping_pong_streaming(config);
@@ -372,6 +376,10 @@ void grpc_end2end_tests(int argc, char **argv,
       no_op(config);
       continue;
     }
+    if (0 == strcmp("packet_coalescing", argv[i])) {
+      packet_coalescing(config);
+      continue;
+    }
     if (0 == strcmp("payload", argv[i])) {
       payload(config);
       continue;

test/core/end2end/tests/packet_coalescing.c

@@ -0,0 +1,288 @@
+/*
+ *
+ * Copyright 2015, 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.
+ *
+ */
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdio.h>
+#include <string.h>
+
+#include <grpc/byte_buffer.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
+#include <grpc/support/useful.h>
+#include "test/core/end2end/cq_verifier.h"
+
+extern void gpr_default_log(gpr_log_func_args *args);
+
+static void *tag(intptr_t t) { return (void *)t; }
+
+static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
+                                            const char *test_name,
+                                            grpc_channel_args *client_args,
+                                            grpc_channel_args *server_args) {
+  grpc_end2end_test_fixture f;
+  gpr_log(GPR_INFO, "%s/%s", test_name, config.name);
+  f = config.create_fixture(client_args, server_args);
+  config.init_server(&f, server_args);
+  config.init_client(&f, client_args);
+  return f;
+}
+
+static gpr_timespec n_seconds_time(int n) {
+  return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);
+}
+
+static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
+
+static void drain_cq(grpc_completion_queue *cq) {
+  grpc_event ev;
+  do {
+    ev = grpc_completion_queue_next(cq, five_seconds_time(), NULL);
+  } while (ev.type != GRPC_QUEUE_SHUTDOWN);
+}
+
+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);
+  grpc_server_destroy(f->server);
+  f->server = NULL;
+}
+
+static void shutdown_client(grpc_end2end_test_fixture *f) {
+  if (!f->client) return;
+  grpc_channel_destroy(f->client);
+  f->client = NULL;
+}
+
+static void end_test(grpc_end2end_test_fixture *f) {
+  shutdown_server(f);
+  shutdown_client(f);
+
+  grpc_completion_queue_shutdown(f->cq);
+  drain_cq(f->cq);
+  grpc_completion_queue_destroy(f->cq);
+}
+
+static bool coalesced_message_and_eos;
+
+static void log_processor(gpr_log_func_args *args) {
+  const int file_len = (int)strlen(args->file);
+  const char suffix1[] = "secure_endpoint.c";
+  const int suffix1_len = sizeof(suffix1) - 1;
+  const char suffix2[] = "tcp_posix.c";
+  const int suffix2_len = sizeof(suffix2) - 1;
+  const char prefix[] = "READ";
+  const int prefix_len = sizeof(prefix) - 1;
+  if (((file_len >= suffix1_len &&
+        0 == strcmp(suffix1, &args->file[file_len - suffix1_len])) ||
+       (file_len >= suffix2_len &&
+        0 == strcmp(suffix2, &args->file[file_len - suffix2_len]))) &&
+      0 == strncmp(prefix, args->message, (size_t)prefix_len) &&
+      strstr(args->message,
+             "00 00 10 00 01 00 00 00 01 00 00 00 00 0b 68 65 6c 6c 6f 20 77 "
+             "6f 72 6c 64")) {
+    fprintf(stderr, "%s, %s\n", args->file, args->message);
+    coalesced_message_and_eos = true;
+  }
+}
+
+/* Request/response with metadata and payload.*/
+static void test_request_response_with_metadata_and_payload(
+    grpc_end2end_test_config config) {
+  grpc_call *c;
+  grpc_call *s;
+  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 = five_seconds_time();
+  grpc_metadata meta_c[2] = {
+      {"key1", "val1", 4, 0, {{NULL, NULL, NULL, NULL}}},
+      {"key2", "val2", 4, 0, {{NULL, NULL, NULL, NULL}}}};
+  grpc_metadata meta_s[2] = {
+      {"key3", "val3", 4, 0, {{NULL, NULL, NULL, NULL}}},
+      {"key4", "val4", 4, 0, {{NULL, NULL, NULL, NULL}}}};
+  grpc_end2end_test_fixture f = begin_test(
+      config, "test_request_response_with_metadata_and_payload", NULL, NULL);
+  cq_verifier *cqv = cq_verifier_create(f.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 *request_payload_recv = NULL;
+  grpc_call_details call_details;
+  grpc_status_code status;
+  grpc_call_error error;
+  char *details = NULL;
+  size_t details_capacity = 0;
+  int was_cancelled = 2;
+  coalesced_message_and_eos = false;
+
+  c = grpc_channel_create_call(
+      f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, "/foo",
+      get_host_override_string("foo.test.google.fr:1234", config), 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 = 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 = &initial_metadata_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->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), tag(1), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
+  cq_verify(cqv);
+
+  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_s;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &request_payload_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), 1);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_OK;
+  op->data.send_status_from_server.status_details = "xyz";
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_OK);
+  GPR_ASSERT(0 == strcmp(details, "xyz"));
+  GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
+  validate_host_override_string("foo.test.google.fr:1234", call_details.host,
+                                config);
+  GPR_ASSERT(was_cancelled == 0);
+  GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, "hello world"));
+  GPR_ASSERT(contains_metadata(&request_metadata_recv, "key1", "val1"));
+  GPR_ASSERT(contains_metadata(&request_metadata_recv, "key2", "val2"));
+  GPR_ASSERT(coalesced_message_and_eos);
+
+  gpr_free(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);
+  grpc_call_destroy(s);
+
+  cq_verifier_destroy(cqv);
+
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void packet_coalescing(grpc_end2end_test_config config) {
+  /* The test case does not support the fixture socketpair_one_byte_at_a_time */
+  if (0 == strcmp(config.name, "chttp2/socketpair_one_byte_at_a_time")) {
+    return;
+  }
+  gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
+  grpc_tracer_set_enabled("all", 1);
+  gpr_set_log_function(log_processor);
+  test_request_response_with_metadata_and_payload(config);
+  gpr_set_log_function(gpr_default_log);
+  grpc_tracer_set_enabled("all", 0);
+}
+
+void packet_coalescing_pre_init(void) {}

third_party/objective_c/Cronet/cronet_c_for_grpc.h

@@ -5,6 +5,8 @@
 #ifndef COMPONENTS_CRONET_IOS_CRONET_C_FOR_GRPC_H_
 #define COMPONENTS_CRONET_IOS_CRONET_C_FOR_GRPC_H_
 
+#define CRONET_EXPORT __attribute__((visibility("default")))
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -15,12 +17,10 @@ extern "C" {
 
 /* Opaque object representing Cronet Engine. Created and configured outside
  * of this API to facilitate sharing with other components */
-typedef struct cronet_engine { void* obj; } cronet_engine;
-
-void cronet_engine_add_quic_hint(cronet_engine* engine,
-                                 const char* host,
-                                 int port,
-                                 int alternate_port);
+typedef struct cronet_engine {
+  void* obj;
+  void* annotation;
+} cronet_engine;
 
 /* Cronet Bidirectional Stream API */
 
@@ -45,11 +45,12 @@ typedef struct cronet_bidirectional_stream_header_array {
 
 /* Set of callbacks used to receive callbacks from bidirectional stream. */
 typedef struct cronet_bidirectional_stream_callback {
-  /* Invoked when request headers are sent. Indicates that stream has initiated
-   * the request. Consumer may call cronet_bidirectional_stream_write() to start
-   * writing data.
+  /* Invoked when the stream is ready for reading and writing.
+   * Consumer may call cronet_bidirectional_stream_read() to start reading data.
+   * Consumer may call cronet_bidirectional_stream_write() to start writing
+   * data.
    */
-  void (*on_request_headers_sent)(cronet_bidirectional_stream* stream);
+  void (*on_stream_ready)(cronet_bidirectional_stream* stream);
 
   /* Invoked when initial response headers are received.
    * Consumer must call cronet_bidirectional_stream_read() to start reading.
@@ -67,20 +68,19 @@ typedef struct cronet_bidirectional_stream_callback {
    * It may be invoked after on_response_trailers_received()}, if there was
    * pending read data before trailers were received.
    *
-   * If count is 0, it means the remote side has signaled that it will send no
-   * more data; future calls to cronet_bidirectional_stream_read() will result
-   * in the on_data_read() callback or on_succeded() callback if
+   * If |bytes_read| is 0, it means the remote side has signaled that it will
+   * send no more data; future calls to cronet_bidirectional_stream_read()
+   * will result in the on_data_read() callback or on_succeded() callback if
    * cronet_bidirectional_stream_write() was invoked with end_of_stream set to
    * true.
    */
   void (*on_read_completed)(cronet_bidirectional_stream* stream,
                             char* data,
-                            int count);
+                            int bytes_read);
 
   /**
    * Invoked when all data passed to cronet_bidirectional_stream_write() is
-   * sent.
-   * To continue writing, call cronet_bidirectional_stream_write().
+   * sent. To continue writing, call cronet_bidirectional_stream_write().
    */
   void (*on_write_completed)(cronet_bidirectional_stream* stream,
                              const char* data);
@@ -117,7 +117,7 @@ typedef struct cronet_bidirectional_stream_callback {
   void (*on_canceled)(cronet_bidirectional_stream* stream);
 } cronet_bidirectional_stream_callback;
 
-/* Create a new stream object that uses |engine| and |callback|. All stream
+/* Creates a new stream object that uses |engine| and |callback|. All stream
  * tasks are performed asynchronously on the |engine| network thread. |callback|
  * methods are invoked synchronously on the |engine| network thread, but must
  * not run tasks on the current thread to prevent blocking networking operations
@@ -129,6 +129,7 @@ typedef struct cronet_bidirectional_stream_callback {
  *
  * Both |calback| and |engine| must remain valid until stream is destroyed.
  */
+CRONET_EXPORT
 cronet_bidirectional_stream* cronet_bidirectional_stream_create(
     cronet_engine* engine,
     void* annotation,
@@ -136,15 +137,40 @@ cronet_bidirectional_stream* cronet_bidirectional_stream_create(
 
 /* TBD: The following methods return int. Should it be a custom type? */
 
-/* Destroy stream object. Destroy could be called from any thread, including
+/* Destroys stream object. Destroy could be called from any thread, including
  * network thread, but is posted, so |stream| is valid until calling task is
  * complete.
  */
+CRONET_EXPORT
 int cronet_bidirectional_stream_destroy(cronet_bidirectional_stream* stream);
 
-/* Start the stream by sending request to |url| using |method| and |headers|. If
- * |end_of_stream| is true, then no data is expected to be written.
+/**
+ * Disables or enables auto flush. By default, data is flushed after
+ * every cronet_bidirectional_stream_write(). If the auto flush is disabled,
+ * the client should explicitly call cronet_bidirectional_stream_flush to flush
+ * the data.
+ */
+CRONET_EXPORT void cronet_bidirectional_stream_disable_auto_flush(
+    cronet_bidirectional_stream* stream,
+    bool disable_auto_flush);
+
+/**
+ * Delays sending request headers until cronet_bidirectional_stream_flush()
+ * is called. This flag is currently only respected when QUIC is negotiated.
+ * When true, QUIC will send request header frame along with data frame(s)
+ * as a single packet when possible.
+ */
+CRONET_EXPORT
+void cronet_bidirectional_stream_delay_request_headers_until_flush(
+    cronet_bidirectional_stream* stream,
+    bool delay_headers_until_flush);
+
+/* Starts the stream by sending request to |url| using |method| and |headers|.
+ * If |end_of_stream| is true, then no data is expected to be written. The
+ * |method| is HTTP verb, with PUT having a special meaning to mark idempotent
+ * request, which could use QUIC 0-RTT.
  */
+CRONET_EXPORT
 int cronet_bidirectional_stream_start(
     cronet_bidirectional_stream* stream,
     const char* url,
@@ -153,46 +179,61 @@ int cronet_bidirectional_stream_start(
     const cronet_bidirectional_stream_header_array* headers,
     bool end_of_stream);
 
-/* Read response data into |buffer| of |capacity| length. Must only be called at
- * most once in response to each invocation of the
- * on_response_headers_received() and on_read_completed() methods of the
- * cronet_bidirectional_stream_callback.
- * Each call will result in an invocation of one of the callback's
- * on_read_completed  method if data is read, its on_succeeded() method if
- * the stream is closed, or its on_failed() method if there's an error.
+/* Reads response data into |buffer| of |capacity| length. Must only be called
+ * at most once in response to each invocation of the
+ * on_stream_ready()/on_response_headers_received() and on_read_completed()
+ * methods of the cronet_bidirectional_stream_callback.
+ * Each call will result in an invocation of the callback's
+ * on_read_completed() method if data is read, or its on_failed() method if
+ * there's an error. The callback's on_succeeded() method is also invoked if
+ * there is no more data to read and |end_of_stream| was previously sent.
  */
+CRONET_EXPORT
 int cronet_bidirectional_stream_read(cronet_bidirectional_stream* stream,
                                      char* buffer,
                                      int capacity);
 
-/* Read response data into |buffer| of |capacity| length. Must only be called at
- * most once in response to each invocation of the
- * on_response_headers_received() and on_read_completed() methods of the
- * cronet_bidirectional_stream_callback.
- * Each call will result in an invocation of one of the callback's
- * on_read_completed  method if data is read, its on_succeeded() method if
- * the stream is closed, or its on_failed() method if there's an error.
+/* Writes request data from |buffer| of |buffer_length| length. If auto flush is
+ * disabled, data will be sent only after cronet_bidirectional_stream_flush() is
+ * called.
+ * Each call will result in an invocation the callback's on_write_completed()
+ * method if data is sent, or its on_failed() method if there's an error.
+ * The callback's on_succeeded() method is also invoked if |end_of_stream| is
+ * set and all response data has been read.
  */
+CRONET_EXPORT
 int cronet_bidirectional_stream_write(cronet_bidirectional_stream* stream,
                                       const char* buffer,
-                                      int count,
+                                      int buffer_length,
                                       bool end_of_stream);
 
+/**
+ * Flushes pending writes. This method should not be called before invocation of
+ * on_stream_ready() method of the cronet_bidirectional_stream_callback.
+ * For each previously called cronet_bidirectional_stream_write()
+ * a corresponding on_write_completed() callback will be invoked when the buffer
+ * is sent.
+ */
+CRONET_EXPORT
+void cronet_bidirectional_stream_flush(cronet_bidirectional_stream* stream);
+
 /* Cancels the stream. Can be called at any time after
  * cronet_bidirectional_stream_start(). The on_canceled() method of
  * cronet_bidirectional_stream_callback will be invoked when cancelation
  * is complete and no further callback methods will be invoked. If the
  * stream has completed or has not started, calling
  * cronet_bidirectional_stream_cancel() has no effect and on_canceled() will not
- * be  invoked. At most one callback method may be invoked after
+ * be invoked. At most one callback method may be invoked after
  * cronet_bidirectional_stream_cancel() has completed.
  */
-int cronet_bidirectional_stream_cancel(cronet_bidirectional_stream* stream);
+CRONET_EXPORT
+void cronet_bidirectional_stream_cancel(cronet_bidirectional_stream* stream);
 
 /* Returns true if the |stream| was successfully started and is now done
  * (succeeded, canceled, or failed).
  * Returns false if the |stream| stream is not yet started or is in progress.
  */
+CRONET_EXPORT
 bool cronet_bidirectional_stream_is_done(cronet_bidirectional_stream* stream);
 
 #ifdef __cplusplus