Merge branch 'master' into node_protobuf_js_6_upgrade

1 files changed, 57 insertions, 15 deletions

diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c
index d6acd392c1..d6c7aee40d 100644
--- a/src/core/lib/surface/channel.c
+++ b/src/core/lib/surface/channel.c
@@ -68,6 +68,8 @@ struct grpc_channel {
   grpc_compression_options compression_options;
   grpc_mdelem default_authority;
 
+  gpr_atm call_size_estimate;
+
   gpr_mu registered_call_mu;
   registered_call *registered_calls;
 
@@ -83,19 +85,10 @@ struct grpc_channel {
 static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg,
                             grpc_error *error);
 
-grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
-                                  const grpc_channel_args *input_args,
-                                  grpc_channel_stack_type channel_stack_type,
-                                  grpc_transport *optional_transport) {
-  grpc_channel_stack_builder *builder = grpc_channel_stack_builder_create();
-  grpc_channel_stack_builder_set_channel_arguments(exec_ctx, builder,
-                                                   input_args);
-  grpc_channel_stack_builder_set_target(builder, target);
-  grpc_channel_stack_builder_set_transport(builder, optional_transport);
-  if (!grpc_channel_init_create_stack(exec_ctx, builder, channel_stack_type)) {
-    grpc_channel_stack_builder_destroy(exec_ctx, builder);
-    return NULL;
-  }
+grpc_channel *grpc_channel_create_with_builder(
+    grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder,
+    grpc_channel_stack_type channel_stack_type) {
+  char *target = gpr_strdup(grpc_channel_stack_builder_get_target(builder));
   grpc_channel_args *args = grpc_channel_args_copy(
       grpc_channel_stack_builder_get_channel_arguments(builder));
   grpc_channel *channel;
@@ -106,15 +99,20 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
     gpr_log(GPR_ERROR, "channel stack builder failed: %s",
             grpc_error_string(error));
     GRPC_ERROR_UNREF(error);
+    gpr_free(target);
     goto done;
   }
 
   memset(channel, 0, sizeof(*channel));
-  channel->target = gpr_strdup(target);
+  channel->target = target;
   channel->is_client = grpc_channel_stack_type_is_client(channel_stack_type);
   gpr_mu_init(&channel->registered_call_mu);
   channel->registered_calls = NULL;
 
+  gpr_atm_no_barrier_store(
+      &channel->call_size_estimate,
+      (gpr_atm)CHANNEL_STACK_FROM_CHANNEL(channel)->call_stack_size);
+
   grpc_compression_options_init(&channel->compression_options);
   for (size_t i = 0; i < args->num_args; i++) {
     if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) {
@@ -177,6 +175,49 @@ done:
   return channel;
 }
 
+grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
+                                  const grpc_channel_args *input_args,
+                                  grpc_channel_stack_type channel_stack_type,
+                                  grpc_transport *optional_transport) {
+  grpc_channel_stack_builder *builder = grpc_channel_stack_builder_create();
+  grpc_channel_stack_builder_set_channel_arguments(exec_ctx, builder,
+                                                   input_args);
+  grpc_channel_stack_builder_set_target(builder, target);
+  grpc_channel_stack_builder_set_transport(builder, optional_transport);
+  if (!grpc_channel_init_create_stack(exec_ctx, builder, channel_stack_type)) {
+    grpc_channel_stack_builder_destroy(exec_ctx, builder);
+    return NULL;
+  }
+  return grpc_channel_create_with_builder(exec_ctx, builder,
+                                          channel_stack_type);
+}
+
+size_t grpc_channel_get_call_size_estimate(grpc_channel *channel) {
+#define ROUND_UP_SIZE 256
+  return ((size_t)gpr_atm_no_barrier_load(&channel->call_size_estimate) +
+          ROUND_UP_SIZE) &
+         ~(size_t)(ROUND_UP_SIZE - 1);
+}
+
+void grpc_channel_update_call_size_estimate(grpc_channel *channel,
+                                            size_t size) {
+  size_t cur = (size_t)gpr_atm_no_barrier_load(&channel->call_size_estimate);
+  if (cur < size) {
+    /* size grew: update estimate */
+    gpr_atm_no_barrier_cas(&channel->call_size_estimate, (gpr_atm)cur,
+                           (gpr_atm)size);
+    /* if we lose: never mind, something else will likely update soon enough */
+  } else if (cur == size) {
+    /* no change: holding pattern */
+  } else if (cur > 0) {
+    /* size shrank: decrease estimate */
+    gpr_atm_no_barrier_cas(
+        &channel->call_size_estimate, (gpr_atm)cur,
+        (gpr_atm)(GPR_MIN(cur - 1, (255 * cur + size) / 256)));
+    /* if we lose: never mind, something else will likely update soon enough */
+  }
+}
+
 char *grpc_channel_get_target(grpc_channel *channel) {
   GRPC_API_TRACE("grpc_channel_get_target(channel=%p)", 1, (channel));
   return gpr_strdup(channel->target);
@@ -348,7 +389,8 @@ void grpc_channel_destroy(grpc_channel *channel) {
   grpc_channel_element *elem;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   GRPC_API_TRACE("grpc_channel_destroy(channel=%p)", 1, (channel));
-  op->disconnect_with_error = GRPC_ERROR_CREATE("Channel Destroyed");
+  op->disconnect_with_error =
+      GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel Destroyed");
   elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CHANNEL(channel), 0);
   elem->filter->start_transport_op(&exec_ctx, elem, op);