Ping API

10 files changed, 76 insertions, 18 deletions

diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 0212f4f2c8..c2a6648f18 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -252,7 +252,9 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
   grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 1);
 
   GPR_ASSERT(op->set_accept_stream == NULL);
-  GPR_ASSERT(op->bind_pollset == NULL || op->send_ping != NULL);
+  if (op->bind_pollset != NULL) {
+    grpc_pollset_set_add_pollset(exec_ctx, &chand->interested_parties, op->bind_pollset);
+  }
 
   gpr_mu_lock(&chand->mu_config);
   if (op->on_connectivity_state_change != NULL) {
@@ -267,7 +269,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
     if (chand->lb_policy == NULL) {
       grpc_exec_ctx_enqueue(exec_ctx, op->send_ping, 0);
     } else {
-      grpc_lb_policy_ping_one(exec_ctx, chand->lb_policy, op->bind_pollset, op->send_ping);
+      grpc_lb_policy_ping_one(exec_ctx, chand->lb_policy, op->send_ping);
       op->bind_pollset = NULL;
     }
     op->send_ping = NULL;
diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c
index 8a8b60ba39..7dad7f7cd4 100644
--- a/src/core/client_config/lb_policies/pick_first.c
+++ b/src/core/client_config/lb_policies/pick_first.c
@@ -348,6 +348,17 @@ void pf_notify_on_state_change(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
   gpr_mu_unlock(&p->mu);
 }
 
+void pf_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_closure *closure) {
+  pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
+  gpr_mu_lock(&p->mu);
+  if (p->selected) {
+    grpc_connected_subchannel_ping(exec_ctx, p->selected, closure);
+  } else {
+    grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+  }
+  gpr_mu_unlock(&p->mu);
+}
+
 static const grpc_lb_policy_vtable pick_first_lb_policy_vtable = {
     pf_destroy, pf_shutdown, pf_pick, pf_cancel_pick, pf_ping_one, pf_exit_idle,
     pf_check_connectivity, pf_notify_on_state_change};
diff --git a/src/core/client_config/lb_policies/round_robin.c b/src/core/client_config/lb_policies/round_robin.c
index b86dba20ee..d487456363 100644
--- a/src/core/client_config/lb_policies/round_robin.c
+++ b/src/core/client_config/lb_policies/round_robin.c
@@ -467,8 +467,24 @@ static void rr_notify_on_state_change(grpc_exec_ctx *exec_ctx,
   gpr_mu_unlock(&p->mu);
 }
 
+static void rr_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
+                        grpc_closure *closure) {
+  round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
+  ready_list *selected;
+  grpc_connected_subchannel *target;
+  gpr_mu_lock(&p->mu);
+  if ((selected = peek_next_connected_locked(p))) {
+    gpr_mu_unlock(&p->mu);
+    target = grpc_subchannel_get_connected_subchannel(selected->subchannel);
+    grpc_connected_subchannel_ping(exec_ctx, target, closure);
+  } else {
+    gpr_mu_unlock(&p->mu);
+    grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+  }
+}
+
 static const grpc_lb_policy_vtable round_robin_lb_policy_vtable = {
-    rr_destroy, rr_shutdown, rr_pick, rr_cancel_pick, rr_exit_idle,
+    rr_destroy, rr_shutdown, rr_pick, rr_cancel_pick, rr_ping_one, rr_exit_idle,
     rr_check_connectivity, rr_notify_on_state_change};
 
 static void round_robin_factory_ref(grpc_lb_policy_factory *factory) {}
diff --git a/src/core/client_config/lb_policy.c b/src/core/client_config/lb_policy.c
index d254161546..3230434966 100644
--- a/src/core/client_config/lb_policy.c
+++ b/src/core/client_config/lb_policy.c
@@ -116,6 +116,10 @@ void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
   policy->vtable->exit_idle(exec_ctx, policy);
 }
 
+void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_closure *closure) {
+  policy->vtable->ping_one(exec_ctx, policy, closure);
+}
+
 void grpc_lb_policy_notify_on_state_change(grpc_exec_ctx *exec_ctx,
                                            grpc_lb_policy *policy,
                                            grpc_connectivity_state *state,
diff --git a/src/core/client_config/lb_policy.h b/src/core/client_config/lb_policy.h
index 01c57a24ff..c194bf11fe 100644
--- a/src/core/client_config/lb_policy.h
+++ b/src/core/client_config/lb_policy.h
@@ -63,7 +63,7 @@ struct grpc_lb_policy_vtable {
   void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                       grpc_connected_subchannel **target);
 
-  void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_pollset *pollset, grpc_closure *closure);
+  void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_closure *closure);
 
   /** try to enter a READY connectivity state */
   void (*exit_idle)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
@@ -123,7 +123,7 @@ int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                         grpc_connected_subchannel **target,
                         grpc_closure *on_complete);
 
-void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_pollset *pollset, grpc_closure *closure);
+void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_closure *closure);
 
 void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
                                 grpc_connected_subchannel **target);
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 6631e9bae2..ce4919a8cb 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -461,6 +461,17 @@ void grpc_connected_subchannel_notify_on_state_change(
                                 closure);
 }
 
+void grpc_connected_subchannel_ping(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
+    grpc_closure *closure) {
+  grpc_transport_op op;
+  grpc_channel_element *elem;
+  memset(&op, 0, sizeof(op));
+  op.send_ping = closure;
+  elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
+  elem->filter->start_transport_op(exec_ctx, elem, &op);
+}
+
 static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
   size_t channel_stack_size;
   grpc_connected_subchannel *con;
diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h
index 74ebcecfba..c28c3a86c9 100644
--- a/src/core/client_config/subchannel.h
+++ b/src/core/client_config/subchannel.h
@@ -124,6 +124,9 @@ void grpc_connected_subchannel_notify_on_state_change(
     grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *channel,
     grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
     grpc_closure *notify);
+void grpc_connected_subchannel_ping(
+    grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *channel,
+    grpc_closure *notify);
 
 /** retrieve the grpc_connected_subchannel - or NULL if called before
     the subchannel becomes connected */
diff --git a/src/core/transport/chttp2/frame_ping.c b/src/core/transport/chttp2/frame_ping.c
index 4d2c54269d..8e763278ff 100644
--- a/src/core/transport/chttp2/frame_ping.c
+++ b/src/core/transport/chttp2/frame_ping.c
@@ -76,7 +76,6 @@ grpc_chttp2_parse_error grpc_chttp2_ping_parser_parse(
   gpr_uint8 *const end = GPR_SLICE_END_PTR(slice);
   gpr_uint8 *cur = beg;
   grpc_chttp2_ping_parser *p = parser;
-  grpc_chttp2_outstanding_ping *ping;
 
   while (p->byte != 8 && cur != end) {
     p->opaque_8bytes[p->byte] = *cur;
@@ -87,15 +86,7 @@ grpc_chttp2_parse_error grpc_chttp2_ping_parser_parse(
   if (p->byte == 8) {
     GPR_ASSERT(is_last);
     if (p->is_ack) {
-      for (ping = transport_parsing->pings.next;
-           ping != &transport_parsing->pings; ping = ping->next) {
-        if (0 == memcmp(p->opaque_8bytes, ping->id, 8)) {
-          grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, 1);
-        }
-        ping->next->prev = ping->prev;
-        ping->prev->next = ping->next;
-        gpr_free(ping);
-      }
+      grpc_chttp2_ack_ping(exec_ctx, transport_parsing, p->opaque_8bytes);
     } else {
       gpr_slice_buffer_add(&transport_parsing->qbuf,
                            grpc_chttp2_ping_create(1, p->opaque_8bytes));
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index 45d2599cdc..fc35ea6f93 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -283,9 +283,6 @@ struct grpc_chttp2_transport_parsing {
   gpr_slice goaway_text;
 
   gpr_int64 outgoing_window;
-
-  /** pings awaiting responses */
-  grpc_chttp2_outstanding_ping pings;
 };
 
 struct grpc_chttp2_transport {
@@ -747,4 +744,8 @@ void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,
 void grpc_chttp2_incoming_byte_stream_finished(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs);
 
+void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
+                          grpc_chttp2_transport_parsing *parsing,
+                          const gpr_uint8 *opaque_8bytes);
+
 #endif
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 6ba9db8348..661e80d3a4 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -901,6 +901,25 @@ static void send_ping_locked(grpc_chttp2_transport *t, grpc_closure *on_recv) {
   gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id));
 }
 
+void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, 
+                                 grpc_chttp2_transport_parsing *transport_parsing,
+                                 const gpr_uint8 *opaque_8bytes) {
+  grpc_chttp2_outstanding_ping *ping;
+  grpc_chttp2_transport *t = TRANSPORT_FROM_PARSING(transport_parsing);
+  grpc_chttp2_transport_global *transport_global = &t->global;
+  lock(t);
+  for (ping = transport_global->pings.next;
+       ping != &transport_global->pings; ping = ping->next) {
+    if (0 == memcmp(opaque_8bytes, ping->id, 8)) {
+      grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, 1);
+    }
+    ping->next->prev = ping->prev;
+    ping->prev->next = ping->next;
+    gpr_free(ping);
+  }
+  unlock(exec_ctx, t);
+}
+
 static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                                  grpc_transport_op *op) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;