Merge branch 'fix-robbies-thing' into direct-calls

src/core/lib/iomgr/ev_epoll_linux.c

@@ -297,8 +297,8 @@ static void pi_add_ref(polling_island *pi);
 static void pi_unref(grpc_exec_ctx *exec_ctx, polling_island *pi);
 
 #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG
-static void pi_add_ref_dbg(polling_island *pi, char *reason, char *file,
-                           int line) {
+static void pi_add_ref_dbg(polling_island *pi, const char *reason,
+                           const char *file, int line) {
   long old_cnt = gpr_atm_acq_load(&pi->ref_count);
   pi_add_ref(pi);
   gpr_log(GPR_DEBUG, "Add ref pi: %p, old: %ld -> new:%ld (%s) - (%s, %d)",
@@ -306,7 +306,7 @@ static void pi_add_ref_dbg(polling_island *pi, char *reason, char *file,
 }
 
 static void pi_unref_dbg(grpc_exec_ctx *exec_ctx, polling_island *pi,
-                         char *reason, char *file, int line) {
+                         const char *reason, const char *file, int line) {
   long old_cnt = gpr_atm_acq_load(&pi->ref_count);
   pi_unref(exec_ctx, pi);
   gpr_log(GPR_DEBUG, "Unref pi: %p, old:%ld -> new:%ld (%s) - (%s, %d)",
@@ -317,7 +317,7 @@ static grpc_workqueue *workqueue_ref(grpc_workqueue *workqueue,
                                      const char *file, int line,
                                      const char *reason) {
   if (workqueue != NULL) {
-    pi_add_ref_debug((polling_island *)workqueue, reason, file, line);
+    pi_add_ref_dbg((polling_island *)workqueue, reason, file, line);
   }
   return workqueue;
 }
@@ -325,7 +325,7 @@ static grpc_workqueue *workqueue_ref(grpc_workqueue *workqueue,
 static void workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue,
                             const char *file, int line, const char *reason) {
   if (workqueue != NULL) {
-    pi_unref_dbg((polling_island *)workqueue, reason, file, line);
+    pi_unref_dbg(exec_ctx, (polling_island *)workqueue, reason, file, line);
   }
 }
 #else
@@ -775,16 +775,21 @@ static polling_island *polling_island_merge(polling_island *p,
 static void workqueue_enqueue(grpc_exec_ctx *exec_ctx,
                               grpc_workqueue *workqueue, grpc_closure *closure,
                               grpc_error *error) {
-  polling_island *pi = (polling_island *)workqueue;
   GPR_TIMER_BEGIN("workqueue.enqueue", 0);
+  /* take a ref to the workqueue: otherwise it can happen that whatever events
+   * this kicks off ends up destroying the workqueue before this function
+   * completes */
+  GRPC_WORKQUEUE_REF(workqueue, "enqueue");
+  polling_island *pi = (polling_island *)workqueue;
   gpr_atm last = gpr_atm_no_barrier_fetch_add(&pi->workqueue_item_count, 1);
   closure->error_data.error = error;
   gpr_mpscq_push(&pi->workqueue_items, &closure->next_data.atm_next);
   if (last == 0) {
     workqueue_maybe_wakeup(pi);
   }
-  GPR_TIMER_END("workqueue.enqueue", 0);
   workqueue_move_items_to_parent(pi);
+  GRPC_WORKQUEUE_UNREF(exec_ctx, workqueue, "enqueue");
+  GPR_TIMER_END("workqueue.enqueue", 0);
 }
 
 static grpc_error *polling_island_global_init() {
@@ -1117,8 +1122,8 @@ static void fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
 
 static grpc_workqueue *fd_get_workqueue(grpc_fd *fd) {
   gpr_mu_lock(&fd->mu);
-  grpc_workqueue *workqueue =
-      grpc_workqueue_ref((grpc_workqueue *)fd->polling_island);
+  grpc_workqueue *workqueue = GRPC_WORKQUEUE_REF(
+      (grpc_workqueue *)fd->polling_island, "fd_get_workqueue");
   gpr_mu_unlock(&fd->mu);
   return workqueue;
 }

src/core/lib/iomgr/ev_posix.h

@@ -40,6 +40,7 @@
 #include "src/core/lib/iomgr/pollset.h"
 #include "src/core/lib/iomgr/pollset_set.h"
 #include "src/core/lib/iomgr/wakeup_fd_posix.h"
+#include "src/core/lib/iomgr/workqueue.h"
 
 typedef struct grpc_fd grpc_fd;
 

src/core/lib/iomgr/workqueue.h

@@ -54,7 +54,7 @@
    string will be printed alongside the refcount. When it is not defined, the
    string will be discarded at compilation time. */
 
-//#define GRPC_WORKQUEUE_REFCOUNT_DEBUG
+#define GRPC_WORKQUEUE_REFCOUNT_DEBUG
 #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG
 #define GRPC_WORKQUEUE_REF(p, r) \
   grpc_workqueue_ref((p), __FILE__, __LINE__, (r))

test/core/end2end/cq_verifier.c

@@ -126,14 +126,14 @@ static gpr_slice merge_slices(gpr_slice *slices, size_t nslices) {
   return out;
 }
 
-static int byte_buffer_eq_slice(grpc_byte_buffer *bb, gpr_slice b) {
+int raw_byte_buffer_eq_slice(grpc_byte_buffer *rbb, gpr_slice b) {
   gpr_slice a;
   int ok;
 
-  if (!bb) return 0;
+  if (!rbb) return 0;
 
-  a = merge_slices(bb->data.raw.slice_buffer.slices,
-                   bb->data.raw.slice_buffer.count);
+  a = merge_slices(rbb->data.raw.slice_buffer.slices,
+                   rbb->data.raw.slice_buffer.count);
   ok = GPR_SLICE_LENGTH(a) == GPR_SLICE_LENGTH(b) &&
        0 == memcmp(GPR_SLICE_START_PTR(a), GPR_SLICE_START_PTR(b),
                    GPR_SLICE_LENGTH(a));
@@ -142,6 +142,21 @@ static int byte_buffer_eq_slice(grpc_byte_buffer *bb, gpr_slice b) {
   return ok;
 }
 
+int byte_buffer_eq_slice(grpc_byte_buffer *bb, gpr_slice b) {
+  grpc_byte_buffer_reader reader;
+  grpc_byte_buffer *rbb;
+  int res;
+
+  GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, bb) &&
+             "Couldn't init byte buffer reader");
+  rbb = grpc_raw_byte_buffer_from_reader(&reader);
+  res = raw_byte_buffer_eq_slice(rbb, b);
+  grpc_byte_buffer_reader_destroy(&reader);
+  grpc_byte_buffer_destroy(rbb);
+
+  return res;
+}
+
 int byte_buffer_eq_string(grpc_byte_buffer *bb, const char *str) {
   grpc_byte_buffer_reader reader;
   grpc_byte_buffer *rbb;
@@ -150,7 +165,7 @@ int byte_buffer_eq_string(grpc_byte_buffer *bb, const char *str) {
   GPR_ASSERT(grpc_byte_buffer_reader_init(&reader, bb) &&
              "Couldn't init byte buffer reader");
   rbb = grpc_raw_byte_buffer_from_reader(&reader);
-  res = byte_buffer_eq_slice(rbb, gpr_slice_from_copied_string(str));
+  res = raw_byte_buffer_eq_slice(rbb, gpr_slice_from_copied_string(str));
   grpc_byte_buffer_reader_destroy(&reader);
   grpc_byte_buffer_destroy(rbb);
 

test/core/end2end/cq_verifier.h

@@ -67,6 +67,7 @@ void cq_expect_completion(cq_verifier *v, const char *file, int line, void *tag,
 #define CQ_EXPECT_COMPLETION(v, tag, success) \
   cq_expect_completion(v, __FILE__, __LINE__, tag, success)
 
+int byte_buffer_eq_slice(grpc_byte_buffer *bb, gpr_slice b);
 int byte_buffer_eq_string(grpc_byte_buffer *byte_buffer, const char *string);
 int contains_metadata(grpc_metadata_array *array, const char *key,
                       const char *value);

test/core/end2end/tests/payload.c

@@ -95,9 +95,25 @@ static void end_test(grpc_end2end_test_fixture *f) {
   grpc_completion_queue_destroy(f->cq);
 }
 
+/* Creates and returns a gpr_slice containing random alphanumeric characters. */
+static gpr_slice generate_random_slice() {
+  size_t i;
+  static const char chars[] = "abcdefghijklmnopqrstuvwxyz1234567890";
+  char output[1024 * 1024]; /* 1 MB */
+  for (i = 0; i < 1024 * 1024 - 1; ++i) {
+    output[i] = chars[rand() % (int)(sizeof(chars) - 1)];
+  }
+  output[1024 * 1024 - 1] = '\0';
+  return gpr_slice_from_copied_string(output);
+}
+
 static void request_response_with_payload(grpc_end2end_test_fixture f) {
-  gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
-  gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
+  /* Create large request and response bodies. These are big enough to require
+   * multiple round trips to deliver to the peer, and their exact contents of
+   * will be verified on completion. */
+  gpr_slice request_payload_slice = generate_random_slice();
+  gpr_slice response_payload_slice = generate_random_slice();
+
   grpc_call *c;
   grpc_call *s;
   grpc_byte_buffer *request_payload =
@@ -222,8 +238,9 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) {
   GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
   GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr"));
   GPR_ASSERT(was_cancelled == 0);
-  GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, "hello world"));
-  GPR_ASSERT(byte_buffer_eq_string(response_payload_recv, "hello you"));
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  GPR_ASSERT(
+      byte_buffer_eq_slice(response_payload_recv, response_payload_slice));
 
   gpr_free(details);
   grpc_metadata_array_destroy(&initial_metadata_recv);