Get subchannel index working

build.yaml

@@ -2505,8 +2505,8 @@ configs:
     LDXX: clang++
     compile_the_world: true
     test_environ:
-      ASAN_OPTIONS: suppressions=tools/asan_suppressions.txt:detect_leaks=1:color=always
+      ASAN_OPTIONS: detect_leaks=1:color=always
-      LSAN_OPTIONS: suppressions=tools/asan_suppressions.txt:report_objects=1
+      LSAN_OPTIONS: suppressions=tools/lsan_suppressions.txt:report_objects=1
     timeout_multiplier: 1.5
   asan-noleaks:
     CC: clang

grpc.gemspec

@@ -172,6 +172,7 @@ Gem::Specification.new do |s|
   s.files += %w( src/core/client_config/resolvers/sockaddr_resolver.h )
   s.files += %w( src/core/client_config/subchannel.h )
   s.files += %w( src/core/client_config/subchannel_factory.h )
+  s.files += %w( src/core/client_config/subchannel_index.h )
   s.files += %w( src/core/client_config/uri_parser.h )
   s.files += %w( src/core/compression/algorithm_metadata.h )
   s.files += %w( src/core/compression/message_compress.h )
@@ -310,6 +311,7 @@ Gem::Specification.new do |s|
   s.files += %w( src/core/client_config/resolvers/sockaddr_resolver.c )
   s.files += %w( src/core/client_config/subchannel.c )
   s.files += %w( src/core/client_config/subchannel_factory.c )
+  s.files += %w( src/core/client_config/subchannel_index.c )
   s.files += %w( src/core/client_config/uri_parser.c )
   s.files += %w( src/core/compression/algorithm.c )
   s.files += %w( src/core/compression/message_compress.c )

include/grpc/support/avl.h

@@ -81,7 +81,8 @@ void gpr_avl_unref(gpr_avl avl);
     if key exists in avl, the new tree's key entry updated
     (i.e. a duplicate is not created) */
 gpr_avl gpr_avl_add(gpr_avl avl, void *key, void *value);
-/** return a new tree with key deleted */
+/** return a new tree with key deleted
+    implicitly unrefs avl to allow easy chaining. */
 gpr_avl gpr_avl_remove(gpr_avl avl, void *key);
 /** lookup key, and return the associated value.
     does not mutate avl.

include/grpc/support/useful.h

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

package.json

@@ -123,6 +123,7 @@
     "src/core/client_config/resolvers/sockaddr_resolver.h",
     "src/core/client_config/subchannel.h",
     "src/core/client_config/subchannel_factory.h",
+    "src/core/client_config/subchannel_index.h",
     "src/core/client_config/uri_parser.h",
     "src/core/compression/algorithm_metadata.h",
     "src/core/compression/message_compress.h",
@@ -261,6 +262,7 @@
     "src/core/client_config/resolvers/sockaddr_resolver.c",
     "src/core/client_config/subchannel.c",
     "src/core/client_config/subchannel_factory.c",
+    "src/core/client_config/subchannel_index.c",
     "src/core/client_config/uri_parser.c",
     "src/core/compression/algorithm.c",
     "src/core/compression/message_compress.c",

src/core/channel/channel_args.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -93,7 +93,7 @@ grpc_channel_args *grpc_channel_args_merge(const grpc_channel_args *a,
 }
 
 static int cmp_arg(const grpc_arg *a, const grpc_arg *b) {
-  int c = a->type - b->type;
+  int c = GPR_ICMP(a->type, b->type);
   if (c != 0) return c;
   c = strcmp(a->key, b->key);
   if (c != 0) return c;

src/core/channel/channel_args.h

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

src/core/client_config/connector.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

src/core/client_config/connector.h

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

src/core/client_config/subchannel.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -210,6 +210,7 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
   grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
   grpc_connector_unref(exec_ctx, c->connector);
   grpc_pollset_set_destroy(&c->pollset_set);
+  grpc_subchannel_key_destroy(exec_ctx, c->key);
   gpr_free(c);
 }
 
@@ -225,18 +226,21 @@ static gpr_atm ref_mutate(grpc_subchannel *c, gpr_atm delta,
   return old_val;
 }
 
-void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+grpc_subchannel *grpc_subchannel_ref(grpc_subchannel *c
+                                         GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
   gpr_atm old_refs;
   old_refs = ref_mutate(c, (1 << INTERNAL_REF_BITS),
                         0 REF_MUTATE_PURPOSE("STRONG_REF"));
   GPR_ASSERT((old_refs & STRONG_REF_MASK) != 0);
+  return c;
 }
 
-void grpc_subchannel_weak_ref(grpc_subchannel *c
+grpc_subchannel *grpc_subchannel_weak_ref(grpc_subchannel *c
-                                  GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+                                              GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
   gpr_atm old_refs;
   old_refs = ref_mutate(c, 1, 0 REF_MUTATE_PURPOSE("WEAK_REF"));
   GPR_ASSERT(old_refs != 0);
+  return c;
 }
 
 grpc_subchannel *grpc_subchannel_ref_from_weak_ref(
@@ -302,7 +306,7 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
   grpc_subchannel_key *key = grpc_subchannel_key_create(connector, args);
   grpc_subchannel *c = grpc_subchannel_index_find(exec_ctx, key);
   if (c) {
-    grpc_subchannel_key_destroy(key);
+    grpc_subchannel_key_destroy(exec_ctx, key);
     return c;
   }
 

src/core/client_config/subchannel.h

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -83,15 +83,15 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
 #define GRPC_SUBCHANNEL_REF_EXTRA_ARGS
 #endif
 
-void grpc_subchannel_ref(grpc_subchannel *channel
+grpc_subchannel *grpc_subchannel_ref(grpc_subchannel *channel
-                             GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+                                         GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 grpc_subchannel *grpc_subchannel_ref_from_weak_ref(
     grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
                            grpc_subchannel *channel
                                GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
-void grpc_subchannel_weak_ref(grpc_subchannel *channel
+grpc_subchannel *grpc_subchannel_weak_ref(grpc_subchannel *channel
-                                  GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+                                              GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
                                 grpc_subchannel *channel
                                     GRPC_SUBCHANNEL_REF_EXTRA_ARGS);

src/core/client_config/subchannel_index.c

@@ -1,35 +1,35 @@
-/*
+//
- *
+//
- * Copyright 2015, Google Inc.
+// Copyright 2016, Google Inc.
- * All rights reserved.
+// All rights reserved.
- *
+//
- * Redistribution and use in source and binary forms, with or without
+// Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
+// modification, are permitted provided that the following conditions are
- * met:
+// met:
- *
+//
- *     * Redistributions of source code must retain the above copyright
+//     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
+// notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
+//     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
+// copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
+// in the documentation and/or other materials provided with the
- * distribution.
+// distribution.
- *     * Neither the name of Google Inc. nor the names of its
+//     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
+// contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
+// this software without specific prior written permission.
- *
+//
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
+//
- */
+//
 
 #include "src/core/client_config/subchannel_index.h"
 
@@ -42,8 +42,8 @@
 
 #include "src/core/channel/channel_args.h"
 
-/* a map of subchannel_key --> subchannel, used for detecting connections
+// a map of subchannel_key --> subchannel, used for detecting connections
-   to the same destination in order to share them */
+// to the same destination in order to share them
 static gpr_avl g_subchannel_index;
 
 static gpr_mu g_mu;
@@ -111,14 +111,18 @@ static int subchannel_key_compare(grpc_subchannel_key *a,
   return grpc_channel_args_compare(a->args.args, b->args.args);
 }
 
-void grpc_subchannel_key_destroy(grpc_subchannel_key *k) {
+void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx,
+                                 grpc_subchannel_key *k) {
+  grpc_connector_unref(exec_ctx, k->connector);
   gpr_free(k->args.addr);
   gpr_free(k->args.filters);
   grpc_channel_args_destroy((grpc_channel_args *)k->args.args);
   gpr_free(k);
 }
 
-static void sck_avl_destroy(void *p) { grpc_subchannel_key_destroy(p); }
+static void sck_avl_destroy(void *p) {
+  grpc_subchannel_key_destroy(current_ctx(), p);
+}
 
 static void *sck_avl_copy(void *p) { return subchannel_key_copy(p); }
 
@@ -127,8 +131,7 @@ static long sck_avl_compare(void *a, void *b) {
 }
 
 static void scv_avl_destroy(void *p) {
-  grpc_exec_ctx *exec_ctx = current_ctx();
+  GRPC_SUBCHANNEL_WEAK_UNREF(current_ctx(), p, "subchannel_index");
-  GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, p, "subchannel_index");
 }
 
 static void *scv_avl_copy(void *p) {
@@ -157,6 +160,8 @@ grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx,
                                             grpc_subchannel_key *key) {
   enter_ctx(exec_ctx);
 
+  // Lock, and take a reference to the subchannel index.
+  // We don't need to do the search under a lock as avl's are immutable.
   gpr_mu_lock(&g_mu);
   gpr_avl index = gpr_avl_ref(g_subchannel_index);
   gpr_mu_unlock(&g_mu);
@@ -177,22 +182,34 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
   grpc_subchannel *c = NULL;
 
   while (c == NULL) {
+    // Compare and swap loop:
+    // - take a reference to the current index
     gpr_mu_lock(&g_mu);
     gpr_avl index = gpr_avl_ref(g_subchannel_index);
     gpr_mu_unlock(&g_mu);
 
+    // - Check to see if a subchannel already exists
     c = gpr_avl_get(index, key);
     if (c != NULL) {
+      // yes -> we're done
       GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, constructed, "index_register");
     } else {
-      gpr_avl updated = gpr_avl_add(index, key, constructed);
+      // no -> update the avl and compare/swap
-
+      gpr_avl updated =
+          gpr_avl_add(gpr_avl_ref(index), subchannel_key_copy(key),
+                      GRPC_SUBCHANNEL_WEAK_REF(constructed, "index_register"));
+
+      // it may happen (but it's expected to be unlikely)
+      // that some other thread has changed the index:
+      // compare/swap here to check that, and retry as necessary
       gpr_mu_lock(&g_mu);
       if (index.root == g_subchannel_index.root) {
         GPR_SWAP(gpr_avl, updated, g_subchannel_index);
         c = constructed;
       }
       gpr_mu_unlock(&g_mu);
+
+      gpr_avl_unref(updated);
     }
     gpr_avl_unref(index);
   }
@@ -209,26 +226,32 @@ void grpc_subchannel_index_unregister(grpc_exec_ctx *exec_ctx,
 
   bool done = false;
   while (!done) {
+    // Compare and swap loop:
+    // - take a reference to the current index
     gpr_mu_lock(&g_mu);
     gpr_avl index = gpr_avl_ref(g_subchannel_index);
     gpr_mu_unlock(&g_mu);
 
+    // Check to see if this key still refers to the previously
+    // registered subchannel
     grpc_subchannel *c = gpr_avl_get(index, key);
     if (c != constructed) {
+      gpr_avl_unref(index);
       break;
     }
 
-    gpr_avl updated = gpr_avl_remove(index, key);
+    // compare and swap the update (some other thread may have
+    // mutated the index behind us)
+    gpr_avl updated = gpr_avl_remove(gpr_avl_ref(index), key);
 
     gpr_mu_lock(&g_mu);
     if (index.root == g_subchannel_index.root) {
       GPR_SWAP(gpr_avl, updated, g_subchannel_index);
       done = true;
-    } else {
-      GPR_SWAP(gpr_avl, updated, index);
     }
     gpr_mu_unlock(&g_mu);
 
+    gpr_avl_unref(updated);
     gpr_avl_unref(index);
   }
 

src/core/client_config/subchannel_index.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,25 +37,41 @@
 #include "src/core/client_config/connector.h"
 #include "src/core/client_config/subchannel.h"
 
+/** \file Provides an index of active subchannels so that they can be
+    shared amongst channels */
+
 typedef struct grpc_subchannel_key grpc_subchannel_key;
 
+/** Create a key that can be used to uniquely identify a subchannel */
 grpc_subchannel_key *grpc_subchannel_key_create(grpc_connector *con,
                                                 grpc_subchannel_args *args);
 
-void grpc_subchannel_key_destroy(grpc_subchannel_key *key);
+/** Destroy a subchannel key */
+void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx,
+                                 grpc_subchannel_key *key);
 
+/** Given a subchannel key, find the subchannel registered for it.
+    Returns NULL if no such channel exists.
+    Thread-safe. */
 grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx,
                                             grpc_subchannel_key *key);
 
+/** Register a subchannel against a key. 
+    Takes ownership of \a constructed.
+    Returns the registered subchannel. This may be different from
+    \a constructed in the case of a registration race. */
 grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
                                                 grpc_subchannel_key *key,
                                                 grpc_subchannel *constructed);
 
+/** Remove \a constructed as the registered subchannel for \a key. */
 void grpc_subchannel_index_unregister(grpc_exec_ctx *exec_ctx,
                                       grpc_subchannel_key *key,
                                       grpc_subchannel *constructed);
 
+/** Initialize the subchannel index (global) */
 void grpc_subchannel_index_init(void);
+/** Shutdown the subchannel index (global) */
 void grpc_subchannel_index_shutdown(void);
 
 #endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_INDEX_H */

src/core/security/security_context.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

src/core/surface/channel_create.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

src/core/surface/secure_channel_create.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

src/cpp/common/channel_arguments.cc

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

test/core/end2end/fixtures/h2_uchannel.c

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without

tools/distrib/check_copyright.py

@@ -68,9 +68,9 @@ with open('LICENSE') as f:
 # that given a line of license text, returns what should
 # be in the file
 LICENSE_PREFIX = {
-  '.c':         r'\s*\*\s*',
+  '.c':         r'\s*(//|\*)\s*',
-  '.cc':        r'\s*\*\s*',
+  '.cc':        r'\s*(//|\*)\s*',
-  '.h':         r'\s*\*\s*',
+  '.h':         r'\s*(//|\*)\s*',
   '.m':         r'\s*\*\s*',
   '.php':       r'\s*\*\s*',
   '.js':        r'\s*\*\s*',

tools/run_tests/configs.json

@@ -45,8 +45,8 @@
   {
     "config": "asan", 
     "environ": {
-      "ASAN_OPTIONS": "suppressions=tools/asan_suppressions.txt:detect_leaks=1:color=always", 
+      "ASAN_OPTIONS": "detect_leaks=1:color=always", 
-      "LSAN_OPTIONS": "suppressions=tools/asan_suppressions.txt:report_objects=1"
+      "LSAN_OPTIONS": "suppressions=tools/lsan_suppressions.txt:report_objects=1"
     }, 
     "timeout_multiplier": 1.5
   },