Merge branch 'master' of github.com:grpc/grpc into microchannels

5 files changed, 129 insertions, 17 deletions

diff --git a/src/core/surface/alarm.c b/src/core/surface/alarm.c
new file mode 100644
index 0000000000..7c47dd56f8
--- /dev/null
+++ b/src/core/surface/alarm.c
@@ -0,0 +1,83 @@
+/*
+ *
+ * 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 "src/core/iomgr/timer.h"
+#include "src/core/surface/completion_queue.h"
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+
+struct grpc_alarm {
+  grpc_timer alarm;
+  grpc_cq_completion completion;
+  /** completion queue where events about this alarm will be posted */
+  grpc_completion_queue *cq;
+  /** user supplied tag */
+  void *tag;
+};
+
+static void do_nothing_end_completion(grpc_exec_ctx *exec_ctx, void *arg,
+                                      grpc_cq_completion *c) {}
+
+static void alarm_cb(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+  grpc_alarm *alarm = arg;
+  grpc_cq_end_op(exec_ctx, alarm->cq, alarm->tag, success,
+                 do_nothing_end_completion, NULL, &alarm->completion);
+}
+
+grpc_alarm *grpc_alarm_create(grpc_completion_queue *cq, gpr_timespec deadline,
+                              void *tag) {
+  grpc_alarm *alarm = gpr_malloc(sizeof(grpc_alarm));
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+
+  GRPC_CQ_INTERNAL_REF(cq, "alarm");
+  alarm->cq = cq;
+  alarm->tag = tag;
+
+  grpc_timer_init(&exec_ctx, &alarm->alarm, deadline, alarm_cb, alarm,
+                  gpr_now(GPR_CLOCK_MONOTONIC));
+  grpc_cq_begin_op(cq);
+  grpc_exec_ctx_finish(&exec_ctx);
+  return alarm;
+}
+
+void grpc_alarm_cancel(grpc_alarm *alarm) {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_timer_cancel(&exec_ctx, &alarm->alarm);
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
+void grpc_alarm_destroy(grpc_alarm *alarm) {
+  grpc_alarm_cancel(alarm);
+  GRPC_CQ_INTERNAL_UNREF(alarm->cq, "alarm");
+  gpr_free(alarm);
+}
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index b40e74d61b..81ff215c0c 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -42,7 +42,7 @@
 #include <grpc/support/useful.h>
 
 #include "src/core/channel/channel_stack.h"
-#include "src/core/iomgr/alarm.h"
+#include "src/core/iomgr/timer.h"
 #include "src/core/profiling/timers.h"
 #include "src/core/support/string.h"
 #include "src/core/surface/api_trace.h"
@@ -237,7 +237,7 @@ struct grpc_call {
   grpc_call_context_element context[GRPC_CONTEXT_COUNT];
 
   /* Deadline alarm - if have_alarm is non-zero */
-  grpc_alarm alarm;
+  grpc_timer alarm;
 
   /* Call refcount - to keep the call alive during asynchronous operations */
   gpr_refcount internal_refcount;
@@ -539,12 +539,24 @@ grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
   return algorithm;
 }
 
-static void set_encodings_accepted_by_peer(
-    grpc_call *call, const gpr_slice accept_encoding_slice) {
+static void destroy_encodings_accepted_by_peer(void *p) { return; }
+
+static void set_encodings_accepted_by_peer(grpc_call *call, grpc_mdelem *mdel) {
   size_t i;
   grpc_compression_algorithm algorithm;
   gpr_slice_buffer accept_encoding_parts;
+  gpr_slice accept_encoding_slice;
+  void *accepted_user_data;
+
+  accepted_user_data =
+      grpc_mdelem_get_user_data(mdel, destroy_encodings_accepted_by_peer);
+  if (accepted_user_data != NULL) {
+    call->encodings_accepted_by_peer =
+        (gpr_uint32)(((gpr_uintptr)accepted_user_data) - 1);
+    return;
+  }
 
+  accept_encoding_slice = mdel->value->slice;
   gpr_slice_buffer_init(&accept_encoding_parts);
   gpr_slice_split(accept_encoding_slice, ",", &accept_encoding_parts);
 
@@ -568,6 +580,12 @@ static void set_encodings_accepted_by_peer(
       gpr_free(accept_encoding_entry_str);
     }
   }
+
+  gpr_slice_buffer_destroy(&accept_encoding_parts);
+
+  grpc_mdelem_set_user_data(
+      mdel, destroy_encodings_accepted_by_peer,
+      (void *)(((gpr_uintptr)call->encodings_accepted_by_peer) + 1));
 }
 
 gpr_uint32 grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) {
@@ -1027,7 +1045,7 @@ static void call_on_done_recv(grpc_exec_ctx *exec_ctx, void *pc, int success) {
       GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED);
       call->read_state = READ_STATE_STREAM_CLOSED;
       if (call->have_alarm) {
-        grpc_alarm_cancel(exec_ctx, &call->alarm);
+        grpc_timer_cancel(exec_ctx, &call->alarm);
       }
       /* propagate cancellation to any interested children */
       child_call = call->first_child;
@@ -1345,7 +1363,7 @@ void grpc_call_destroy(grpc_call *c) {
   GPR_ASSERT(!c->destroy_called);
   c->destroy_called = 1;
   if (c->have_alarm) {
-    grpc_alarm_cancel(&exec_ctx, &c->alarm);
+    grpc_timer_cancel(&exec_ctx, &c->alarm);
   }
   cancel = c->read_state != READ_STATE_STREAM_CLOSED;
   unlock(&exec_ctx, c);
@@ -1470,7 +1488,7 @@ static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call,
   GRPC_CALL_INTERNAL_REF(call, "alarm");
   call->have_alarm = 1;
   call->send_deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
-  grpc_alarm_init(exec_ctx, &call->alarm, call->send_deadline, call_alarm, call,
+  grpc_timer_init(exec_ctx, &call->alarm, call->send_deadline, call_alarm, call,
                   gpr_now(GPR_CLOCK_MONOTONIC));
 }
 
@@ -1528,7 +1546,6 @@ static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call,
   grpc_metadata_array *dest;
   grpc_metadata *mdusr;
   int is_trailing;
-  grpc_mdctx *mdctx = call->metadata_context;
 
   is_trailing = call->read_state >= READ_STATE_GOT_INITIAL_METADATA;
   for (l = md->list.head; l != NULL; l = l->next) {
@@ -1550,7 +1567,7 @@ static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call,
     } else if (key == grpc_channel_get_encodings_accepted_by_peer_string(
                           call->channel)) {
       GPR_TIMER_BEGIN("encodings_accepted_by_peer", 0);
-      set_encodings_accepted_by_peer(call, mdel->value->slice);
+      set_encodings_accepted_by_peer(call, mdel);
       GPR_TIMER_END("encodings_accepted_by_peer", 0);
     } else {
       GPR_TIMER_BEGIN("report_up", 0);
@@ -1588,14 +1605,12 @@ static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call,
     call->read_state = READ_STATE_GOT_INITIAL_METADATA;
   }
 
-  grpc_mdctx_lock(mdctx);
   for (l = md->list.head; l; l = l->next) {
-    if (l->md) GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
+    if (l->md) GRPC_MDELEM_UNREF(l->md);
   }
   for (l = md->garbage.head; l; l = l->next) {
-    GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
+    GRPC_MDELEM_UNREF(l->md);
   }
-  grpc_mdctx_unlock(mdctx);
 }
 
 grpc_call_stack *grpc_call_get_call_stack(grpc_call *call) {
diff --git a/src/core/surface/channel_connectivity.c b/src/core/surface/channel_connectivity.c
index 52a64fb2cd..df2774b527 100644
--- a/src/core/surface/channel_connectivity.c
+++ b/src/core/surface/channel_connectivity.c
@@ -38,7 +38,7 @@
 
 #include "src/core/channel/client_channel.h"
 #include "src/core/channel/client_uchannel.h"
-#include "src/core/iomgr/alarm.h"
+#include "src/core/iomgr/timer.h"
 #include "src/core/surface/api_trace.h"
 #include "src/core/surface/completion_queue.h"
 
@@ -85,7 +85,7 @@ typedef struct {
   int success;
   int removed;
   grpc_closure on_complete;
-  grpc_alarm alarm;
+  grpc_timer alarm;
   grpc_connectivity_state state;
   grpc_completion_queue *cq;
   grpc_cq_completion completion_storage;
@@ -155,7 +155,7 @@ static void partly_done(grpc_exec_ctx *exec_ctx, state_watcher *w,
     gpr_mu_lock(&w->mu);
     w->success = 1;
     gpr_mu_unlock(&w->mu);
-    grpc_alarm_cancel(exec_ctx, &w->alarm);
+    grpc_timer_cancel(exec_ctx, &w->alarm);
   }
 
   gpr_mu_lock(&w->mu);
@@ -217,7 +217,7 @@ void grpc_channel_watch_connectivity_state(
   w->tag = tag;
   w->channel = channel;
 
-  grpc_alarm_init(&exec_ctx, &w->alarm,
+  grpc_timer_init(&exec_ctx, &w->alarm,
                   gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
                   timeout_complete, w, gpr_now(GPR_CLOCK_MONOTONIC));
 
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index bcdb363873..aa90b3f7f5 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -36,6 +36,7 @@
 #include <stdio.h>
 #include <string.h>
 
+#include "src/core/iomgr/timer.h"
 #include "src/core/iomgr/pollset.h"
 #include "src/core/support/string.h"
 #include "src/core/surface/api_trace.h"
@@ -46,6 +47,7 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/atm.h>
 #include <grpc/support/log.h>
+#include <grpc/support/time.h>
 
 typedef struct {
   grpc_pollset_worker *worker;
@@ -72,6 +74,15 @@ struct grpc_completion_queue {
   grpc_closure pollset_destroy_done;
 };
 
+struct grpc_cq_alarm {
+  grpc_timer alarm;
+  grpc_cq_completion completion;
+  /** completion queue where events about this alarm will be posted */
+  grpc_completion_queue *cq;
+  /** user supplied tag */
+  void *tag;
+};
+
 static void on_pollset_destroy_done(grpc_exec_ctx *exec_ctx, void *cc,
                                     int success);
 
diff --git a/src/core/surface/init.c b/src/core/surface/init.c
index 715c90a5e1..b2e66a830e 100644
--- a/src/core/surface/init.c
+++ b/src/core/surface/init.c
@@ -47,6 +47,7 @@
 #include "src/core/client_config/resolvers/dns_resolver.h"
 #include "src/core/client_config/resolvers/sockaddr_resolver.h"
 #include "src/core/debug/trace.h"
+#include "src/core/iomgr/executor.h"
 #include "src/core/iomgr/iomgr.h"
 #include "src/core/profiling/timers.h"
 #include "src/core/surface/api_trace.h"
@@ -108,6 +109,7 @@ void grpc_init(void) {
     grpc_register_tracer("connectivity_state", &grpc_connectivity_state_trace);
     grpc_security_pre_init();
     grpc_iomgr_init();
+    grpc_executor_init();
     grpc_tracer_init("GRPC_TRACE");
     /* Only initialize census if noone else has. */
     if (census_enabled() == CENSUS_FEATURE_NONE) {
@@ -132,6 +134,7 @@ void grpc_shutdown(void) {
   gpr_mu_lock(&g_init_mu);
   if (--g_initializations == 0) {
     grpc_iomgr_shutdown();
+    grpc_executor_shutdown();
     census_shutdown();
     gpr_timers_global_destroy();
     grpc_tracer_shutdown();