1 files changed, 76 insertions, 140 deletions

diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 030a8b4e6f..86481af02c 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -45,34 +45,20 @@
 #include <grpc/support/atm.h>
 #include <grpc/support/log.h>
 
-#define NUM_TAG_BUCKETS 31
-
-/* A single event: extends grpc_event to form a linked list with a destruction
-   function (on_finish) that is hidden from outside this module */
-typedef struct event {
-  grpc_event base;
-  struct event *queue_next;
-  struct event *queue_prev;
-  struct event *bucket_next;
-  struct event *bucket_prev;
-} event;
-
 /* Completion queue structure */
 struct grpc_completion_queue {
-  /* When refs drops to zero, we are in shutdown mode, and will be destroyable
-     once all queued events are drained */
-  gpr_refcount refs;
-  /* Once owning_refs drops to zero, we will destroy the cq */
+  /** completed events */
+  grpc_cq_completion completed_head;
+  grpc_cq_completion *completed_tail;
+  /** Number of pending events (+1 if we're not shutdown) */
+  gpr_refcount pending_events;
+  /** Once owning_refs drops to zero, we will destroy the cq */
   gpr_refcount owning_refs;
-  /* the set of low level i/o things that concern this cq */
+  /** the set of low level i/o things that concern this cq */
   grpc_pollset pollset;
-  /* 0 initially, 1 once we've begun shutting down */
+  /** 0 initially, 1 once we've begun shutting down */
   int shutdown;
   int shutdown_called;
-  /* Head of a linked list of queued events (prev points to the last element) */
-  event *queue;
-  /* Fixed size chained hash table of events for pluck() */
-  event *buckets[NUM_TAG_BUCKETS];
   int is_server_cq;
 };
 
@@ -80,10 +66,12 @@ grpc_completion_queue *grpc_completion_queue_create(void) {
   grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue));
   memset(cc, 0, sizeof(*cc));
   /* Initial ref is dropped by grpc_completion_queue_shutdown */
-  gpr_ref_init(&cc->refs, 1);
+  gpr_ref_init(&cc->pending_events, 1);
   /* One for destroy(), one for pollset_shutdown */
   gpr_ref_init(&cc->owning_refs, 2);
   grpc_pollset_init(&cc->pollset);
+  cc->completed_tail = &cc->completed_head;
+  cc->completed_head.next = (gpr_uintptr) cc->completed_tail;
   return cc;
 }
 
@@ -114,179 +102,127 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason,
 void grpc_cq_internal_unref(grpc_completion_queue *cc) {
 #endif
   if (gpr_unref(&cc->owning_refs)) {
-    GPR_ASSERT(cc->queue == NULL);
+    GPR_ASSERT(cc->completed_head.next == (gpr_uintptr) &cc->completed_head);
     grpc_pollset_destroy(&cc->pollset);
     gpr_free(cc);
   }
 }
 
-/* Create and append an event to the queue. Returns the event so that its data
-   members can be filled in.
-   Requires GRPC_POLLSET_MU(&cc->pollset) locked. */
-static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type,
-                         void *tag, grpc_call *call) {
-  event *ev = gpr_malloc(sizeof(event));
-  gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS;
-  ev->base.type = type;
-  ev->base.tag = tag;
-  if (cc->queue == NULL) {
-    cc->queue = ev->queue_next = ev->queue_prev = ev;
-  } else {
-    ev->queue_next = cc->queue;
-    ev->queue_prev = cc->queue->queue_prev;
-    ev->queue_next->queue_prev = ev->queue_prev->queue_next = ev;
-  }
-  if (cc->buckets[bucket] == NULL) {
-    cc->buckets[bucket] = ev->bucket_next = ev->bucket_prev = ev;
-  } else {
-    ev->bucket_next = cc->buckets[bucket];
-    ev->bucket_prev = cc->buckets[bucket]->bucket_prev;
-    ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev;
-  }
-  grpc_pollset_kick(&cc->pollset);
-  return ev;
-}
-
-void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call) {
-  gpr_ref(&cc->refs);
-  if (call) GRPC_CALL_INTERNAL_REF(call, "cq");
+void grpc_cq_begin_op(grpc_completion_queue *cc) {
+  gpr_ref(&cc->pending_events);
 }
 
 /* Signal the end of an operation - if this is the last waiting-to-be-queued
    event, then enter shutdown mode */
-void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
-                    int success) {
-  event *ev;
-  int shutdown = 0;
-  gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
-  ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call);
-  ev->base.success = success;
-  if (gpr_unref(&cc->refs)) {
+/* Queue a GRPC_OP_COMPLETED operation */
+void grpc_cq_end_op(
+    grpc_completion_queue *cc, 
+    void *tag, 
+    int success, 
+    void (*done)(void *done_arg, grpc_cq_completion *storage), 
+    void *done_arg,
+    grpc_cq_completion *storage) {
+  int shutdown = gpr_unref(&cc->pending_events);
+
+  storage->tag = tag;
+  storage->done = done;
+  storage->done_arg = done_arg;
+  storage->next = ((gpr_uintptr)&cc->completed_head) | ((gpr_uintptr)(success != 0));
+
+  if (!shutdown) {
+    gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
+    cc->completed_tail->next = ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next);
+    cc->completed_tail = storage;
+    gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+  } else {
+    gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
+    cc->completed_tail->next = ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next);
+    cc->completed_tail = storage;
     GPR_ASSERT(!cc->shutdown);
     GPR_ASSERT(cc->shutdown_called);
     cc->shutdown = 1;
-    shutdown = 1;
-  }
-  gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
-  if (call) GRPC_CALL_INTERNAL_UNREF(call, "cq", 0);
-  if (shutdown) {
+    gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
     grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
   }
 }
 
-/* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */
-static event *create_shutdown_event(void) {
-  event *ev = gpr_malloc(sizeof(event));
-  ev->base.type = GRPC_QUEUE_SHUTDOWN;
-  ev->base.tag = NULL;
-  return ev;
-}
-
 grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
                                       gpr_timespec deadline) {
-  event *ev = NULL;
   grpc_event ret;
 
   GRPC_CQ_INTERNAL_REF(cc, "next");
   gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
   for (;;) {
-    if (cc->queue != NULL) {
-      gpr_uintptr bucket;
-      ev = cc->queue;
-      bucket = ((gpr_uintptr)ev->base.tag) % NUM_TAG_BUCKETS;
-      cc->queue = ev->queue_next;
-      ev->queue_next->queue_prev = ev->queue_prev;
-      ev->queue_prev->queue_next = ev->queue_next;
-      ev->bucket_next->bucket_prev = ev->bucket_prev;
-      ev->bucket_prev->bucket_next = ev->bucket_next;
-      if (ev == cc->buckets[bucket]) {
-        cc->buckets[bucket] = ev->bucket_next;
-        if (ev == cc->buckets[bucket]) {
-          cc->buckets[bucket] = NULL;
-        }
-      }
-      if (cc->queue == ev) {
-        cc->queue = NULL;
+    if (cc->completed_tail != &cc->completed_head) {
+      grpc_cq_completion *c = (grpc_cq_completion *) cc->completed_head.next;
+      cc->completed_head.next = c->next & ~(gpr_uintptr)1;
+      if (c == cc->completed_tail) {
+        cc->completed_tail = &cc->completed_head;
       }
+      gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+      ret.type = GRPC_OP_COMPLETE;
+      ret.success = c->next & 1u;
+      ret.tag = c->tag;
+      c->done(c->done_arg, c);
       break;
     }
     if (cc->shutdown) {
-      ev = create_shutdown_event();
+      gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+      memset(&ret, 0, sizeof(ret));
+      ret.type = GRPC_QUEUE_SHUTDOWN;
       break;
     }
     if (!grpc_pollset_work(&cc->pollset, deadline)) {
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
-      GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
-      GRPC_CQ_INTERNAL_UNREF(cc, "next");
-      return ret;
+      break;
     }
   }
-  gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
-  ret = ev->base;
-  gpr_free(ev);
   GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
   GRPC_CQ_INTERNAL_UNREF(cc, "next");
   return ret;
 }
 
-static event *pluck_event(grpc_completion_queue *cc, void *tag) {
-  gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS;
-  event *ev = cc->buckets[bucket];
-  if (ev == NULL) return NULL;
-  do {
-    if (ev->base.tag == tag) {
-      ev->queue_next->queue_prev = ev->queue_prev;
-      ev->queue_prev->queue_next = ev->queue_next;
-      ev->bucket_next->bucket_prev = ev->bucket_prev;
-      ev->bucket_prev->bucket_next = ev->bucket_next;
-      if (ev == cc->buckets[bucket]) {
-        cc->buckets[bucket] = ev->bucket_next;
-        if (ev == cc->buckets[bucket]) {
-          cc->buckets[bucket] = NULL;
-        }
-      }
-      if (cc->queue == ev) {
-        cc->queue = ev->queue_next;
-        if (cc->queue == ev) {
-          cc->queue = NULL;
-        }
-      }
-      return ev;
-    }
-    ev = ev->bucket_next;
-  } while (ev != cc->buckets[bucket]);
-  return NULL;
-}
-
 grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
                                        gpr_timespec deadline) {
-  event *ev = NULL;
   grpc_event ret;
+  grpc_cq_completion *c;
+  grpc_cq_completion *prev;
 
   GRPC_CQ_INTERNAL_REF(cc, "pluck");
   gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
   for (;;) {
-    if ((ev = pluck_event(cc, tag))) {
-      break;
+    prev = &cc->completed_head;
+    while ((c = (grpc_cq_completion*)(prev->next & ~(gpr_uintptr)1)) != &cc->completed_head) {
+      if (c->tag == tag) {
+        prev->next = (prev->next & (gpr_uintptr)1) | (c->next & ~(gpr_uintptr)1);
+        if (c == cc->completed_tail) {
+          cc->completed_tail = prev;
+        }
+        gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+        ret.type = GRPC_OP_COMPLETE;
+        ret.success = c->next & 1u;
+        ret.tag = c->tag;
+        c->done(c->done_arg, c);
+        goto done;
+      }
+      prev = c;
     }
     if (cc->shutdown) {
-      ev = create_shutdown_event();
+      gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+      memset(&ret, 0, sizeof(ret));
+      ret.type = GRPC_QUEUE_SHUTDOWN;
       break;
     }
     if (!grpc_pollset_work(&cc->pollset, deadline)) {
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
-      GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
-      GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
-      return ret;
+      break;
     }
   }
-  gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
-  ret = ev->base;
-  gpr_free(ev);
+done:
   GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
   GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
   return ret;
@@ -303,7 +239,7 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
   cc->shutdown_called = 1;
   gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
 
-  if (gpr_unref(&cc->refs)) {
+  if (gpr_unref(&cc->pending_events)) {
     gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
     GPR_ASSERT(!cc->shutdown);
     cc->shutdown = 1;