Initial channel watching

3 files changed, 42 insertions, 19 deletions

diff --git a/test/core/end2end/fuzzers/api_fuzzer.c b/test/core/end2end/fuzzers/api_fuzzer.c
index 64fb310a59..d8385264ea 100644
--- a/test/core/end2end/fuzzers/api_fuzzer.c
+++ b/test/core/end2end/fuzzers/api_fuzzer.c
@@ -38,13 +38,13 @@
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 
+#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 #include "src/core/lib/iomgr/tcp_client.h"
 #include "src/core/lib/iomgr/timer.h"
-#include "src/core/lib/transport/metadata.h"
-#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
 #include "src/core/lib/surface/server.h"
+#include "src/core/lib/transport/metadata.h"
 #include "test/core/util/passthru_endpoint.h"
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -190,7 +190,8 @@ extern void (*grpc_tcp_client_connect_impl)(
     grpc_pollset_set *interested_parties, const struct sockaddr *addr,
     size_t addr_len, gpr_timespec deadline);
 
-static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure, grpc_endpoint **ep, gpr_timespec deadline);
+static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
+                          grpc_endpoint **ep, gpr_timespec deadline);
 
 typedef struct {
   grpc_timer timer;
@@ -222,13 +223,14 @@ static void do_connect(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
   gpr_free(fc);
 }
 
-static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure, grpc_endpoint **ep, gpr_timespec deadline) {
+static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
+                          grpc_endpoint **ep, gpr_timespec deadline) {
   if (gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) <= 0) {
     *ep = NULL;
     grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
     return;
   }
-  
+
   future_connect *fc = gpr_malloc(sizeof(*fc));
   fc->closure = closure;
   fc->ep = ep;
@@ -252,6 +254,7 @@ static void my_tcp_client_connect(grpc_exec_ctx *exec_ctx,
 
 typedef enum {
   SERVER_SHUTDOWN,
+  CHANNEL_WATCH,
 } tag_name;
 
 static void *tag(tag_name name) { return (void *)(uintptr_t)name; }
@@ -270,10 +273,12 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
 
   bool server_shutdown = false;
   int pending_server_shutdowns = 0;
+  int pending_channel_watches = 0;
 
   grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
 
-  while (!is_eof(&inp) || g_channel != NULL || g_server != NULL) {
+  while (!is_eof(&inp) || g_channel != NULL || g_server != NULL ||
+         pending_channel_watches > 0) {
     if (is_eof(&inp)) {
       if (g_channel != NULL) {
         grpc_channel_destroy(g_channel);
@@ -309,6 +314,10 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
                 GPR_ASSERT(pending_server_shutdowns);
                 pending_server_shutdowns--;
                 break;
+              case CHANNEL_WATCH:
+                GPR_ASSERT(pending_channel_watches > 0);
+                pending_channel_watches--;
+                break;
               default:
                 GPR_ASSERT(false);
             }
@@ -396,6 +405,22 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
         }
         break;
       }
+      // watch connectivity
+      case 9: {
+        if (g_channel != NULL) {
+          grpc_connectivity_state st =
+              grpc_channel_check_connectivity_state(g_channel, 0);
+          if (st != GRPC_CHANNEL_FATAL_FAILURE) {
+            grpc_channel_watch_connectivity_state(
+                g_channel, st,
+                gpr_time_add(
+                    gpr_now(GPR_CLOCK_REALTIME),
+                    gpr_time_from_micros(read_uint32(&inp), GPR_TIMESPAN)), cq,
+                    tag(CHANNEL_WATCH));
+            pending_channel_watches++;
+          }
+        }
+      }
     }
   }
 
diff --git a/test/core/util/passthru_endpoint.c b/test/core/util/passthru_endpoint.c
index a47baeb637..cec8865744 100644
--- a/test/core/util/passthru_endpoint.c
+++ b/test/core/util/passthru_endpoint.c
@@ -60,8 +60,7 @@ static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   gpr_mu_lock(&m->parent->mu);
   if (m->parent->shutdown) {
     grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
-  } else
-  if (m->read_buffer.count > 0) {
+  } else if (m->read_buffer.count > 0) {
     gpr_slice_buffer_swap(&m->read_buffer, slices);
     grpc_exec_ctx_enqueue(exec_ctx, cb, true, NULL);
   } else {
@@ -72,8 +71,7 @@ static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
 }
 
 static half *other_half(half *h) {
-  if (h == &h->parent->client)
-    return &h->parent->server;
+  if (h == &h->parent->client) return &h->parent->server;
   return &h->parent->client;
 }
 
@@ -81,11 +79,10 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                      gpr_slice_buffer *slices, grpc_closure *cb) {
   half *m = other_half((half *)ep);
   gpr_mu_lock(&m->parent->mu);
-  bool ok= true;
+  bool ok = true;
   if (m->parent->shutdown) {
-   ok = false; 
-  }
-  else if (m->on_read != NULL) {
+    ok = false;
+  } else if (m->on_read != NULL) {
     gpr_slice_buffer_addn(m->on_read_out, slices->slices, slices->count);
     grpc_exec_ctx_enqueue(exec_ctx, m->on_read, true, NULL);
     m->on_read = NULL;
@@ -103,7 +100,7 @@ static void me_add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                                   grpc_pollset_set *pollset) {}
 
 static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
-  half *m = (half*)ep;
+  half *m = (half *)ep;
   gpr_mu_lock(&m->parent->mu);
   m->parent->shutdown = true;
   if (m->on_read) {
@@ -119,7 +116,7 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
 }
 
 static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
-  passthru_endpoint *p = ((half*)ep)->parent;
+  passthru_endpoint *p = ((half *)ep)->parent;
   gpr_mu_lock(&p->mu);
   if (0 == --p->halves) {
     gpr_mu_unlock(&p->mu);
@@ -147,7 +144,8 @@ static void half_init(half *m) {
   m->on_read = NULL;
 }
 
-void grpc_passthru_endpoint_create(grpc_endpoint **client, grpc_endpoint **server) {
+void grpc_passthru_endpoint_create(grpc_endpoint **client,
+                                   grpc_endpoint **server) {
   passthru_endpoint *m = gpr_malloc(sizeof(*m));
   half_init(&m->client);
   half_init(&m->server);
@@ -155,4 +153,3 @@ void grpc_passthru_endpoint_create(grpc_endpoint **client, grpc_endpoint **serve
   *client = &m->client.base;
   *server = &m->server.base;
 }
-
diff --git a/test/core/util/passthru_endpoint.h b/test/core/util/passthru_endpoint.h
index ba075d508a..aa1d3a1763 100644
--- a/test/core/util/passthru_endpoint.h
+++ b/test/core/util/passthru_endpoint.h
@@ -36,6 +36,7 @@
 
 #include "src/core/lib/iomgr/endpoint.h"
 
-void grpc_passthru_endpoint_create(grpc_endpoint **client, grpc_endpoint **server);
+void grpc_passthru_endpoint_create(grpc_endpoint **client,
+                                   grpc_endpoint **server);
 
 #endif