Merge github.com:grpc/grpc into consistent_client_count

1 files changed, 54 insertions, 14 deletions

diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index 5eb7cf1bf4..2cc6aa74e0 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -50,6 +50,8 @@
 #include "src/core/lib/surface/event_string.h"
 #include "src/core/lib/surface/surface_trace.h"
 
+int grpc_trace_operation_failures;
+
 typedef struct {
   grpc_pollset_worker **worker;
   void *tag;
@@ -92,7 +94,7 @@ static gpr_mu g_freelist_mu;
 static grpc_completion_queue *g_freelist;
 
 static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
-                                     bool success);
+                                     grpc_error *error);
 
 void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); }
 
@@ -176,7 +178,7 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) {
 }
 
 static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg,
-                                     bool success) {
+                                     grpc_error *error) {
   grpc_completion_queue *cc = arg;
   GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
 }
@@ -219,7 +221,7 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, void *tag) {
    event, then enter shutdown mode */
 /* Queue a GRPC_OP_COMPLETED operation */
 void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
-                    void *tag, int success,
+                    void *tag, grpc_error *error,
                     void (*done)(grpc_exec_ctx *exec_ctx, void *done_arg,
                                  grpc_cq_completion *storage),
                     void *done_arg, grpc_cq_completion *storage) {
@@ -231,16 +233,24 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
 #endif
 
   GPR_TIMER_BEGIN("grpc_cq_end_op", 0);
-  GRPC_API_TRACE(
-      "grpc_cq_end_op(exec_ctx=%p, cc=%p, tag=%p, success=%d, done=%p, "
-      "done_arg=%p, storage=%p)",
-      7, (exec_ctx, cc, tag, success, done, done_arg, storage));
+  if (grpc_api_trace ||
+      (grpc_trace_operation_failures && error != GRPC_ERROR_NONE)) {
+    const char *errmsg = grpc_error_string(error);
+    GRPC_API_TRACE(
+        "grpc_cq_end_op(exec_ctx=%p, cc=%p, tag=%p, error=%s, done=%p, "
+        "done_arg=%p, storage=%p)",
+        7, (exec_ctx, cc, tag, errmsg, done, done_arg, storage));
+    if (grpc_trace_operation_failures) {
+      gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag, errmsg);
+    }
+    grpc_error_free_string(errmsg);
+  }
 
   storage->tag = tag;
   storage->done = done;
   storage->done_arg = done_arg;
-  storage->next =
-      ((uintptr_t)&cc->completed_head) | ((uintptr_t)(success != 0));
+  storage->next = ((uintptr_t)&cc->completed_head) |
+                  ((uintptr_t)(error == GRPC_ERROR_NONE));
 
   gpr_mu_lock(cc->mu);
 #ifndef NDEBUG
@@ -267,8 +277,15 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
         break;
       }
     }
-    grpc_pollset_kick(POLLSET_FROM_CQ(cc), pluck_worker);
+    grpc_error *kick_error =
+        grpc_pollset_kick(POLLSET_FROM_CQ(cc), pluck_worker);
     gpr_mu_unlock(cc->mu);
+    if (kick_error != GRPC_ERROR_NONE) {
+      const char *msg = grpc_error_string(kick_error);
+      gpr_log(GPR_ERROR, "Kick failed: %s", msg);
+      grpc_error_free_string(msg);
+      GRPC_ERROR_UNREF(kick_error);
+    }
   } else {
     cc->completed_tail->next =
         ((uintptr_t)storage) | (1u & (uintptr_t)cc->completed_tail->next);
@@ -282,6 +299,8 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
   }
 
   GPR_TIMER_END("grpc_cq_end_op", 0);
+
+  GRPC_ERROR_UNREF(error);
 }
 
 grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
@@ -347,8 +366,18 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
       gpr_mu_lock(cc->mu);
       continue;
     } else {
-      grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), &worker, now,
-                        iteration_deadline);
+      grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc),
+                                          &worker, now, iteration_deadline);
+      if (err != GRPC_ERROR_NONE) {
+        gpr_mu_unlock(cc->mu);
+        const char *msg = grpc_error_string(err);
+        gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg);
+        grpc_error_free_string(msg);
+        GRPC_ERROR_UNREF(err);
+        memset(&ret, 0, sizeof(ret));
+        ret.type = GRPC_QUEUE_TIMEOUT;
+        break;
+      }
     }
   }
   GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
@@ -464,8 +493,19 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
       grpc_exec_ctx_flush(&exec_ctx);
       gpr_mu_lock(cc->mu);
     } else {
-      grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), &worker, now,
-                        iteration_deadline);
+      grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc),
+                                          &worker, now, iteration_deadline);
+      if (err != GRPC_ERROR_NONE) {
+        del_plucker(cc, tag, &worker);
+        gpr_mu_unlock(cc->mu);
+        const char *msg = grpc_error_string(err);
+        gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg);
+        grpc_error_free_string(msg);
+        GRPC_ERROR_UNREF(err);
+        memset(&ret, 0, sizeof(ret));
+        ret.type = GRPC_QUEUE_TIMEOUT;
+        break;
+      }
     }
     del_plucker(cc, tag, &worker);
   }