Merge github.com:grpc/grpc into metadata_filter

Makefile

@@ -7288,6 +7288,8 @@ LIBEND2END_TESTS_SRC = \
     test/core/end2end/tests/simple_request.c \
     test/core/end2end/tests/streaming_error_response.c \
     test/core/end2end/tests/trailing_metadata.c \
+    test/core/end2end/tests/write_buffering.c \
+    test/core/end2end/tests/write_buffering_at_end.c \
 
 PUBLIC_HEADERS_C += \
 
@@ -7374,6 +7376,8 @@ LIBEND2END_NOSEC_TESTS_SRC = \
     test/core/end2end/tests/simple_request.c \
     test/core/end2end/tests/streaming_error_response.c \
     test/core/end2end/tests/trailing_metadata.c \
+    test/core/end2end/tests/write_buffering.c \
+    test/core/end2end/tests/write_buffering_at_end.c \
 
 PUBLIC_HEADERS_C += \
 

include/grpc/impl/codegen/grpc_types.h

@@ -179,6 +179,9 @@ typedef struct {
     Larger values give lower CPU usage for large messages, but more head of line
     blocking for small messages. */
 #define GRPC_ARG_HTTP2_MAX_FRAME_SIZE "grpc.http2.max_frame_size"
+/** How much data are we willing to queue up per stream if
+    GRPC_WRITE_BUFFER_HINT is set? This is an upper bound */
+#define GRPC_ARG_HTTP2_WRITE_BUFFER_SIZE "grpc.http2.write_buffer_size"
 /** Default authority to pass if none specified on call construction. A string.
  * */
 #define GRPC_ARG_DEFAULT_AUTHORITY "grpc.default_authority"

src/core/ext/transport/chttp2/transport/chttp2_transport.c

@@ -63,7 +63,7 @@
 #define DEFAULT_WINDOW 65535
 #define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024)
 #define MAX_WINDOW 0x7fffffffu
-
+#define MAX_WRITE_BUFFER_SIZE (64 * 1024 * 1024)
 #define DEFAULT_MAX_HEADER_LIST_SIZE (16 * 1024)
 
 #define MAX_CLIENT_STREAM_ID 0x7fffffffu
@@ -272,6 +272,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
      window -- this should by rights be 0 */
   t->force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
   t->sent_local_settings = 0;
+  t->write_buffer_size = DEFAULT_WINDOW;
 
   if (is_client) {
     grpc_slice_buffer_add(&t->outbuf, grpc_slice_from_copied_string(
@@ -322,6 +323,11 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
           grpc_chttp2_hpack_compressor_set_max_usable_size(&t->hpack_compressor,
                                                            (uint32_t)value);
         }
+      } else if (0 == strcmp(channel_args->args[i].key,
+                             GRPC_ARG_HTTP2_WRITE_BUFFER_SIZE)) {
+        t->write_buffer_size = (uint32_t)grpc_channel_arg_get_integer(
+            &channel_args->args[i],
+            (grpc_integer_options){0, 0, MAX_WRITE_BUFFER_SIZE});
       } else {
         static const struct {
           const char *channel_arg_name;
@@ -896,15 +902,22 @@ static bool contains_non_ok_status(grpc_metadata_batch *batch) {
   return false;
 }
 
+static void maybe_become_writable_due_to_send_msg(grpc_exec_ctx *exec_ctx,
+                                                  grpc_chttp2_transport *t,
+                                                  grpc_chttp2_stream *s) {
+  if (s->id != 0 && (!s->write_buffering ||
+                     s->flow_controlled_buffer.length > t->write_buffer_size)) {
+    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
+  }
+}
+
 static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx,
                                      grpc_chttp2_transport *t,
                                      grpc_chttp2_stream *s) {
   s->fetched_send_message_length +=
       (uint32_t)GRPC_SLICE_LENGTH(s->fetching_slice);
   grpc_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice);
-  if (s->id != 0) {
-    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
-  }
+  maybe_become_writable_due_to_send_msg(exec_ctx, t, s);
 }
 
 static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx,
@@ -1099,14 +1112,13 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
                                    (int64_t)len;
       s->complete_fetch_covered_by_poller = op->covered_by_poller;
       if (flags & GRPC_WRITE_BUFFER_HINT) {
-        /* allow up to 64kb to be buffered */
-        /* TODO(ctiller): make this configurable */
-        s->next_message_end_offset -= 65536;
+        s->next_message_end_offset -= t->write_buffer_size;
+        s->write_buffering = true;
+      } else {
+        s->write_buffering = false;
       }
       continue_fetching_send_locked(exec_ctx, t, s);
-      if (s->id != 0) {
-        grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
-      }
+      maybe_become_writable_due_to_send_msg(exec_ctx, t, s);
     }
   }
 
@@ -1115,6 +1127,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
     on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
     s->send_trailing_metadata_finished = add_closure_barrier(on_complete);
     s->send_trailing_metadata = op->send_trailing_metadata;
+    s->write_buffering = false;
     const size_t metadata_size =
         grpc_metadata_batch_size(op->send_trailing_metadata);
     const size_t metadata_peer_limit =

src/core/ext/transport/chttp2/transport/internal.h

@@ -249,6 +249,9 @@ struct grpc_chttp2_transport {
   int64_t announce_incoming_window;
   /** how much window would we like to have for incoming_window */
   uint32_t connection_window_target;
+  /** how much data are we willing to buffer when the WRITE_BUFFER_HINT is set?
+   */
+  uint32_t write_buffer_size;
 
   /** have we seen a goaway */
   uint8_t seen_goaway;
@@ -403,6 +406,9 @@ struct grpc_chttp2_stream {
   /** Has this stream seen an error.
       If true, then pending incoming frames can be thrown away. */
   bool seen_error;
+  /** Are we buffering writes on this stream? If yes, we won't become writable
+      until there's enough queued up in the flow_controlled_buffer */
+  bool write_buffering;
 
   /** the error that resulted in this stream being read-closed */
   grpc_error *read_closed_error;

test/core/end2end/end2end_nosec_tests.c

@@ -135,6 +135,10 @@ 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 write_buffering(grpc_end2end_test_config config);
+extern void write_buffering_pre_init(void);
+extern void write_buffering_at_end(grpc_end2end_test_config config);
+extern void write_buffering_at_end_pre_init(void);
 
 void grpc_end2end_tests_pre_init(void) {
   GPR_ASSERT(!g_pre_init_called);
@@ -185,6 +189,8 @@ void grpc_end2end_tests_pre_init(void) {
   simple_request_pre_init();
   streaming_error_response_pre_init();
   trailing_metadata_pre_init();
+  write_buffering_pre_init();
+  write_buffering_at_end_pre_init();
 }
 
 void grpc_end2end_tests(int argc, char **argv,
@@ -240,6 +246,8 @@ void grpc_end2end_tests(int argc, char **argv,
     simple_request(config);
     streaming_error_response(config);
     trailing_metadata(config);
+    write_buffering(config);
+    write_buffering_at_end(config);
     return;
   }
 
@@ -428,6 +436,14 @@ void grpc_end2end_tests(int argc, char **argv,
       trailing_metadata(config);
       continue;
     }
+    if (0 == strcmp("write_buffering", argv[i])) {
+      write_buffering(config);
+      continue;
+    }
+    if (0 == strcmp("write_buffering_at_end", argv[i])) {
+      write_buffering_at_end(config);
+      continue;
+    }
     gpr_log(GPR_DEBUG, "not a test: '%s'", argv[i]);
     abort();
   }

test/core/end2end/end2end_tests.c

@@ -137,6 +137,10 @@ 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 write_buffering(grpc_end2end_test_config config);
+extern void write_buffering_pre_init(void);
+extern void write_buffering_at_end(grpc_end2end_test_config config);
+extern void write_buffering_at_end_pre_init(void);
 
 void grpc_end2end_tests_pre_init(void) {
   GPR_ASSERT(!g_pre_init_called);
@@ -188,6 +192,8 @@ void grpc_end2end_tests_pre_init(void) {
   simple_request_pre_init();
   streaming_error_response_pre_init();
   trailing_metadata_pre_init();
+  write_buffering_pre_init();
+  write_buffering_at_end_pre_init();
 }
 
 void grpc_end2end_tests(int argc, char **argv,
@@ -244,6 +250,8 @@ void grpc_end2end_tests(int argc, char **argv,
     simple_request(config);
     streaming_error_response(config);
     trailing_metadata(config);
+    write_buffering(config);
+    write_buffering_at_end(config);
     return;
   }
 
@@ -436,6 +444,14 @@ void grpc_end2end_tests(int argc, char **argv,
       trailing_metadata(config);
       continue;
     }
+    if (0 == strcmp("write_buffering", argv[i])) {
+      write_buffering(config);
+      continue;
+    }
+    if (0 == strcmp("write_buffering_at_end", argv[i])) {
+      write_buffering_at_end(config);
+      continue;
+    }
     gpr_log(GPR_DEBUG, "not a test: '%s'", argv[i]);
     abort();
   }

test/core/end2end/gen_build_yaml.py

@@ -143,6 +143,8 @@ END2END_TESTS = {
     '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,
 }
 
 

test/core/end2end/generate_tests.bzl

@@ -131,6 +131,8 @@ END2END_TESTS = {
     'trailing_metadata': test_options(),
     'authority_not_supported': test_options(),
     'filter_latency': test_options(),
+    'write_buffering': test_options(),
+    'write_buffering_at_end': test_options(),
 }
 
 

test/core/end2end/tests/write_buffering.c

@@ -0,0 +1,291 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#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"
+
+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);
+}
+
+/* Client sends a request with payload, server reads then returns status. */
+static void test_invoke_request_with_payload(grpc_end2end_test_config config) {
+  grpc_call *c;
+  grpc_call *s;
+  grpc_slice request_payload_slice1 =
+      grpc_slice_from_copied_string("hello world");
+  grpc_byte_buffer *request_payload1 =
+      grpc_raw_byte_buffer_create(&request_payload_slice1, 1);
+  grpc_slice request_payload_slice2 = grpc_slice_from_copied_string("abc123");
+  grpc_byte_buffer *request_payload2 =
+      grpc_raw_byte_buffer_create(&request_payload_slice2, 1);
+  gpr_timespec deadline = five_seconds_time();
+  grpc_end2end_test_fixture f =
+      begin_test(config, "test_invoke_request_with_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_recv1 = NULL;
+  grpc_byte_buffer *request_payload_recv2 = 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;
+
+  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 = 0;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata = &initial_metadata_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(
+                                 f.server, &s, &call_details,
+                                 &request_metadata_recv, f.cq, f.cq, tag(101)));
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true); /* send message is buffered */
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message = request_payload1;
+  op->flags = GRPC_WRITE_BUFFER_HINT;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  /* recv message should not succeed yet - it's buffered at the client still */
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &request_payload_recv1;
+  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(2), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  /* send another message, this time not buffered */
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message = request_payload2;
+  op->flags = 0;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  /* now the first send should match up with the first recv */
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+  cq_verify(cqv);
+
+  /* and the next recv should be ready immediately also */
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &request_payload_recv2;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  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(4), NULL);
+
+  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(105), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(105), 1);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), 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_recv1, "hello world"));
+  GPR_ASSERT(byte_buffer_eq_string(request_payload_recv2, "abc123"));
+
+  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_payload1);
+  grpc_byte_buffer_destroy(request_payload_recv1);
+  grpc_byte_buffer_destroy(request_payload2);
+  grpc_byte_buffer_destroy(request_payload_recv2);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void write_buffering(grpc_end2end_test_config config) {
+  test_invoke_request_with_payload(config);
+}
+
+void write_buffering_pre_init(void) {}

test/core/end2end/tests/write_buffering_at_end.c

@@ -0,0 +1,280 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#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"
+
+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);
+}
+
+/* Client sends a request with payload, server reads then returns status. */
+static void test_invoke_request_with_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_end2end_test_fixture f =
+      begin_test(config, "test_invoke_request_with_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_recv1 = NULL;
+  grpc_byte_buffer *request_payload_recv2 = 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;
+
+  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 = 0;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata = &initial_metadata_recv;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(
+                                 f.server, &s, &call_details,
+                                 &request_metadata_recv, f.cq, f.cq, tag(101)));
+  CQ_EXPECT_COMPLETION(cqv, tag(1), true); /* send message is buffered */
+  CQ_EXPECT_COMPLETION(cqv, tag(101), true);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message = request_payload;
+  op->flags = GRPC_WRITE_BUFFER_HINT;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  /* recv message should not succeed yet - it's buffered at the client still */
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &request_payload_recv1;
+  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(2), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(3), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(102), true);
+  cq_verify(cqv);
+
+  /* send end of stream: should release the buffering */
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op++;
+  error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  /* now the first send should match up with the first recv */
+  CQ_EXPECT_COMPLETION(cqv, tag(103), true);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), true);
+  cq_verify(cqv);
+
+  /* and the next recv should be ready immediately also (and empty) */
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message = &request_payload_recv2;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(104), true);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  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(4), NULL);
+
+  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(105), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(105), 1);
+  CQ_EXPECT_COMPLETION(cqv, tag(4), 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_recv1, "hello world"));
+  GPR_ASSERT(request_payload_recv2 == NULL);
+
+  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_recv1);
+
+  end_test(&f);
+  config.tear_down_data(&f);
+}
+
+void write_buffering_at_end(grpc_end2end_test_config config) {
+  test_invoke_request_with_payload(config);
+}
+
+void write_buffering_at_end_pre_init(void) {}

tools/run_tests/generated/sources_and_headers.json

@@ -6458,7 +6458,9 @@
       "test/core/end2end/tests/simple_metadata.c", 
       "test/core/end2end/tests/simple_request.c", 
       "test/core/end2end/tests/streaming_error_response.c", 
-      "test/core/end2end/tests/trailing_metadata.c"
+      "test/core/end2end/tests/trailing_metadata.c", 
+      "test/core/end2end/tests/write_buffering.c", 
+      "test/core/end2end/tests/write_buffering_at_end.c"
     ], 
     "third_party": false, 
     "type": "lib"
@@ -6527,7 +6529,9 @@
       "test/core/end2end/tests/simple_metadata.c", 
       "test/core/end2end/tests/simple_request.c", 
       "test/core/end2end/tests/streaming_error_response.c", 
-      "test/core/end2end/tests/trailing_metadata.c"
+      "test/core/end2end/tests/trailing_metadata.c", 
+      "test/core/end2end/tests/write_buffering.c", 
+      "test/core/end2end/tests/write_buffering_at_end.c"
     ], 
     "third_party": false, 
     "type": "lib"

vsprojects/vcxproj/test/end2end/tests/end2end_nosec_tests/end2end_nosec_tests.vcxproj

@@ -247,6 +247,10 @@
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\trailing_metadata.c">
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering.c">
+    </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering_at_end.c">
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="$(SolutionDir)\..\vsprojects\vcxproj\.\grpc_test_util_unsecure\grpc_test_util_unsecure.vcxproj">

vsprojects/vcxproj/test/end2end/tests/end2end_nosec_tests/end2end_nosec_tests.vcxproj.filters

@@ -145,6 +145,12 @@
     <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\trailing_metadata.c">
       <Filter>test\core\end2end\tests</Filter>
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering.c">
+      <Filter>test\core\end2end\tests</Filter>
+    </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering_at_end.c">
+      <Filter>test\core\end2end\tests</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ClInclude Include="$(SolutionDir)\..\test\core\end2end\tests\cancel_test_helpers.h">

vsprojects/vcxproj/test/end2end/tests/end2end_tests/end2end_tests.vcxproj

@@ -249,6 +249,10 @@
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\trailing_metadata.c">
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering.c">
+    </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering_at_end.c">
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="$(SolutionDir)\..\vsprojects\vcxproj\.\grpc_test_util\grpc_test_util.vcxproj">

vsprojects/vcxproj/test/end2end/tests/end2end_tests/end2end_tests.vcxproj.filters

@@ -148,6 +148,12 @@
     <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\trailing_metadata.c">
       <Filter>test\core\end2end\tests</Filter>
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering.c">
+      <Filter>test\core\end2end\tests</Filter>
+    </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\core\end2end\tests\write_buffering_at_end.c">
+      <Filter>test\core\end2end\tests</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ClInclude Include="$(SolutionDir)\..\test\core\end2end\tests\cancel_test_helpers.h">