Rewrite async_exec_lock using mpscq

15 files changed, 43 insertions, 88 deletions

diff --git a/BUILD b/BUILD
index bd00ff670f..5dc0d86abe 100644
--- a/BUILD
+++ b/BUILD
@@ -83,6 +83,7 @@ cc_library(
     "src/core/lib/support/murmur_hash.c",
     "src/core/lib/support/slice.c",
     "src/core/lib/support/slice_buffer.c",
+    "src/core/lib/support/stack_lockfree.c",
     "src/core/lib/support/string.c",
     "src/core/lib/support/string_posix.c",
     "src/core/lib/support/string_util_win32.c",
@@ -1224,6 +1225,7 @@ objc_library(
     "src/core/lib/support/murmur_hash.c",
     "src/core/lib/support/slice.c",
     "src/core/lib/support/slice_buffer.c",
+    "src/core/lib/support/stack_lockfree.c",
     "src/core/lib/support/string.c",
     "src/core/lib/support/string_posix.c",
     "src/core/lib/support/string_util_win32.c",
diff --git a/Makefile b/Makefile
index 50efcf349f..319f483e75 100644
--- a/Makefile
+++ b/Makefile
@@ -2352,6 +2352,7 @@ LIBGPR_SRC = \
     src/core/lib/support/murmur_hash.c \
     src/core/lib/support/slice.c \
     src/core/lib/support/slice_buffer.c \
+    src/core/lib/support/stack_lockfree.c \
     src/core/lib/support/string.c \
     src/core/lib/support/string_posix.c \
     src/core/lib/support/string_util_win32.c \
diff --git a/binding.gyp b/binding.gyp
index ccd811a30b..9d04eb971a 100644
--- a/binding.gyp
+++ b/binding.gyp
@@ -520,6 +520,7 @@
         'src/core/lib/support/murmur_hash.c',
         'src/core/lib/support/slice.c',
         'src/core/lib/support/slice_buffer.c',
+        'src/core/lib/support/stack_lockfree.c',
         'src/core/lib/support/string.c',
         'src/core/lib/support/string_posix.c',
         'src/core/lib/support/string_util_win32.c',
diff --git a/build.yaml b/build.yaml
index e6f263aef7..addb91fcce 100644
--- a/build.yaml
+++ b/build.yaml
@@ -102,6 +102,7 @@ filegroups:
   - src/core/lib/support/murmur_hash.c
   - src/core/lib/support/slice.c
   - src/core/lib/support/slice_buffer.c
+  - src/core/lib/support/stack_lockfree.c
   - src/core/lib/support/string.c
   - src/core/lib/support/string_posix.c
   - src/core/lib/support/string_util_win32.c
diff --git a/config.m4 b/config.m4
index 9ec7f493da..67ce4944ef 100644
--- a/config.m4
+++ b/config.m4
@@ -61,6 +61,7 @@ if test "$PHP_GRPC" != "no"; then
     src/core/lib/support/murmur_hash.c \
     src/core/lib/support/slice.c \
     src/core/lib/support/slice_buffer.c \
+    src/core/lib/support/stack_lockfree.c \
     src/core/lib/support/string.c \
     src/core/lib/support/string_posix.c \
     src/core/lib/support/string_util_win32.c \
diff --git a/gRPC.podspec b/gRPC.podspec
index 3df13f9f42..f995e8ad65 100644
--- a/gRPC.podspec
+++ b/gRPC.podspec
@@ -143,6 +143,7 @@ Pod::Spec.new do |s|
                       'src/core/lib/support/murmur_hash.c',
                       'src/core/lib/support/slice.c',
                       'src/core/lib/support/slice_buffer.c',
+                      'src/core/lib/support/stack_lockfree.c',
                       'src/core/lib/support/string.c',
                       'src/core/lib/support/string_posix.c',
                       'src/core/lib/support/string_util_win32.c',
diff --git a/grpc.gemspec b/grpc.gemspec
index 2aad578a26..de488d6f44 100755
--- a/grpc.gemspec
+++ b/grpc.gemspec
@@ -123,6 +123,7 @@ Gem::Specification.new do |s|
   s.files += %w( src/core/lib/support/murmur_hash.c )
   s.files += %w( src/core/lib/support/slice.c )
   s.files += %w( src/core/lib/support/slice_buffer.c )
+  s.files += %w( src/core/lib/support/stack_lockfree.c )
   s.files += %w( src/core/lib/support/string.c )
   s.files += %w( src/core/lib/support/string_posix.c )
   s.files += %w( src/core/lib/support/string_util_win32.c )
diff --git a/package.xml b/package.xml
index ee2468bc87..169d047a08 100644
--- a/package.xml
+++ b/package.xml
@@ -130,6 +130,7 @@
     <file baseinstalldir="/" name="src/core/lib/support/murmur_hash.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/support/slice.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/support/slice_buffer.c" role="src" />
+    <file baseinstalldir="/" name="src/core/lib/support/stack_lockfree.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/support/string.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/support/string_posix.c" role="src" />
     <file baseinstalldir="/" name="src/core/lib/support/string_util_win32.c" role="src" />
diff --git a/src/core/lib/iomgr/async_execution_lock.c b/src/core/lib/iomgr/async_execution_lock.c
index 00cd27504b..b6b7e0d92c 100644
--- a/src/core/lib/iomgr/async_execution_lock.c
+++ b/src/core/lib/iomgr/async_execution_lock.c
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -40,105 +40,44 @@
 
 #define NO_CONSUMER ((gpr_atm)1)
 
-static void bad_action(grpc_exec_ctx *exec_ctx, void *arg) {
-  GPR_UNREACHABLE_CODE(return );
-}
-
 void grpc_aelock_init(grpc_aelock *lock, grpc_workqueue *optional_workqueue) {
   lock->optional_workqueue = optional_workqueue;
-  gpr_atm_no_barrier_store(&lock->head, NO_CONSUMER);
-  gpr_atm_no_barrier_store(&lock->tombstone.next, 0);
-  lock->tombstone.action = bad_action;
-  lock->tail = &lock->tombstone;
+  gpr_atm_no_barrier_store(&lock->locked, 0);
+  gpr_mpscq_init(&lock->queue);
 }
 
 void grpc_aelock_destroy(grpc_aelock *lock) {
-  GPR_ASSERT(gpr_atm_no_barrier_load(&lock->head) == NO_CONSUMER);
+  GPR_ASSERT(gpr_atm_no_barrier_load(&lock->locked) == 0);
+  gpr_mpscq_destroy(&lock->queue);
 }
 
 static void finish(grpc_exec_ctx *exec_ctx, grpc_aelock *lock) {
-  for (;;) {
-    grpc_aelock_qnode *tail = lock->tail;
-    grpc_aelock_qnode *next =
-        (grpc_aelock_qnode *)gpr_atm_acq_load(&tail->next);
-    if (tail == &lock->tombstone) {
-      if (next == NULL) {
-        if (gpr_atm_rel_cas(&lock->head, (gpr_atm)&lock->tombstone,
-                            NO_CONSUMER)) {
-          return;
-        }
-        // TODO(ctiller): consider sleeping
-        continue;
-      } else {
-        // skip the tombstone: we'll re-add it later
-        lock->tail = next;
-        tail = next;
-        next = (grpc_aelock_qnode *)gpr_atm_acq_load(&tail->next);
-      }
-    }
-    if (next != NULL) {
-      // found a node
-      lock->tail = next;
-      tail->action(exec_ctx, tail->arg);
-      gpr_free(tail);
-    } else {
-      // nothing there: might be in an incosistant state
-      grpc_aelock_qnode *head =
-          (grpc_aelock_qnode *)gpr_atm_acq_load(&lock->head);
-      if (head != tail) {
-        // non-empty list: spin for a bit
-        // TODO(ctiller): consider sleeping?
-        continue;
-      }
-      // must have swallowed tombstone above: re-add it
-      gpr_atm_no_barrier_store(&lock->tombstone.next, 0);
-      while (!gpr_atm_rel_cas(&lock->head, (gpr_atm)head,
-                              (gpr_atm)&lock->tombstone)) {
-        head = (grpc_aelock_qnode *)gpr_atm_acq_load(&lock->head);
-      }
-      gpr_atm_rel_store(&head->next, (gpr_atm)&lock->tombstone);
+  while (gpr_atm_full_fetch_add(&lock->locked, -1) != 1) {
+    gpr_mpscq_node *n;
+    while ((n = gpr_mpscq_pop(&lock->queue)) == NULL) {
+      // TODO(ctiller): find something to fill in the time
     }
+    grpc_aelock_qnode *ln = (grpc_aelock_qnode*)n;
+    ln->action(exec_ctx, ln->arg);
+    gpr_free(ln);
   }
 }
 
 void grpc_aelock_execute(grpc_exec_ctx *exec_ctx, grpc_aelock *lock,
                          grpc_aelock_action action, void *arg,
                          size_t sizeof_arg) {
-  gpr_atm head;
-retry_top:
-  head = gpr_atm_acq_load(&lock->head);
-  if (head == NO_CONSUMER) {
-    if (!gpr_atm_rel_cas(&lock->head, NO_CONSUMER, (gpr_atm)&lock->tombstone)) {
-      goto retry_top;
-    }
+  if (gpr_atm_full_fetch_add(&lock->locked, 1) == 0) {
     action(exec_ctx, arg);
     finish(exec_ctx, lock);
-    return;  // early out
-  }
-
-  grpc_aelock_qnode *n = gpr_malloc(sizeof(*n) + sizeof_arg);
-  n->action = action;
-  if (sizeof_arg > 0) {
-    memcpy(n + 1, arg, sizeof_arg);
-    n->arg = n + 1;
   } else {
-    n->arg = arg;
-  }
-  gpr_atm_rel_store(&n->next, 0);
-  while (!gpr_atm_rel_cas(&lock->head, head, (gpr_atm)n)) {
-  retry_queue_load:
-    head = gpr_atm_acq_load(&lock->head);
-    if (head == NO_CONSUMER) {
-      if (!gpr_atm_rel_cas(&lock->head, NO_CONSUMER,
-                           (gpr_atm)&lock->tombstone)) {
-        goto retry_queue_load;
-      }
-      gpr_free(n);
-      action(exec_ctx, arg);
-      finish(exec_ctx, lock);
-      return;  // early out
+    grpc_aelock_qnode *n = gpr_malloc(sizeof(*n) + sizeof_arg);
+    n->action = action;
+    if (sizeof_arg > 0) {
+      memcpy(n + 1, arg, sizeof_arg);
+      n->arg = n + 1;
+    } else {
+      n->arg = arg;
     }
+    gpr_mpscq_push(&lock->queue, &n->mpscq_node);
   }
-  GPR_ASSERT(gpr_atm_rel_cas(&((grpc_aelock_qnode *)head)->next, 0, (gpr_atm)n));
-//  gpr_atm_rel_store(&((grpc_aelock_qnode *)head)->next, (gpr_atm)n);
 }
diff --git a/src/core/lib/iomgr/async_execution_lock.h b/src/core/lib/iomgr/async_execution_lock.h
index cd1c4811fe..20cbcaf2e1 100644
--- a/src/core/lib/iomgr/async_execution_lock.h
+++ b/src/core/lib/iomgr/async_execution_lock.h
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -37,22 +37,21 @@
 #include <stddef.h>
 
 #include <grpc/support/atm.h>
+#include "src/core/lib/support/mpscq.h"
 #include "src/core/lib/iomgr/exec_ctx.h"
 
 typedef void (*grpc_aelock_action)(grpc_exec_ctx *exec_ctx, void *arg);
 
 typedef struct grpc_aelock_qnode {
+  gpr_mpscq_node mpscq_node;
   grpc_aelock_action action;
   void *arg;
-  gpr_atm next;
 } grpc_aelock_qnode;
 
 typedef struct grpc_aelock {
   grpc_workqueue *optional_workqueue;
-  // grpc_aelock_qnode*
-  gpr_atm head;
-  grpc_aelock_qnode *tail;
-  grpc_aelock_qnode tombstone;
+  gpr_mpscq queue;
+  gpr_atm locked;
 } grpc_aelock;
 
 void grpc_aelock_init(grpc_aelock *lock, grpc_workqueue *optional_workqueue);
diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index 58425a7a29..c843ed4062 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -55,6 +55,7 @@ CORE_SOURCE_FILES = [
   'src/core/lib/support/murmur_hash.c',
   'src/core/lib/support/slice.c',
   'src/core/lib/support/slice_buffer.c',
+  'src/core/lib/support/stack_lockfree.c',
   'src/core/lib/support/string.c',
   'src/core/lib/support/string_posix.c',
   'src/core/lib/support/string_util_win32.c',
diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal
index 534775e6cd..4add517477 100644
--- a/tools/doxygen/Doxyfile.core.internal
+++ b/tools/doxygen/Doxyfile.core.internal
@@ -1171,6 +1171,7 @@ src/core/lib/support/mpscq.c \
 src/core/lib/support/murmur_hash.c \
 src/core/lib/support/slice.c \
 src/core/lib/support/slice_buffer.c \
+src/core/lib/support/stack_lockfree.c \
 src/core/lib/support/string.c \
 src/core/lib/support/string_posix.c \
 src/core/lib/support/string_util_win32.c \
diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json
index 2b042b7e08..4369771170 100644
--- a/tools/run_tests/sources_and_headers.json
+++ b/tools/run_tests/sources_and_headers.json
@@ -5529,6 +5529,7 @@
       "src/core/lib/support/murmur_hash.h", 
       "src/core/lib/support/slice.c", 
       "src/core/lib/support/slice_buffer.c", 
+      "src/core/lib/support/stack_lockfree.c", 
       "src/core/lib/support/stack_lockfree.h", 
       "src/core/lib/support/string.c", 
       "src/core/lib/support/string.h", 
diff --git a/vsprojects/vcxproj/gpr/gpr.vcxproj b/vsprojects/vcxproj/gpr/gpr.vcxproj
index 43fa41324d..1a67cb9f61 100644
--- a/vsprojects/vcxproj/gpr/gpr.vcxproj
+++ b/vsprojects/vcxproj/gpr/gpr.vcxproj
@@ -256,6 +256,8 @@
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\slice_buffer.c">
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\stack_lockfree.c">
+    </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\string.c">
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\string_posix.c">
diff --git a/vsprojects/vcxproj/gpr/gpr.vcxproj.filters b/vsprojects/vcxproj/gpr/gpr.vcxproj.filters
index 80c6ac7b55..c2d6b51c5b 100644
--- a/vsprojects/vcxproj/gpr/gpr.vcxproj.filters
+++ b/vsprojects/vcxproj/gpr/gpr.vcxproj.filters
@@ -76,6 +76,9 @@
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\slice_buffer.c">
       <Filter>src\core\lib\support</Filter>
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\stack_lockfree.c">
+      <Filter>src\core\lib\support</Filter>
+    </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\src\core\lib\support\string.c">
       <Filter>src\core\lib\support</Filter>
     </ClCompile>