Merge branch 'slice_interning' into metadata_filter

20 files changed, 850 insertions, 778 deletions

diff --git a/src/core/lib/channel/handshaker.c b/src/core/lib/channel/handshaker.c
index a45a39981c..90626dc2d1 100644
--- a/src/core/lib/channel/handshaker.c
+++ b/src/core/lib/channel/handshaker.c
@@ -38,67 +38,66 @@
 
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/handshaker.h"
+#include "src/core/lib/iomgr/timer.h"
 
 //
 // grpc_handshaker
 //
 
-void grpc_handshaker_init(const struct grpc_handshaker_vtable* vtable,
+void grpc_handshaker_init(const grpc_handshaker_vtable* vtable,
                           grpc_handshaker* handshaker) {
   handshaker->vtable = vtable;
 }
 
-void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx,
-                             grpc_handshaker* handshaker) {
+static void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx,
+                                    grpc_handshaker* handshaker) {
   handshaker->vtable->destroy(exec_ctx, handshaker);
 }
 
-void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
-                              grpc_handshaker* handshaker) {
+static void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
+                                     grpc_handshaker* handshaker) {
   handshaker->vtable->shutdown(exec_ctx, handshaker);
 }
 
-void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
-                                  grpc_handshaker* handshaker,
-                                  grpc_endpoint* endpoint,
-                                  grpc_channel_args* args,
-                                  grpc_slice_buffer* read_buffer,
-                                  gpr_timespec deadline,
-                                  grpc_tcp_server_acceptor* acceptor,
-                                  grpc_handshaker_done_cb cb, void* user_data) {
-  handshaker->vtable->do_handshake(exec_ctx, handshaker, endpoint, args,
-                                   read_buffer, deadline, acceptor, cb,
-                                   user_data);
+static void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
+                                         grpc_handshaker* handshaker,
+                                         grpc_tcp_server_acceptor* acceptor,
+                                         grpc_closure* on_handshake_done,
+                                         grpc_handshaker_args* args) {
+  handshaker->vtable->do_handshake(exec_ctx, handshaker, acceptor,
+                                   on_handshake_done, args);
 }
 
 //
 // grpc_handshake_manager
 //
 
-// State used while chaining handshakers.
-struct grpc_handshaker_state {
-  // The index of the handshaker to invoke next.
-  size_t index;
-  // The deadline for all handshakers.
-  gpr_timespec deadline;
-  // The acceptor to call the handshakers with.
-  grpc_tcp_server_acceptor* acceptor;
-  // The final callback and user_data to invoke after the last handshaker.
-  grpc_handshaker_done_cb final_cb;
-  void* final_user_data;
-};
-
 struct grpc_handshake_manager {
+  gpr_mu mu;
+  gpr_refcount refs;
+  bool shutdown;
   // An array of handshakers added via grpc_handshake_manager_add().
   size_t count;
   grpc_handshaker** handshakers;
-  // State used while chaining handshakers.
-  struct grpc_handshaker_state* state;
+  // The index of the handshaker to invoke next and closure to invoke it.
+  size_t index;
+  grpc_closure call_next_handshaker;
+  // The acceptor to call the handshakers with.
+  grpc_tcp_server_acceptor* acceptor;
+  // Deadline timer across all handshakers.
+  grpc_timer deadline_timer;
+  // The final callback and user_data to invoke after the last handshaker.
+  grpc_closure on_handshake_done;
+  void* user_data;
+  // Handshaker args.
+  grpc_handshaker_args args;
 };
 
 grpc_handshake_manager* grpc_handshake_manager_create() {
   grpc_handshake_manager* mgr = gpr_malloc(sizeof(grpc_handshake_manager));
   memset(mgr, 0, sizeof(*mgr));
+  gpr_mu_init(&mgr->mu);
+  gpr_ref_init(&mgr->refs, 1);
   return mgr;
 }
 
@@ -106,6 +105,7 @@ static bool is_power_of_2(size_t n) { return (n & (n - 1)) == 0; }
 
 void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
                                 grpc_handshaker* handshaker) {
+  gpr_mu_lock(&mgr->mu);
   // To avoid allocating memory for each handshaker we add, we double
   // the number of elements every time we need more.
   size_t realloc_count = 0;
@@ -119,85 +119,116 @@ void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
         gpr_realloc(mgr->handshakers, realloc_count * sizeof(grpc_handshaker*));
   }
   mgr->handshakers[mgr->count++] = handshaker;
+  gpr_mu_unlock(&mgr->mu);
+}
+
+static void grpc_handshake_manager_unref(grpc_exec_ctx* exec_ctx,
+                                         grpc_handshake_manager* mgr) {
+  if (gpr_unref(&mgr->refs)) {
+    for (size_t i = 0; i < mgr->count; ++i) {
+      grpc_handshaker_destroy(exec_ctx, mgr->handshakers[i]);
+    }
+    gpr_free(mgr->handshakers);
+    gpr_mu_destroy(&mgr->mu);
+    gpr_free(mgr);
+  }
 }
 
 void grpc_handshake_manager_destroy(grpc_exec_ctx* exec_ctx,
                                     grpc_handshake_manager* mgr) {
-  for (size_t i = 0; i < mgr->count; ++i) {
-    grpc_handshaker_destroy(exec_ctx, mgr->handshakers[i]);
-  }
-  gpr_free(mgr->handshakers);
-  gpr_free(mgr);
+  grpc_handshake_manager_unref(exec_ctx, mgr);
 }
 
 void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx,
                                      grpc_handshake_manager* mgr) {
-  for (size_t i = 0; i < mgr->count; ++i) {
-    grpc_handshaker_shutdown(exec_ctx, mgr->handshakers[i]);
+  gpr_mu_lock(&mgr->mu);
+  // Shutdown the handshaker that's currently in progress, if any.
+  if (!mgr->shutdown && mgr->index > 0) {
+    mgr->shutdown = true;
+    grpc_handshaker_shutdown(exec_ctx, mgr->handshakers[mgr->index - 1]);
   }
-  if (mgr->state != NULL) {
-    gpr_free(mgr->state);
-    mgr->state = NULL;
+  gpr_mu_unlock(&mgr->mu);
+}
+
+// Helper function to call either the next handshaker or the
+// on_handshake_done callback.
+// Returns true if we've scheduled the on_handshake_done callback.
+static bool call_next_handshaker_locked(grpc_exec_ctx* exec_ctx,
+                                        grpc_handshake_manager* mgr,
+                                        grpc_error* error) {
+  GPR_ASSERT(mgr->index <= mgr->count);
+  // If we got an error or we've been shut down or we've finished the last
+  // handshaker, invoke the on_handshake_done callback.  Otherwise, call the
+  // next handshaker.
+  if (error != GRPC_ERROR_NONE || mgr->shutdown || mgr->index == mgr->count) {
+    // Cancel deadline timer, since we're invoking the on_handshake_done
+    // callback now.
+    grpc_timer_cancel(exec_ctx, &mgr->deadline_timer);
+    grpc_exec_ctx_sched(exec_ctx, &mgr->on_handshake_done, error, NULL);
+    mgr->shutdown = true;
+  } else {
+    grpc_handshaker_do_handshake(exec_ctx, mgr->handshakers[mgr->index],
+                                 mgr->acceptor, &mgr->call_next_handshaker,
+                                 &mgr->args);
   }
+  ++mgr->index;
+  return mgr->shutdown;
 }
 
 // A function used as the handshaker-done callback when chaining
 // handshakers together.
-static void call_next_handshaker(grpc_exec_ctx* exec_ctx,
-                                 grpc_endpoint* endpoint,
-                                 grpc_channel_args* args,
-                                 grpc_slice_buffer* read_buffer,
-                                 void* user_data, grpc_error* error) {
-  grpc_handshake_manager* mgr = user_data;
-  GPR_ASSERT(mgr->state != NULL);
-  GPR_ASSERT(mgr->state->index < mgr->count);
-  // If we got an error, skip all remaining handshakers and invoke the
-  // caller-supplied callback immediately.
-  if (error != GRPC_ERROR_NONE) {
-    mgr->state->final_cb(exec_ctx, endpoint, args, read_buffer,
-                         mgr->state->final_user_data, error);
-    return;
+static void call_next_handshaker(grpc_exec_ctx* exec_ctx, void* arg,
+                                 grpc_error* error) {
+  grpc_handshake_manager* mgr = arg;
+  gpr_mu_lock(&mgr->mu);
+  bool done = call_next_handshaker_locked(exec_ctx, mgr, GRPC_ERROR_REF(error));
+  gpr_mu_unlock(&mgr->mu);
+  // If we're invoked the final callback, we won't be coming back
+  // to this function, so we can release our reference to the
+  // handshake manager.
+  if (done) {
+    grpc_handshake_manager_unref(exec_ctx, mgr);
   }
-  grpc_handshaker_done_cb cb = call_next_handshaker;
-  // If this is the last handshaker, use the caller-supplied callback
-  // and user_data instead of chaining back to this function again.
-  if (mgr->state->index == mgr->count - 1) {
-    cb = mgr->state->final_cb;
-    user_data = mgr->state->final_user_data;
-  }
-  // Invoke handshaker.
-  grpc_handshaker_do_handshake(
-      exec_ctx, mgr->handshakers[mgr->state->index], endpoint, args,
-      read_buffer, mgr->state->deadline, mgr->state->acceptor, cb, user_data);
-  ++mgr->state->index;
-  // If this is the last handshaker, clean up state.
-  if (mgr->state->index == mgr->count) {
-    gpr_free(mgr->state);
-    mgr->state = NULL;
+}
+
+// Callback invoked when deadline is exceeded.
+static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
+  grpc_handshake_manager* mgr = arg;
+  if (error == GRPC_ERROR_NONE) {  // Timer fired, rather than being cancelled.
+    grpc_handshake_manager_shutdown(exec_ctx, mgr);
   }
+  grpc_handshake_manager_unref(exec_ctx, mgr);
 }
 
 void grpc_handshake_manager_do_handshake(
     grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr,
-    grpc_endpoint* endpoint, const grpc_channel_args* args,
+    grpc_endpoint* endpoint, const grpc_channel_args* channel_args,
     gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
-    grpc_handshaker_done_cb cb, void* user_data) {
-  grpc_channel_args* args_copy = grpc_channel_args_copy(args);
-  grpc_slice_buffer* read_buffer = gpr_malloc(sizeof(*read_buffer));
-  grpc_slice_buffer_init(read_buffer);
-  if (mgr->count == 0) {
-    // No handshakers registered, so we just immediately call the done
-    // callback with the passed-in endpoint.
-    cb(exec_ctx, endpoint, args_copy, read_buffer, user_data, GRPC_ERROR_NONE);
-  } else {
-    GPR_ASSERT(mgr->state == NULL);
-    mgr->state = gpr_malloc(sizeof(struct grpc_handshaker_state));
-    memset(mgr->state, 0, sizeof(*mgr->state));
-    mgr->state->deadline = deadline;
-    mgr->state->acceptor = acceptor;
-    mgr->state->final_cb = cb;
-    mgr->state->final_user_data = user_data;
-    call_next_handshaker(exec_ctx, endpoint, args_copy, read_buffer, mgr,
-                         GRPC_ERROR_NONE);
+    grpc_iomgr_cb_func on_handshake_done, void* user_data) {
+  gpr_mu_lock(&mgr->mu);
+  GPR_ASSERT(mgr->index == 0);
+  GPR_ASSERT(!mgr->shutdown);
+  // Construct handshaker args.  These will be passed through all
+  // handshakers and eventually be freed by the on_handshake_done callback.
+  mgr->args.endpoint = endpoint;
+  mgr->args.args = grpc_channel_args_copy(channel_args);
+  mgr->args.user_data = user_data;
+  mgr->args.read_buffer = gpr_malloc(sizeof(*mgr->args.read_buffer));
+  grpc_slice_buffer_init(mgr->args.read_buffer);
+  // Initialize state needed for calling handshakers.
+  mgr->acceptor = acceptor;
+  grpc_closure_init(&mgr->call_next_handshaker, call_next_handshaker, mgr);
+  grpc_closure_init(&mgr->on_handshake_done, on_handshake_done, &mgr->args);
+  // Start deadline timer, which owns a ref.
+  gpr_ref(&mgr->refs);
+  grpc_timer_init(exec_ctx, &mgr->deadline_timer,
+                  gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
+                  on_timeout, mgr, gpr_now(GPR_CLOCK_MONOTONIC));
+  // Start first handshaker, which also owns a ref.
+  gpr_ref(&mgr->refs);
+  bool done = call_next_handshaker_locked(exec_ctx, mgr, GRPC_ERROR_NONE);
+  gpr_mu_unlock(&mgr->mu);
+  if (done) {
+    grpc_handshake_manager_unref(exec_ctx, mgr);
   }
 }
diff --git a/src/core/lib/channel/handshaker.h b/src/core/lib/channel/handshaker.h
index f8a36c6473..ebbc1ff7f3 100644
--- a/src/core/lib/channel/handshaker.h
+++ b/src/core/lib/channel/handshaker.h
@@ -54,15 +54,30 @@
 
 typedef struct grpc_handshaker grpc_handshaker;
 
-/// Callback type invoked when a handshaker is done.
-/// Takes ownership of \a args and \a read_buffer.
-typedef void (*grpc_handshaker_done_cb)(grpc_exec_ctx* exec_ctx,
-                                        grpc_endpoint* endpoint,
-                                        grpc_channel_args* args,
-                                        grpc_slice_buffer* read_buffer,
-                                        void* user_data, grpc_error* error);
-
-struct grpc_handshaker_vtable {
+/// Arguments passed through handshakers and to the on_handshake_done callback.
+///
+/// For handshakers, all members are input/output parameters; for
+/// example, a handshaker may read from or write to \a endpoint and
+/// then later replace it with a wrapped endpoint.  Similarly, a
+/// handshaker may modify \a args.
+///
+/// A handshaker takes ownership of the members while a handshake is in
+/// progress.  Upon failure or shutdown of an in-progress handshaker,
+/// the handshaker is responsible for destroying the members and setting
+/// them to NULL before invoking the on_handshake_done callback.
+///
+/// For the on_handshake_done callback, all members are input arguments,
+/// which the callback takes ownership of.
+typedef struct {
+  grpc_endpoint* endpoint;
+  grpc_channel_args* args;
+  grpc_slice_buffer* read_buffer;
+  // User data passed through the handshake manager.  Not used by
+  // individual handshakers.
+  void* user_data;
+} grpc_handshaker_args;
+
+typedef struct {
   /// Destroys the handshaker.
   void (*destroy)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker);
 
@@ -70,44 +85,26 @@ struct grpc_handshaker_vtable {
   /// aborted in the middle).
   void (*shutdown)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker);
 
-  /// Performs handshaking.  When finished, calls \a cb with \a user_data.
-  /// Takes ownership of \a args.
-  /// Takes ownership of \a read_buffer, which contains leftover bytes read
-  /// from the endpoint by the previous handshaker.
+  /// Performs handshaking, modifying \a args as needed (e.g., to
+  /// replace \a endpoint with a wrapped endpoint).
+  /// When finished, invokes \a on_handshake_done.
   /// \a acceptor will be NULL for client-side handshakers.
   void (*do_handshake)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker,
-                       grpc_endpoint* endpoint, grpc_channel_args* args,
-                       grpc_slice_buffer* read_buffer, gpr_timespec deadline,
                        grpc_tcp_server_acceptor* acceptor,
-                       grpc_handshaker_done_cb cb, void* user_data);
-};
+                       grpc_closure* on_handshake_done,
+                       grpc_handshaker_args* args);
+} grpc_handshaker_vtable;
 
 /// Base struct.  To subclass, make this the first member of the
 /// implementation struct.
 struct grpc_handshaker {
-  const struct grpc_handshaker_vtable* vtable;
+  const grpc_handshaker_vtable* vtable;
 };
 
 /// Called by concrete implementations to initialize the base struct.
-void grpc_handshaker_init(const struct grpc_handshaker_vtable* vtable,
+void grpc_handshaker_init(const grpc_handshaker_vtable* vtable,
                           grpc_handshaker* handshaker);
 
-/// Convenient wrappers for invoking methods via the vtable.
-/// These probably do not need to be called from anywhere but
-/// grpc_handshake_manager.
-void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx,
-                             grpc_handshaker* handshaker);
-void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
-                              grpc_handshaker* handshaker);
-void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
-                                  grpc_handshaker* handshaker,
-                                  grpc_endpoint* endpoint,
-                                  grpc_channel_args* args,
-                                  grpc_slice_buffer* read_buffer,
-                                  gpr_timespec deadline,
-                                  grpc_tcp_server_acceptor* acceptor,
-                                  grpc_handshaker_done_cb cb, void* user_data);
-
 ///
 /// grpc_handshake_manager
 ///
@@ -134,15 +131,21 @@ void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx,
                                      grpc_handshake_manager* mgr);
 
 /// Invokes handshakers in the order they were added.
-/// Does NOT take ownership of \a args.  Instead, makes a copy before
+/// Takes ownership of \a endpoint, and then passes that ownership to
+/// the \a on_handshake_done callback.
+/// Does NOT take ownership of \a channel_args.  Instead, makes a copy before
 /// invoking the first handshaker.
 /// \a acceptor will be NULL for client-side handshakers.
-/// Invokes \a cb with \a user_data after either a handshaker fails or
-/// all handshakers have completed successfully.
+///
+/// When done, invokes \a on_handshake_done with a grpc_handshaker_args
+/// object as its argument.  If the callback is invoked with error !=
+/// GRPC_ERROR_NONE, then handshaking failed and the handshaker has done
+/// the necessary clean-up.  Otherwise, the callback takes ownership of
+/// the arguments.
 void grpc_handshake_manager_do_handshake(
     grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr,
-    grpc_endpoint* endpoint, const grpc_channel_args* args,
+    grpc_endpoint* endpoint, const grpc_channel_args* channel_args,
     gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
-    grpc_handshaker_done_cb cb, void* user_data);
+    grpc_iomgr_cb_func on_handshake_done, void* user_data);
 
 #endif /* GRPC_CORE_LIB_CHANNEL_HANDSHAKER_H */
diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c
index 3b47147685..31e9bde9d3 100644
--- a/src/core/lib/channel/http_client_filter.c
+++ b/src/core/lib/channel/http_client_filter.c
@@ -117,7 +117,7 @@ static grpc_error *client_filter_incoming_metadata(grpc_exec_ctx *exec_ctx,
         GRPC_MDVALUE(b->idx.named.grpc_message->md));
     if (grpc_slice_is_equivalent(pct_decoded_msg,
                                  GRPC_MDVALUE(b->idx.named.grpc_message->md))) {
-      grpc_slice_unref(pct_decoded_msg);
+      grpc_slice_unref_internal(exec_ctx, pct_decoded_msg);
     } else {
       grpc_metadata_batch_set_value(exec_ctx, b->idx.named.grpc_message,
                                     pct_decoded_msg);
diff --git a/src/core/lib/http/httpcli_security_connector.c b/src/core/lib/http/httpcli_security_connector.c
index e8575d4e6e..414383f8ca 100644
--- a/src/core/lib/http/httpcli_security_connector.c
+++ b/src/core/lib/http/httpcli_security_connector.c
@@ -38,7 +38,10 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
-#include "src/core/lib/security/transport/handshake.h"
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/security/transport/security_handshaker.h"
+#include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/tsi/ssl_transport_security.h"
 
@@ -59,52 +62,42 @@ static void httpcli_ssl_destroy(grpc_exec_ctx *exec_ctx,
   gpr_free(sc);
 }
 
-static void httpcli_ssl_do_handshake(grpc_exec_ctx *exec_ctx,
-                                     grpc_channel_security_connector *sc,
-                                     grpc_endpoint *nonsecure_endpoint,
-                                     grpc_slice_buffer *read_buffer,
-                                     gpr_timespec deadline,
-                                     grpc_security_handshake_done_cb cb,
-                                     void *user_data) {
+static void httpcli_ssl_create_handshakers(
+    grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc,
+    grpc_handshake_manager *handshake_mgr) {
   grpc_httpcli_ssl_channel_security_connector *c =
       (grpc_httpcli_ssl_channel_security_connector *)sc;
-  tsi_result result = TSI_OK;
-  tsi_handshaker *handshaker;
-  if (c->handshaker_factory == NULL) {
-    gpr_free(read_buffer);
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL);
-    return;
-  }
-  result = tsi_ssl_handshaker_factory_create_handshaker(
-      c->handshaker_factory, c->secure_peer_name, &handshaker);
-  if (result != TSI_OK) {
-    gpr_log(GPR_ERROR, "Handshaker creation failed with error %s.",
-            tsi_result_to_string(result));
-    gpr_free(read_buffer);
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL);
-  } else {
-    grpc_do_security_handshake(exec_ctx, handshaker, &sc->base, true,
-                               nonsecure_endpoint, read_buffer, deadline, cb,
-                               user_data);
+  tsi_handshaker *handshaker = NULL;
+  if (c->handshaker_factory != NULL) {
+    tsi_result result = tsi_ssl_handshaker_factory_create_handshaker(
+        c->handshaker_factory, c->secure_peer_name, &handshaker);
+    if (result != TSI_OK) {
+      gpr_log(GPR_ERROR, "Handshaker creation failed with error %s.",
+              tsi_result_to_string(result));
+    }
   }
+  grpc_security_create_handshakers(exec_ctx, handshaker, &sc->base,
+                                   handshake_mgr);
 }
 
 static void httpcli_ssl_check_peer(grpc_exec_ctx *exec_ctx,
                                    grpc_security_connector *sc, tsi_peer peer,
-                                   grpc_security_peer_check_cb cb,
-                                   void *user_data) {
+                                   grpc_auth_context **auth_context,
+                                   grpc_closure *on_peer_checked) {
   grpc_httpcli_ssl_channel_security_connector *c =
       (grpc_httpcli_ssl_channel_security_connector *)sc;
-  grpc_security_status status = GRPC_SECURITY_OK;
+  grpc_error *error = GRPC_ERROR_NONE;
 
   /* Check the peer name. */
   if (c->secure_peer_name != NULL &&
       !tsi_ssl_peer_matches_name(&peer, c->secure_peer_name)) {
-    gpr_log(GPR_ERROR, "Peer name %s is not in peer certificate",
-            c->secure_peer_name);
-    status = GRPC_SECURITY_ERROR;
+    char *msg;
+    gpr_asprintf(&msg, "Peer name %s is not in peer certificate",
+                 c->secure_peer_name);
+    error = GRPC_ERROR_CREATE(msg);
+    gpr_free(msg);
   }
-  cb(exec_ctx, user_data, status, NULL);
+  grpc_exec_ctx_sched(exec_ctx, on_peer_checked, error, NULL);
   tsi_peer_destruct(&peer);
 }
 
@@ -142,7 +135,7 @@ static grpc_security_status httpcli_ssl_channel_security_connector_create(
     *sc = NULL;
     return GRPC_SECURITY_ERROR;
   }
-  c->base.do_handshake = httpcli_ssl_do_handshake;
+  c->base.create_handshakers = httpcli_ssl_create_handshakers;
   *sc = &c->base;
   return GRPC_SECURITY_OK;
 }
@@ -152,19 +145,25 @@ static grpc_security_status httpcli_ssl_channel_security_connector_create(
 typedef struct {
   void (*func)(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *endpoint);
   void *arg;
+  grpc_handshake_manager *handshake_mgr;
 } on_done_closure;
 
-static void on_secure_transport_setup_done(grpc_exec_ctx *exec_ctx, void *rp,
-                                           grpc_security_status status,
-                                           grpc_endpoint *secure_endpoint,
-                                           grpc_auth_context *auth_context) {
-  on_done_closure *c = rp;
-  if (status != GRPC_SECURITY_OK) {
-    gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status);
+static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
+                              grpc_error *error) {
+  grpc_handshaker_args *args = arg;
+  on_done_closure *c = args->user_data;
+  if (error != GRPC_ERROR_NONE) {
+    const char *msg = grpc_error_string(error);
+    gpr_log(GPR_ERROR, "Secure transport setup failed: %s", msg);
+    grpc_error_free_string(msg);
     c->func(exec_ctx, c->arg, NULL);
   } else {
-    c->func(exec_ctx, c->arg, secure_endpoint);
+    grpc_channel_args_destroy(exec_ctx, args->args);
+    grpc_slice_buffer_destroy_internal(exec_ctx, args->read_buffer);
+    gpr_free(args->read_buffer);
+    c->func(exec_ctx, c->arg, args->endpoint);
   }
+  grpc_handshake_manager_destroy(exec_ctx, c->handshake_mgr);
   gpr_free(c);
 }
 
@@ -185,11 +184,15 @@ static void ssl_handshake(grpc_exec_ctx *exec_ctx, void *arg,
   }
   c->func = on_done;
   c->arg = arg;
+  c->handshake_mgr = grpc_handshake_manager_create();
   GPR_ASSERT(httpcli_ssl_channel_security_connector_create(
                  exec_ctx, pem_root_certs, pem_root_certs_size, host, &sc) ==
              GRPC_SECURITY_OK);
-  grpc_channel_security_connector_do_handshake(
-      exec_ctx, sc, tcp, NULL, deadline, on_secure_transport_setup_done, c);
+  grpc_channel_security_connector_create_handshakers(exec_ctx, sc,
+                                                     c->handshake_mgr);
+  grpc_handshake_manager_do_handshake(
+      exec_ctx, c->handshake_mgr, tcp, NULL /* channel_args */, deadline,
+      NULL /* acceptor */, on_handshake_done, c /* user_data */);
   GRPC_SECURITY_CONNECTOR_UNREF(exec_ctx, &sc->base, "httpcli");
 }
 
diff --git a/src/core/lib/iomgr/combiner.c b/src/core/lib/iomgr/combiner.c
index 60ee14eb23..cfc67020ae 100644
--- a/src/core/lib/iomgr/combiner.c
+++ b/src/core/lib/iomgr/combiner.c
@@ -90,6 +90,12 @@ static bool is_covered_by_poller(grpc_combiner *lock) {
          gpr_atm_acq_load(&lock->elements_covered_by_poller) > 0;
 }
 
+#define IS_COVERED_BY_POLLER_FMT "(final=%d elems=%" PRIdPTR ")->%d"
+#define IS_COVERED_BY_POLLER_ARGS(lock)                      \
+  (lock)->final_list_covered_by_poller,                      \
+      gpr_atm_acq_load(&(lock)->elements_covered_by_poller), \
+      is_covered_by_poller((lock))
+
 grpc_combiner *grpc_combiner_create(grpc_workqueue *optional_workqueue) {
   grpc_combiner *lock = gpr_malloc(sizeof(*lock));
   lock->next_combiner_on_this_exec_ctx = NULL;
@@ -197,9 +203,10 @@ bool grpc_combiner_continue_exec_ctx(grpc_exec_ctx *exec_ctx) {
   GRPC_COMBINER_TRACE(
       gpr_log(GPR_DEBUG,
               "C:%p grpc_combiner_continue_exec_ctx workqueue=%p "
-              "is_covered_by_poller=%d exec_ctx_ready_to_finish=%d "
+              "is_covered_by_poller=" IS_COVERED_BY_POLLER_FMT
+              " exec_ctx_ready_to_finish=%d "
               "time_to_execute_final_list=%d",
-              lock, lock->optional_workqueue, is_covered_by_poller(lock),
+              lock, lock->optional_workqueue, IS_COVERED_BY_POLLER_ARGS(lock),
               grpc_exec_ctx_ready_to_finish(exec_ctx),
               lock->time_to_execute_final_list));
 
diff --git a/src/core/lib/iomgr/resource_quota.c b/src/core/lib/iomgr/resource_quota.c
index 1862223b77..3dbc810102 100644
--- a/src/core/lib/iomgr/resource_quota.c
+++ b/src/core/lib/iomgr/resource_quota.c
@@ -144,6 +144,12 @@ struct grpc_resource_quota {
   /* Closure around rq_reclamation_done */
   grpc_closure rq_reclamation_done_closure;
 
+  /* This is only really usable for debugging: it's always a stale pointer, but
+     a stale pointer that might just be fresh enough to guide us to where the
+     reclamation system is stuck */
+  grpc_closure *debug_only_last_initiated_reclaimer;
+  grpc_resource_user *debug_only_last_reclaimer_resource_user;
+
   /* Roots of all resource user lists */
   grpc_resource_user *roots[GRPC_RULIST_COUNT];
 
@@ -225,6 +231,7 @@ static void rulist_remove(grpc_resource_user *resource_user, grpc_rulist list) {
       resource_user->links[list].prev;
   resource_user->links[list].prev->links[list].next =
       resource_user->links[list].next;
+  resource_user->links[list].next = resource_user->links[list].prev = NULL;
 }
 
 /*******************************************************************************
@@ -340,6 +347,9 @@ static bool rq_reclaim(grpc_exec_ctx *exec_ctx,
   resource_quota->reclaiming = true;
   grpc_resource_quota_ref_internal(resource_quota);
   grpc_closure *c = resource_user->reclaimers[destructive];
+  GPR_ASSERT(c);
+  resource_quota->debug_only_last_reclaimer_resource_user = resource_user;
+  resource_quota->debug_only_last_initiated_reclaimer = c;
   resource_user->reclaimers[destructive] = NULL;
   grpc_closure_run(exec_ctx, c, GRPC_ERROR_NONE);
   return true;
@@ -469,6 +479,8 @@ static void ru_shutdown(grpc_exec_ctx *exec_ctx, void *ru, grpc_error *error) {
                       GRPC_ERROR_CANCELLED, NULL);
   resource_user->reclaimers[0] = NULL;
   resource_user->reclaimers[1] = NULL;
+  rulist_remove(resource_user, GRPC_RULIST_RECLAIMER_BENIGN);
+  rulist_remove(resource_user, GRPC_RULIST_RECLAIMER_DESTRUCTIVE);
 }
 
 static void ru_destroy(grpc_exec_ctx *exec_ctx, void *ru, grpc_error *error) {
diff --git a/src/core/lib/iomgr/tcp_client_uv.c b/src/core/lib/iomgr/tcp_client_uv.c
index b07f9ceffa..5f2ccfee76 100644
--- a/src/core/lib/iomgr/tcp_client_uv.c
+++ b/src/core/lib/iomgr/tcp_client_uv.c
@@ -59,7 +59,7 @@ typedef struct grpc_uv_tcp_connect {
 
 static void uv_tcp_connect_cleanup(grpc_exec_ctx *exec_ctx,
                                    grpc_uv_tcp_connect *connect) {
-  grpc_resource_quota_internal_unref(exec_ctx, connect->resource_quota);
+  grpc_resource_quota_unref_internal(exec_ctx, connect->resource_quota);
   gpr_free(connect);
 }
 
@@ -128,8 +128,8 @@ static void tcp_client_connect_impl(grpc_exec_ctx *exec_ctx,
   if (channel_args != NULL) {
     for (size_t i = 0; i < channel_args->num_args; i++) {
       if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) {
-        grpc_resource_quota_internal_unref(exec_ctx, resource_quota);
-        resource_quota = grpc_resource_quota_internal_ref(
+        grpc_resource_quota_unref_internal(exec_ctx, resource_quota);
+        resource_quota = grpc_resource_quota_ref_internal(
             channel_args->args[i].value.pointer.p);
       }
     }
diff --git a/src/core/lib/iomgr/tcp_server_uv.c b/src/core/lib/iomgr/tcp_server_uv.c
index b5b9b92a20..050bb9e141 100644
--- a/src/core/lib/iomgr/tcp_server_uv.c
+++ b/src/core/lib/iomgr/tcp_server_uv.c
@@ -89,11 +89,11 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx,
   for (size_t i = 0; i < (args == NULL ? 0 : args->num_args); i++) {
     if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) {
       if (args->args[i].type == GRPC_ARG_POINTER) {
-        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);
+        grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota);
         s->resource_quota =
-            grpc_resource_quota_internal_ref(args->args[i].value.pointer.p);
+            grpc_resource_quota_ref_internal(args->args[i].value.pointer.p);
       } else {
-        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);
+        grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota);
         gpr_free(s);
         return GRPC_ERROR_CREATE(GRPC_ARG_RESOURCE_QUOTA
                                  " must be a pointer to a buffer pool");
@@ -136,7 +136,7 @@ static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
     gpr_free(sp->handle);
     gpr_free(sp);
   }
-  grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);
+  grpc_resource_quota_unref_internal(exec_ctx, s->resource_quota);
   gpr_free(s);
 }
 
diff --git a/src/core/lib/iomgr/tcp_uv.c b/src/core/lib/iomgr/tcp_uv.c
index 6e2ad1dbe9..257a4778c8 100644
--- a/src/core/lib/iomgr/tcp_uv.c
+++ b/src/core/lib/iomgr/tcp_uv.c
@@ -181,7 +181,7 @@ static void uv_endpoint_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   GPR_ASSERT(tcp->read_cb == NULL);
   tcp->read_cb = cb;
   tcp->read_slices = read_slices;
-  grpc_slice_buffer_reset_and_unref(read_slices);
+  grpc_slice_buffer_reset_and_unref_internal(exec_ctx,read_slices);
   TCP_REF(tcp, "read");
   // TODO(murgatroid99): figure out what the return value here means
   status =
diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c
index 56fdaa5d82..8b3b44aaaa 100644
--- a/src/core/lib/iomgr/tcp_windows.c
+++ b/src/core/lib/iomgr/tcp_windows.c
@@ -53,6 +53,7 @@
 #include "src/core/lib/iomgr/socket_windows.h"
 #include "src/core/lib/iomgr/tcp_client.h"
 #include "src/core/lib/iomgr/timer.h"
+#include "src/core/lib/slice/slice_internal.h"
 
 #if defined(__MSYS__) && defined(GPR_ARCH_64)
 /* Nasty workaround for nasty bug when using the 64 bits msys compiler
diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c
index fd0c7a0f9d..3c24ea9afa 100644
--- a/src/core/lib/iomgr/udp_server.c
+++ b/src/core/lib/iomgr/udp_server.c
@@ -388,7 +388,8 @@ int grpc_udp_server_add_port(grpc_udp_server *s,
     /* Try listening on IPv6 first. */
     addr = &wild6;
     // TODO(rjshade): Test and propagate the returned grpc_error*:
-    grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd);
+    GRPC_ERROR_UNREF(grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP,
+                                                  &dsmode, &fd));
     allocated_port1 = add_socket_to_server(s, fd, addr, read_cb, orphan_cb);
     if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) {
       goto done;
@@ -402,7 +403,8 @@ int grpc_udp_server_add_port(grpc_udp_server *s,
   }
 
   // TODO(rjshade): Test and propagate the returned grpc_error*:
-  grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode, &fd);
+  GRPC_ERROR_UNREF(grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP,
+                                                &dsmode, &fd));
   if (fd < 0) {
     gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno));
   }
diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.c b/src/core/lib/security/credentials/google_default/google_default_credentials.c
index 65eb02bbb9..190950619e 100644
--- a/src/core/lib/security/credentials/google_default/google_default_credentials.c
+++ b/src/core/lib/security/credentials/google_default/google_default_credentials.c
@@ -160,6 +160,7 @@ static int is_stack_running_on_compute_engine(grpc_exec_ctx *exec_ctx) {
                         grpc_polling_entity_pollset(&detector.pollent),
                         &destroy_closure);
   g_polling_mu = NULL;
+  grpc_exec_ctx_flush(exec_ctx);
 
   gpr_free(grpc_polling_entity_pollset(&detector.pollent));
   grpc_http_response_destroy(&detector.response);
diff --git a/src/core/lib/security/transport/handshake.c b/src/core/lib/security/transport/handshake.c
deleted file mode 100644
index 16d758d7a1..0000000000
--- a/src/core/lib/security/transport/handshake.c
+++ /dev/null
@@ -1,377 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- *     * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- *     * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- *     * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/core/lib/security/transport/handshake.h"
-
-#include <stdbool.h>
-#include <string.h>
-
-#include <grpc/slice_buffer.h>
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-#include "src/core/lib/iomgr/timer.h"
-#include "src/core/lib/security/context/security_context.h"
-#include "src/core/lib/security/transport/secure_endpoint.h"
-#include "src/core/lib/security/transport/tsi_error.h"
-#include "src/core/lib/slice/slice_internal.h"
-
-#define GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE 256
-
-typedef struct {
-  grpc_security_connector *connector;
-  tsi_handshaker *handshaker;
-  bool is_client_side;
-  unsigned char *handshake_buffer;
-  size_t handshake_buffer_size;
-  grpc_endpoint *wrapped_endpoint;
-  grpc_endpoint *secure_endpoint;
-  grpc_slice_buffer left_overs;
-  grpc_slice_buffer incoming;
-  grpc_slice_buffer outgoing;
-  grpc_security_handshake_done_cb cb;
-  void *user_data;
-  grpc_closure on_handshake_data_sent_to_peer;
-  grpc_closure on_handshake_data_received_from_peer;
-  grpc_auth_context *auth_context;
-  grpc_timer timer;
-  gpr_refcount refs;
-} grpc_security_handshake;
-
-static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
-                                                 void *setup,
-                                                 grpc_error *error);
-
-static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx, void *setup,
-                                           grpc_error *error);
-
-static void security_connector_remove_handshake(grpc_security_handshake *h) {
-  GPR_ASSERT(!h->is_client_side);
-  grpc_security_connector_handshake_list *node;
-  grpc_security_connector_handshake_list *tmp;
-  grpc_server_security_connector *sc =
-      (grpc_server_security_connector *)h->connector;
-  gpr_mu_lock(&sc->mu);
-  node = sc->handshaking_handshakes;
-  if (node && node->handshake == h) {
-    sc->handshaking_handshakes = node->next;
-    gpr_free(node);
-    gpr_mu_unlock(&sc->mu);
-    return;
-  }
-  while (node) {
-    if (node->next->handshake == h) {
-      tmp = node->next;
-      node->next = node->next->next;
-      gpr_free(tmp);
-      gpr_mu_unlock(&sc->mu);
-      return;
-    }
-    node = node->next;
-  }
-  gpr_mu_unlock(&sc->mu);
-}
-
-static void unref_handshake(grpc_exec_ctx *exec_ctx,
-                            grpc_security_handshake *h) {
-  if (gpr_unref(&h->refs)) {
-    if (h->handshaker != NULL) tsi_handshaker_destroy(h->handshaker);
-    if (h->handshake_buffer != NULL) gpr_free(h->handshake_buffer);
-    grpc_slice_buffer_destroy_internal(exec_ctx, &h->left_overs);
-    grpc_slice_buffer_destroy_internal(exec_ctx, &h->outgoing);
-    grpc_slice_buffer_destroy_internal(exec_ctx, &h->incoming);
-    GRPC_AUTH_CONTEXT_UNREF(h->auth_context, "handshake");
-    GRPC_SECURITY_CONNECTOR_UNREF(exec_ctx, h->connector, "handshake");
-    gpr_free(h);
-  }
-}
-
-static void security_handshake_done(grpc_exec_ctx *exec_ctx,
-                                    grpc_security_handshake *h,
-                                    grpc_error *error) {
-  grpc_timer_cancel(exec_ctx, &h->timer);
-  if (!h->is_client_side) {
-    security_connector_remove_handshake(h);
-  }
-  if (error == GRPC_ERROR_NONE) {
-    h->cb(exec_ctx, h->user_data, GRPC_SECURITY_OK, h->secure_endpoint,
-          h->auth_context);
-  } else {
-    const char *msg = grpc_error_string(error);
-    gpr_log(GPR_DEBUG, "Security handshake failed: %s", msg);
-    grpc_error_free_string(msg);
-
-    if (h->secure_endpoint != NULL) {
-      grpc_endpoint_shutdown(exec_ctx, h->secure_endpoint);
-      grpc_endpoint_destroy(exec_ctx, h->secure_endpoint);
-    } else {
-      grpc_endpoint_destroy(exec_ctx, h->wrapped_endpoint);
-    }
-    h->cb(exec_ctx, h->user_data, GRPC_SECURITY_ERROR, NULL, NULL);
-  }
-  unref_handshake(exec_ctx, h);
-  GRPC_ERROR_UNREF(error);
-}
-
-static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *user_data,
-                            grpc_security_status status,
-                            grpc_auth_context *auth_context) {
-  grpc_security_handshake *h = user_data;
-  tsi_frame_protector *protector;
-  tsi_result result;
-  if (status != GRPC_SECURITY_OK) {
-    security_handshake_done(
-        exec_ctx, h,
-        grpc_error_set_int(GRPC_ERROR_CREATE("Error checking peer."),
-                           GRPC_ERROR_INT_SECURITY_STATUS, status));
-    return;
-  }
-  h->auth_context = GRPC_AUTH_CONTEXT_REF(auth_context, "handshake");
-  result =
-      tsi_handshaker_create_frame_protector(h->handshaker, NULL, &protector);
-  if (result != TSI_OK) {
-    security_handshake_done(
-        exec_ctx, h,
-        grpc_set_tsi_error_result(
-            GRPC_ERROR_CREATE("Frame protector creation failed"), result));
-    return;
-  }
-  h->secure_endpoint =
-      grpc_secure_endpoint_create(protector, h->wrapped_endpoint,
-                                  h->left_overs.slices, h->left_overs.count);
-  h->left_overs.count = 0;
-  h->left_overs.length = 0;
-  security_handshake_done(exec_ctx, h, GRPC_ERROR_NONE);
-  return;
-}
-
-static void check_peer(grpc_exec_ctx *exec_ctx, grpc_security_handshake *h) {
-  tsi_peer peer;
-  tsi_result result = tsi_handshaker_extract_peer(h->handshaker, &peer);
-
-  if (result != TSI_OK) {
-    security_handshake_done(
-        exec_ctx, h, grpc_set_tsi_error_result(
-                         GRPC_ERROR_CREATE("Peer extraction failed"), result));
-    return;
-  }
-  grpc_security_connector_check_peer(exec_ctx, h->connector, peer,
-                                     on_peer_checked, h);
-}
-
-static void send_handshake_bytes_to_peer(grpc_exec_ctx *exec_ctx,
-                                         grpc_security_handshake *h) {
-  size_t offset = 0;
-  tsi_result result = TSI_OK;
-  grpc_slice to_send;
-
-  do {
-    size_t to_send_size = h->handshake_buffer_size - offset;
-    result = tsi_handshaker_get_bytes_to_send_to_peer(
-        h->handshaker, h->handshake_buffer + offset, &to_send_size);
-    offset += to_send_size;
-    if (result == TSI_INCOMPLETE_DATA) {
-      h->handshake_buffer_size *= 2;
-      h->handshake_buffer =
-          gpr_realloc(h->handshake_buffer, h->handshake_buffer_size);
-    }
-  } while (result == TSI_INCOMPLETE_DATA);
-
-  if (result != TSI_OK) {
-    security_handshake_done(exec_ctx, h,
-                            grpc_set_tsi_error_result(
-                                GRPC_ERROR_CREATE("Handshake failed"), result));
-    return;
-  }
-
-  to_send =
-      grpc_slice_from_copied_buffer((const char *)h->handshake_buffer, offset);
-  grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &h->outgoing);
-  grpc_slice_buffer_add(&h->outgoing, to_send);
-  /* TODO(klempner,jboeuf): This should probably use the client setup
-     deadline */
-  grpc_endpoint_write(exec_ctx, h->wrapped_endpoint, &h->outgoing,
-                      &h->on_handshake_data_sent_to_peer);
-}
-
-static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
-                                                 void *handshake,
-                                                 grpc_error *error) {
-  grpc_security_handshake *h = handshake;
-  size_t consumed_slice_size = 0;
-  tsi_result result = TSI_OK;
-  size_t i;
-  size_t num_left_overs;
-  int has_left_overs_in_current_slice = 0;
-
-  if (error != GRPC_ERROR_NONE) {
-    security_handshake_done(
-        exec_ctx, h,
-        GRPC_ERROR_CREATE_REFERENCING("Handshake read failed", &error, 1));
-    return;
-  }
-
-  for (i = 0; i < h->incoming.count; i++) {
-    consumed_slice_size = GRPC_SLICE_LENGTH(h->incoming.slices[i]);
-    result = tsi_handshaker_process_bytes_from_peer(
-        h->handshaker, GRPC_SLICE_START_PTR(h->incoming.slices[i]),
-        &consumed_slice_size);
-    if (!tsi_handshaker_is_in_progress(h->handshaker)) break;
-  }
-
-  if (tsi_handshaker_is_in_progress(h->handshaker)) {
-    /* We may need more data. */
-    if (result == TSI_INCOMPLETE_DATA) {
-      grpc_endpoint_read(exec_ctx, h->wrapped_endpoint, &h->incoming,
-                         &h->on_handshake_data_received_from_peer);
-      return;
-    } else {
-      send_handshake_bytes_to_peer(exec_ctx, h);
-      return;
-    }
-  }
-
-  if (result != TSI_OK) {
-    security_handshake_done(exec_ctx, h,
-                            grpc_set_tsi_error_result(
-                                GRPC_ERROR_CREATE("Handshake failed"), result));
-    return;
-  }
-
-  /* Handshake is done and successful this point. */
-  has_left_overs_in_current_slice =
-      (consumed_slice_size < GRPC_SLICE_LENGTH(h->incoming.slices[i]));
-  num_left_overs =
-      (has_left_overs_in_current_slice ? 1 : 0) + h->incoming.count - i - 1;
-  if (num_left_overs == 0) {
-    check_peer(exec_ctx, h);
-    return;
-  }
-
-  /* Put the leftovers in our buffer (ownership transfered). */
-  if (has_left_overs_in_current_slice) {
-    grpc_slice_buffer_add(
-        &h->left_overs,
-        grpc_slice_split_tail(&h->incoming.slices[i], consumed_slice_size));
-    grpc_slice_unref_internal(
-        exec_ctx,
-        h->incoming.slices[i]); /* split_tail above increments refcount. */
-  }
-  grpc_slice_buffer_addn(
-      &h->left_overs, &h->incoming.slices[i + 1],
-      num_left_overs - (size_t)has_left_overs_in_current_slice);
-  check_peer(exec_ctx, h);
-}
-
-/* If handshake is NULL, the handshake is done. */
-static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx,
-                                           void *handshake, grpc_error *error) {
-  grpc_security_handshake *h = handshake;
-
-  /* Make sure that write is OK. */
-  if (error != GRPC_ERROR_NONE) {
-    if (handshake != NULL)
-      security_handshake_done(
-          exec_ctx, h,
-          GRPC_ERROR_CREATE_REFERENCING("Handshake write failed", &error, 1));
-    return;
-  }
-
-  /* We may be done. */
-  if (tsi_handshaker_is_in_progress(h->handshaker)) {
-    /* TODO(klempner,jboeuf): This should probably use the client setup
-       deadline */
-    grpc_endpoint_read(exec_ctx, h->wrapped_endpoint, &h->incoming,
-                       &h->on_handshake_data_received_from_peer);
-  } else {
-    check_peer(exec_ctx, h);
-  }
-}
-
-static void on_timeout(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-  grpc_security_handshake *h = arg;
-  if (error == GRPC_ERROR_NONE) {
-    grpc_endpoint_shutdown(exec_ctx, h->wrapped_endpoint);
-  }
-  unref_handshake(exec_ctx, h);
-}
-
-void grpc_do_security_handshake(
-    grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker,
-    grpc_security_connector *connector, bool is_client_side,
-    grpc_endpoint *nonsecure_endpoint, grpc_slice_buffer *read_buffer,
-    gpr_timespec deadline, grpc_security_handshake_done_cb cb,
-    void *user_data) {
-  grpc_security_connector_handshake_list *handshake_node;
-  grpc_security_handshake *h = gpr_malloc(sizeof(grpc_security_handshake));
-  memset(h, 0, sizeof(grpc_security_handshake));
-  h->handshaker = handshaker;
-  h->connector = GRPC_SECURITY_CONNECTOR_REF(connector, "handshake");
-  h->is_client_side = is_client_side;
-  h->handshake_buffer_size = GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE;
-  h->handshake_buffer = gpr_malloc(h->handshake_buffer_size);
-  h->wrapped_endpoint = nonsecure_endpoint;
-  h->user_data = user_data;
-  h->cb = cb;
-  gpr_ref_init(&h->refs, 2); /* timer and handshake proper each get a ref */
-  grpc_closure_init(&h->on_handshake_data_sent_to_peer,
-                    on_handshake_data_sent_to_peer, h);
-  grpc_closure_init(&h->on_handshake_data_received_from_peer,
-                    on_handshake_data_received_from_peer, h);
-  grpc_slice_buffer_init(&h->left_overs);
-  grpc_slice_buffer_init(&h->outgoing);
-  grpc_slice_buffer_init(&h->incoming);
-  if (read_buffer != NULL) {
-    grpc_slice_buffer_move_into(read_buffer, &h->incoming);
-    gpr_free(read_buffer);
-  }
-  if (!is_client_side) {
-    grpc_server_security_connector *server_connector =
-        (grpc_server_security_connector *)connector;
-    handshake_node = gpr_malloc(sizeof(grpc_security_connector_handshake_list));
-    handshake_node->handshake = h;
-    gpr_mu_lock(&server_connector->mu);
-    handshake_node->next = server_connector->handshaking_handshakes;
-    server_connector->handshaking_handshakes = handshake_node;
-    gpr_mu_unlock(&server_connector->mu);
-  }
-  send_handshake_bytes_to_peer(exec_ctx, h);
-  grpc_timer_init(exec_ctx, &h->timer,
-                  gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
-                  on_timeout, h, gpr_now(GPR_CLOCK_MONOTONIC));
-}
-
-void grpc_security_handshake_shutdown(grpc_exec_ctx *exec_ctx,
-                                      void *handshake) {
-  grpc_security_handshake *h = handshake;
-  grpc_endpoint_shutdown(exec_ctx, h->wrapped_endpoint);
-}
diff --git a/src/core/lib/security/transport/security_connector.c b/src/core/lib/security/transport/security_connector.c
index 53bccb3efb..53c9abf0cb 100644
--- a/src/core/lib/security/transport/security_connector.c
+++ b/src/core/lib/security/transport/security_connector.c
@@ -46,8 +46,8 @@
 #include "src/core/lib/iomgr/load_file.h"
 #include "src/core/lib/security/context/security_context.h"
 #include "src/core/lib/security/credentials/credentials.h"
-#include "src/core/lib/security/transport/handshake.h"
 #include "src/core/lib/security/transport/secure_endpoint.h"
+#include "src/core/lib/security/transport/security_handshaker.h"
 #include "src/core/lib/support/env.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/tsi/fake_transport_security.h"
@@ -111,58 +111,34 @@ const tsi_peer_property *tsi_peer_get_property_by_name(const tsi_peer *peer,
   return NULL;
 }
 
-void grpc_server_security_connector_shutdown(
-    grpc_exec_ctx *exec_ctx, grpc_server_security_connector *connector) {
-  grpc_security_connector_handshake_list *tmp;
-  gpr_mu_lock(&connector->mu);
-  while (connector->handshaking_handshakes) {
-    tmp = connector->handshaking_handshakes;
-    grpc_security_handshake_shutdown(
-        exec_ctx, connector->handshaking_handshakes->handshake);
-    connector->handshaking_handshakes = tmp->next;
-    gpr_free(tmp);
+void grpc_channel_security_connector_create_handshakers(
+    grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *connector,
+    grpc_handshake_manager *handshake_mgr) {
+  if (connector != NULL) {
+    connector->create_handshakers(exec_ctx, connector, handshake_mgr);
   }
-  gpr_mu_unlock(&connector->mu);
 }
 
-void grpc_channel_security_connector_do_handshake(
-    grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc,
-    grpc_endpoint *nonsecure_endpoint, grpc_slice_buffer *read_buffer,
-    gpr_timespec deadline, grpc_security_handshake_done_cb cb,
-    void *user_data) {
-  if (sc == NULL || nonsecure_endpoint == NULL) {
-    gpr_free(read_buffer);
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL);
-  } else {
-    sc->do_handshake(exec_ctx, sc, nonsecure_endpoint, read_buffer, deadline,
-                     cb, user_data);
-  }
-}
-
-void grpc_server_security_connector_do_handshake(
-    grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc,
-    grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint,
-    grpc_slice_buffer *read_buffer, gpr_timespec deadline,
-    grpc_security_handshake_done_cb cb, void *user_data) {
-  if (sc == NULL || nonsecure_endpoint == NULL) {
-    gpr_free(read_buffer);
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL);
-  } else {
-    sc->do_handshake(exec_ctx, sc, acceptor, nonsecure_endpoint, read_buffer,
-                     deadline, cb, user_data);
+void grpc_server_security_connector_create_handshakers(
+    grpc_exec_ctx *exec_ctx, grpc_server_security_connector *connector,
+    grpc_handshake_manager *handshake_mgr) {
+  if (connector != NULL) {
+    connector->create_handshakers(exec_ctx, connector, handshake_mgr);
   }
 }
 
 void grpc_security_connector_check_peer(grpc_exec_ctx *exec_ctx,
                                         grpc_security_connector *sc,
                                         tsi_peer peer,
-                                        grpc_security_peer_check_cb cb,
-                                        void *user_data) {
+                                        grpc_auth_context **auth_context,
+                                        grpc_closure *on_peer_checked) {
   if (sc == NULL) {
-    cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL);
+    grpc_exec_ctx_sched(
+        exec_ctx, on_peer_checked,
+        GRPC_ERROR_CREATE("cannot check peer -- no security connector"), NULL);
     tsi_peer_destruct(&peer);
   } else {
-    sc->vtable->check_peer(exec_ctx, sc, peer, cb, user_data);
+    sc->vtable->check_peer(exec_ctx, sc, peer, auth_context, on_peer_checked);
   }
 }
 
@@ -267,44 +243,42 @@ static void fake_channel_destroy(grpc_exec_ctx *exec_ctx,
 
 static void fake_server_destroy(grpc_exec_ctx *exec_ctx,
                                 grpc_security_connector *sc) {
-  grpc_server_security_connector *c = (grpc_server_security_connector *)sc;
-  gpr_mu_destroy(&c->mu);
   gpr_free(sc);
 }
 
 static void fake_check_peer(grpc_exec_ctx *exec_ctx,
                             grpc_security_connector *sc, tsi_peer peer,
-                            grpc_security_peer_check_cb cb, void *user_data) {
+                            grpc_auth_context **auth_context,
+                            grpc_closure *on_peer_checked) {
   const char *prop_name;
-  grpc_security_status status = GRPC_SECURITY_OK;
-  grpc_auth_context *auth_context = NULL;
+  grpc_error *error = GRPC_ERROR_NONE;
+  *auth_context = NULL;
   if (peer.property_count != 1) {
-    gpr_log(GPR_ERROR, "Fake peers should only have 1 property.");
-    status = GRPC_SECURITY_ERROR;
+    error = GRPC_ERROR_CREATE("Fake peers should only have 1 property.");
     goto end;
   }
   prop_name = peer.properties[0].name;
   if (prop_name == NULL ||
       strcmp(prop_name, TSI_CERTIFICATE_TYPE_PEER_PROPERTY)) {
-    gpr_log(GPR_ERROR, "Unexpected property in fake peer: %s.",
-            prop_name == NULL ? "<EMPTY>" : prop_name);
-    status = GRPC_SECURITY_ERROR;
+    char *msg;
+    gpr_asprintf(&msg, "Unexpected property in fake peer: %s.",
+                 prop_name == NULL ? "<EMPTY>" : prop_name);
+    error = GRPC_ERROR_CREATE(msg);
+    gpr_free(msg);
     goto end;
   }
   if (strncmp(peer.properties[0].value.data, TSI_FAKE_CERTIFICATE_TYPE,
               peer.properties[0].value.length)) {
-    gpr_log(GPR_ERROR, "Invalid value for cert type property.");
-    status = GRPC_SECURITY_ERROR;
+    error = GRPC_ERROR_CREATE("Invalid value for cert type property.");
     goto end;
   }
-  auth_context = grpc_auth_context_create(NULL);
+  *auth_context = grpc_auth_context_create(NULL);
   grpc_auth_context_add_cstring_property(
-      auth_context, GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME,
+      *auth_context, GRPC_TRANSPORT_SECURITY_TYPE_PROPERTY_NAME,
       GRPC_FAKE_TRANSPORT_SECURITY_TYPE);
 
 end:
-  cb(exec_ctx, user_data, status, auth_context);
-  grpc_auth_context_unref(auth_context);
+  grpc_exec_ctx_sched(exec_ctx, on_peer_checked, error, NULL);
   tsi_peer_destruct(&peer);
 }
 
@@ -317,26 +291,20 @@ static void fake_channel_check_call_host(grpc_exec_ctx *exec_ctx,
   cb(exec_ctx, user_data, GRPC_SECURITY_OK);
 }
 
-static void fake_channel_do_handshake(grpc_exec_ctx *exec_ctx,
-                                      grpc_channel_security_connector *sc,
-                                      grpc_endpoint *nonsecure_endpoint,
-                                      grpc_slice_buffer *read_buffer,
-                                      gpr_timespec deadline,
-                                      grpc_security_handshake_done_cb cb,
-                                      void *user_data) {
-  grpc_do_security_handshake(exec_ctx, tsi_create_fake_handshaker(1), &sc->base,
-                             true, nonsecure_endpoint, read_buffer, deadline,
-                             cb, user_data);
+static void fake_channel_create_handshakers(
+    grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc,
+    grpc_handshake_manager *handshake_mgr) {
+  grpc_security_create_handshakers(
+      exec_ctx, tsi_create_fake_handshaker(true /* is_client */), &sc->base,
+      handshake_mgr);
 }
 
-static void fake_server_do_handshake(
+static void fake_server_create_handshakers(
     grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc,
-    grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint,
-    grpc_slice_buffer *read_buffer, gpr_timespec deadline,
-    grpc_security_handshake_done_cb cb, void *user_data) {
-  grpc_do_security_handshake(exec_ctx, tsi_create_fake_handshaker(0), &sc->base,
-                             false, nonsecure_endpoint, read_buffer, deadline,
-                             cb, user_data);
+    grpc_handshake_manager *handshake_mgr) {
+  grpc_security_create_handshakers(
+      exec_ctx, tsi_create_fake_handshaker(false /* is_client */), &sc->base,
+      handshake_mgr);
 }
 
 static grpc_security_connector_vtable fake_channel_vtable = {
@@ -354,7 +322,7 @@ grpc_channel_security_connector *grpc_fake_channel_security_connector_create(
   c->base.vtable = &fake_channel_vtable;
   c->request_metadata_creds = grpc_call_credentials_ref(request_metadata_creds);
   c->check_call_host = fake_channel_check_call_host;
-  c->do_handshake = fake_channel_do_handshake;
+  c->create_handshakers = fake_channel_create_handshakers;
   return c;
 }
 
@@ -366,8 +334,7 @@ grpc_server_security_connector *grpc_fake_server_security_connector_create(
   gpr_ref_init(&c->base.refcount, 1);
   c->base.vtable = &fake_server_vtable;
   c->base.url_scheme = GRPC_FAKE_SECURITY_URL_SCHEME;
-  c->do_handshake = fake_server_do_handshake;
-  gpr_mu_init(&c->mu);
+  c->create_handshakers = fake_server_create_handshakers;
   return c;
 }
 
@@ -402,11 +369,9 @@ static void ssl_server_destroy(grpc_exec_ctx *exec_ctx,
                                grpc_security_connector *sc) {
   grpc_ssl_server_security_connector *c =
       (grpc_ssl_server_security_connector *)sc;
-
   if (c->handshaker_factory != NULL) {
     tsi_ssl_handshaker_factory_destroy(c->handshaker_factory);
   }
-  gpr_mu_destroy(&c->base.mu);
   gpr_free(sc);
 }
 
@@ -425,49 +390,33 @@ static grpc_security_status ssl_create_handshaker(
   return GRPC_SECURITY_OK;
 }
 
-static void ssl_channel_do_handshake(grpc_exec_ctx *exec_ctx,
-                                     grpc_channel_security_connector *sc,
-                                     grpc_endpoint *nonsecure_endpoint,
-                                     grpc_slice_buffer *read_buffer,
-                                     gpr_timespec deadline,
-                                     grpc_security_handshake_done_cb cb,
-                                     void *user_data) {
+static void ssl_channel_create_handshakers(
+    grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc,
+    grpc_handshake_manager *handshake_mgr) {
   grpc_ssl_channel_security_connector *c =
       (grpc_ssl_channel_security_connector *)sc;
-  tsi_handshaker *handshaker;
-  grpc_security_status status = ssl_create_handshaker(
-      c->handshaker_factory, true,
-      c->overridden_target_name != NULL ? c->overridden_target_name
-                                        : c->target_name,
-      &handshaker);
-  if (status != GRPC_SECURITY_OK) {
-    gpr_free(read_buffer);
-    cb(exec_ctx, user_data, status, NULL, NULL);
-  } else {
-    grpc_do_security_handshake(exec_ctx, handshaker, &sc->base, true,
-                               nonsecure_endpoint, read_buffer, deadline, cb,
-                               user_data);
-  }
-}
-
-static void ssl_server_do_handshake(
+  // Instantiate TSI handshaker.
+  tsi_handshaker *tsi_hs = NULL;
+  ssl_create_handshaker(c->handshaker_factory, true /* is_client */,
+                        c->overridden_target_name != NULL
+                            ? c->overridden_target_name
+                            : c->target_name,
+                        &tsi_hs);
+  // Create handshakers.
+  grpc_security_create_handshakers(exec_ctx, tsi_hs, &sc->base, handshake_mgr);
+}
+
+static void ssl_server_create_handshakers(
     grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc,
-    grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint,
-    grpc_slice_buffer *read_buffer, gpr_timespec deadline,
-    grpc_security_handshake_done_cb cb, void *user_data) {
+    grpc_handshake_manager *handshake_mgr) {
   grpc_ssl_server_security_connector *c =
       (grpc_ssl_server_security_connector *)sc;
-  tsi_handshaker *handshaker;
-  grpc_security_status status =
-      ssl_create_handshaker(c->handshaker_factory, false, NULL, &handshaker);
-  if (status != GRPC_SECURITY_OK) {
-    gpr_free(read_buffer);
-    cb(exec_ctx, user_data, status, NULL, NULL);
-  } else {
-    grpc_do_security_handshake(exec_ctx, handshaker, &sc->base, false,
-                               nonsecure_endpoint, read_buffer, deadline, cb,
-                               user_data);
-  }
+  // Instantiate TSI handshaker.
+  tsi_handshaker *tsi_hs = NULL;
+  ssl_create_handshaker(c->handshaker_factory, false /* is_client */,
+                        NULL /* peer_name */, &tsi_hs);
+  // Create handshakers.
+  grpc_security_create_handshakers(exec_ctx, tsi_hs, &sc->base, handshake_mgr);
 }
 
 static int ssl_host_matches_name(const tsi_peer *peer, const char *peer_name) {
@@ -524,57 +473,53 @@ grpc_auth_context *tsi_ssl_peer_to_auth_context(const tsi_peer *peer) {
   return ctx;
 }
 
-static grpc_security_status ssl_check_peer(grpc_security_connector *sc,
-                                           const char *peer_name,
-                                           const tsi_peer *peer,
-                                           grpc_auth_context **auth_context) {
+static grpc_error *ssl_check_peer(grpc_security_connector *sc,
+                                  const char *peer_name, const tsi_peer *peer,
+                                  grpc_auth_context **auth_context) {
   /* Check the ALPN. */
   const tsi_peer_property *p =
       tsi_peer_get_property_by_name(peer, TSI_SSL_ALPN_SELECTED_PROTOCOL);
   if (p == NULL) {
-    gpr_log(GPR_ERROR, "Missing selected ALPN property.");
-    return GRPC_SECURITY_ERROR;
+    return GRPC_ERROR_CREATE(
+        "Cannot check peer: missing selected ALPN property.");
   }
   if (!grpc_chttp2_is_alpn_version_supported(p->value.data, p->value.length)) {
-    gpr_log(GPR_ERROR, "Invalid ALPN value.");
-    return GRPC_SECURITY_ERROR;
+    return GRPC_ERROR_CREATE("Cannot check peer: invalid ALPN value.");
   }
 
   /* Check the peer name if specified. */
   if (peer_name != NULL && !ssl_host_matches_name(peer, peer_name)) {
-    gpr_log(GPR_ERROR, "Peer name %s is not in peer certificate", peer_name);
-    return GRPC_SECURITY_ERROR;
+    char *msg;
+    gpr_asprintf(&msg, "Peer name %s is not in peer certificate", peer_name);
+    grpc_error *error = GRPC_ERROR_CREATE(msg);
+    gpr_free(msg);
+    return error;
   }
   *auth_context = tsi_ssl_peer_to_auth_context(peer);
-  return GRPC_SECURITY_OK;
+  return GRPC_ERROR_NONE;
 }
 
 static void ssl_channel_check_peer(grpc_exec_ctx *exec_ctx,
                                    grpc_security_connector *sc, tsi_peer peer,
-                                   grpc_security_peer_check_cb cb,
-                                   void *user_data) {
+                                   grpc_auth_context **auth_context,
+                                   grpc_closure *on_peer_checked) {
   grpc_ssl_channel_security_connector *c =
       (grpc_ssl_channel_security_connector *)sc;
-  grpc_security_status status;
-  grpc_auth_context *auth_context = NULL;
-  status = ssl_check_peer(sc, c->overridden_target_name != NULL
-                                  ? c->overridden_target_name
-                                  : c->target_name,
-                          &peer, &auth_context);
-  cb(exec_ctx, user_data, status, auth_context);
-  grpc_auth_context_unref(auth_context);
+  grpc_error *error = ssl_check_peer(sc, c->overridden_target_name != NULL
+                                             ? c->overridden_target_name
+                                             : c->target_name,
+                                     &peer, auth_context);
+  grpc_exec_ctx_sched(exec_ctx, on_peer_checked, error, NULL);
   tsi_peer_destruct(&peer);
 }
 
 static void ssl_server_check_peer(grpc_exec_ctx *exec_ctx,
                                   grpc_security_connector *sc, tsi_peer peer,
-                                  grpc_security_peer_check_cb cb,
-                                  void *user_data) {
-  grpc_auth_context *auth_context = NULL;
-  grpc_security_status status = ssl_check_peer(sc, NULL, &peer, &auth_context);
+                                  grpc_auth_context **auth_context,
+                                  grpc_closure *on_peer_checked) {
+  grpc_error *error = ssl_check_peer(sc, NULL, &peer, auth_context);
   tsi_peer_destruct(&peer);
-  cb(exec_ctx, user_data, status, auth_context);
-  grpc_auth_context_unref(auth_context);
+  grpc_exec_ctx_sched(exec_ctx, on_peer_checked, error, NULL);
 }
 
 static void add_shallow_auth_property_to_peer(tsi_peer *peer,
@@ -771,7 +716,7 @@ grpc_security_status grpc_ssl_channel_security_connector_create(
   c->base.request_metadata_creds =
       grpc_call_credentials_ref(request_metadata_creds);
   c->base.check_call_host = ssl_channel_check_call_host;
-  c->base.do_handshake = ssl_channel_do_handshake;
+  c->base.create_handshakers = ssl_channel_create_handshakers;
   gpr_split_host_port(target_name, &c->target_name, &port);
   gpr_free(port);
   if (overridden_target_name != NULL) {
@@ -847,8 +792,7 @@ grpc_security_status grpc_ssl_server_security_connector_create(
     *sc = NULL;
     goto error;
   }
-  gpr_mu_init(&c->base.mu);
-  c->base.do_handshake = ssl_server_do_handshake;
+  c->base.create_handshakers = ssl_server_create_handshakers;
   *sc = &c->base;
   gpr_free((void *)alpn_protocol_strings);
   gpr_free(alpn_protocol_string_lengths);
diff --git a/src/core/lib/security/transport/security_connector.h b/src/core/lib/security/transport/security_connector.h
index 6e89bfd779..df77434a65 100644
--- a/src/core/lib/security/transport/security_connector.h
+++ b/src/core/lib/security/transport/security_connector.h
@@ -35,6 +35,8 @@
 #define GRPC_CORE_LIB_SECURITY_TRANSPORT_SECURITY_CONNECTOR_H
 
 #include <grpc/grpc_security.h>
+
+#include "src/core/lib/channel/handshaker.h"
 #include "src/core/lib/iomgr/endpoint.h"
 #include "src/core/lib/iomgr/tcp_server.h"
 #include "src/core/lib/tsi/transport_security_interface.h"
@@ -57,21 +59,11 @@ typedef struct grpc_security_connector grpc_security_connector;
 
 #define GRPC_SECURITY_CONNECTOR_ARG "grpc.security_connector"
 
-typedef void (*grpc_security_peer_check_cb)(grpc_exec_ctx *exec_ctx,
-                                            void *user_data,
-                                            grpc_security_status status,
-                                            grpc_auth_context *auth_context);
-
-/* Ownership of the secure_endpoint is transfered. */
-typedef void (*grpc_security_handshake_done_cb)(
-    grpc_exec_ctx *exec_ctx, void *user_data, grpc_security_status status,
-    grpc_endpoint *secure_endpoint, grpc_auth_context *auth_context);
-
 typedef struct {
   void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_security_connector *sc);
   void (*check_peer)(grpc_exec_ctx *exec_ctx, grpc_security_connector *sc,
-                     tsi_peer peer, grpc_security_peer_check_cb cb,
-                     void *user_data);
+                     tsi_peer peer, grpc_auth_context **auth_context,
+                     grpc_closure *on_peer_checked);
 } grpc_security_connector_vtable;
 
 typedef struct grpc_security_connector_handshake_list {
@@ -109,12 +101,12 @@ void grpc_security_connector_unref(grpc_exec_ctx *exec_ctx,
 #endif
 
 /* Check the peer. Callee takes ownership of the peer object.
-   The callback will include the resulting auth_context. */
+   Sets *auth_context and invokes on_peer_checked when done. */
 void grpc_security_connector_check_peer(grpc_exec_ctx *exec_ctx,
                                         grpc_security_connector *sc,
                                         tsi_peer peer,
-                                        grpc_security_peer_check_cb cb,
-                                        void *user_data);
+                                        grpc_auth_context **auth_context,
+                                        grpc_closure *on_peer_checked);
 
 /* Util to encapsulate the connector in a channel arg. */
 grpc_arg grpc_security_connector_to_arg(grpc_security_connector *sc);
@@ -144,11 +136,9 @@ struct grpc_channel_security_connector {
                           grpc_channel_security_connector *sc, const char *host,
                           grpc_auth_context *auth_context,
                           grpc_security_call_host_check_cb cb, void *user_data);
-  void (*do_handshake)(grpc_exec_ctx *exec_ctx,
-                       grpc_channel_security_connector *sc,
-                       grpc_endpoint *nonsecure_endpoint,
-                       grpc_slice_buffer *read_buffer, gpr_timespec deadline,
-                       grpc_security_handshake_done_cb cb, void *user_data);
+  void (*create_handshakers)(grpc_exec_ctx *exec_ctx,
+                             grpc_channel_security_connector *sc,
+                             grpc_handshake_manager *handshake_mgr);
 };
 
 /* Checks that the host that will be set for a call is acceptable. */
@@ -157,11 +147,10 @@ void grpc_channel_security_connector_check_call_host(
     const char *host, grpc_auth_context *auth_context,
     grpc_security_call_host_check_cb cb, void *user_data);
 
-/* Handshake. */
-void grpc_channel_security_connector_do_handshake(
+/* Registers handshakers with \a handshake_mgr. */
+void grpc_channel_security_connector_create_handshakers(
     grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *connector,
-    grpc_endpoint *nonsecure_endpoint, grpc_slice_buffer *read_buffer,
-    gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data);
+    grpc_handshake_manager *handshake_mgr);
 
 /* --- server_security_connector object. ---
 
@@ -172,25 +161,14 @@ typedef struct grpc_server_security_connector grpc_server_security_connector;
 
 struct grpc_server_security_connector {
   grpc_security_connector base;
-  gpr_mu mu;
-  grpc_security_connector_handshake_list *handshaking_handshakes;
-  const grpc_channel_args *channel_args;
-  void (*do_handshake)(grpc_exec_ctx *exec_ctx,
-                       grpc_server_security_connector *sc,
-                       grpc_tcp_server_acceptor *acceptor,
-                       grpc_endpoint *nonsecure_endpoint,
-                       grpc_slice_buffer *read_buffer, gpr_timespec deadline,
-                       grpc_security_handshake_done_cb cb, void *user_data);
+  void (*create_handshakers)(grpc_exec_ctx *exec_ctx,
+                             grpc_server_security_connector *sc,
+                             grpc_handshake_manager *handshake_mgr);
 };
 
-void grpc_server_security_connector_do_handshake(
+void grpc_server_security_connector_create_handshakers(
     grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc,
-    grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint,
-    grpc_slice_buffer *read_buffer, gpr_timespec deadline,
-    grpc_security_handshake_done_cb cb, void *user_data);
-
-void grpc_server_security_connector_shutdown(
-    grpc_exec_ctx *exec_ctx, grpc_server_security_connector *connector);
+    grpc_handshake_manager *handshake_mgr);
 
 /* --- Creation security connectors. --- */
 
diff --git a/src/core/lib/security/transport/security_handshaker.c b/src/core/lib/security/transport/security_handshaker.c
new file mode 100644
index 0000000000..e96c480044
--- /dev/null
+++ b/src/core/lib/security/transport/security_handshaker.c
@@ -0,0 +1,452 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "src/core/lib/security/transport/security_handshaker.h"
+
+#include <stdbool.h>
+#include <string.h>
+
+#include <grpc/slice_buffer.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/channel/handshaker.h"
+#include "src/core/lib/security/context/security_context.h"
+#include "src/core/lib/security/transport/secure_endpoint.h"
+#include "src/core/lib/security/transport/tsi_error.h"
+#include "src/core/lib/slice/slice_internal.h"
+
+#define GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE 256
+
+typedef struct {
+  grpc_handshaker base;
+
+  // State set at creation time.
+  tsi_handshaker *handshaker;
+  grpc_security_connector *connector;
+
+  gpr_mu mu;
+  gpr_refcount refs;
+
+  bool shutdown;
+  // Endpoint and read buffer to destroy after a shutdown.
+  grpc_endpoint *endpoint_to_destroy;
+  grpc_slice_buffer *read_buffer_to_destroy;
+
+  // State saved while performing the handshake.
+  grpc_handshaker_args *args;
+  grpc_closure *on_handshake_done;
+
+  unsigned char *handshake_buffer;
+  size_t handshake_buffer_size;
+  grpc_slice_buffer left_overs;
+  grpc_slice_buffer outgoing;
+  grpc_closure on_handshake_data_sent_to_peer;
+  grpc_closure on_handshake_data_received_from_peer;
+  grpc_closure on_peer_checked;
+  grpc_auth_context *auth_context;
+} security_handshaker;
+
+static void security_handshaker_unref(grpc_exec_ctx *exec_ctx,
+                                      security_handshaker *h) {
+  if (gpr_unref(&h->refs)) {
+    gpr_mu_destroy(&h->mu);
+    tsi_handshaker_destroy(h->handshaker);
+    if (h->endpoint_to_destroy != NULL) {
+      grpc_endpoint_destroy(exec_ctx, h->endpoint_to_destroy);
+    }
+    if (h->read_buffer_to_destroy != NULL) {
+      grpc_slice_buffer_destroy_internal(exec_ctx, h->read_buffer_to_destroy);
+      gpr_free(h->read_buffer_to_destroy);
+    }
+    gpr_free(h->handshake_buffer);
+    grpc_slice_buffer_destroy_internal(exec_ctx, &h->left_overs);
+    grpc_slice_buffer_destroy_internal(exec_ctx, &h->outgoing);
+    GRPC_AUTH_CONTEXT_UNREF(h->auth_context, "handshake");
+    GRPC_SECURITY_CONNECTOR_UNREF(exec_ctx, h->connector, "handshake");
+    gpr_free(h);
+  }
+}
+
+// Set args fields to NULL, saving the endpoint and read buffer for
+// later destruction.
+static void cleanup_args_for_failure_locked(grpc_exec_ctx *exec_ctx,
+                                            security_handshaker *h) {
+  h->endpoint_to_destroy = h->args->endpoint;
+  h->args->endpoint = NULL;
+  h->read_buffer_to_destroy = h->args->read_buffer;
+  h->args->read_buffer = NULL;
+  grpc_channel_args_destroy(exec_ctx, h->args->args);
+  h->args->args = NULL;
+}
+
+// If the handshake failed or we're shutting down, clean up and invoke the
+// callback with the error.
+static void security_handshake_failed_locked(grpc_exec_ctx *exec_ctx,
+                                             security_handshaker *h,
+                                             grpc_error *error) {
+  if (error == GRPC_ERROR_NONE) {
+    // If we were shut down after the handshake succeeded but before an
+    // endpoint callback was invoked, we need to generate our own error.
+    error = GRPC_ERROR_CREATE("Handshaker shutdown");
+  }
+  const char *msg = grpc_error_string(error);
+  gpr_log(GPR_DEBUG, "Security handshake failed: %s", msg);
+  grpc_error_free_string(msg);
+  if (!h->shutdown) {
+    // TODO(ctiller): It is currently necessary to shutdown endpoints
+    // before destroying them, even if we know that there are no
+    // pending read/write callbacks.  This should be fixed, at which
+    // point this can be removed.
+    grpc_endpoint_shutdown(exec_ctx, h->args->endpoint);
+    // Not shutting down, so the write failed.  Clean up before
+    // invoking the callback.
+    cleanup_args_for_failure_locked(exec_ctx, h);
+  }
+  // Invoke callback.
+  grpc_exec_ctx_sched(exec_ctx, h->on_handshake_done, error, NULL);
+}
+
+static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
+                            grpc_error *error) {
+  security_handshaker *h = arg;
+  gpr_mu_lock(&h->mu);
+  if (error != GRPC_ERROR_NONE || h->shutdown) {
+    security_handshake_failed_locked(exec_ctx, h, GRPC_ERROR_REF(error));
+    goto done;
+  }
+  // Get frame protector.
+  tsi_frame_protector *protector;
+  tsi_result result =
+      tsi_handshaker_create_frame_protector(h->handshaker, NULL, &protector);
+  if (result != TSI_OK) {
+    error = grpc_set_tsi_error_result(
+        GRPC_ERROR_CREATE("Frame protector creation failed"), result);
+    security_handshake_failed_locked(exec_ctx, h, error);
+    goto done;
+  }
+  // Success.
+  // Create secure endpoint.
+  h->args->endpoint = grpc_secure_endpoint_create(
+      protector, h->args->endpoint, h->left_overs.slices, h->left_overs.count);
+  h->left_overs.count = 0;
+  h->left_overs.length = 0;
+  // Clear out the read buffer before it gets passed to the transport,
+  // since any excess bytes were already copied to h->left_overs.
+  grpc_slice_buffer_reset_and_unref_internal(exec_ctx, h->args->read_buffer);
+  // Add auth context to channel args.
+  grpc_arg auth_context_arg = grpc_auth_context_to_arg(h->auth_context);
+  grpc_channel_args *tmp_args = h->args->args;
+  h->args->args =
+      grpc_channel_args_copy_and_add(tmp_args, &auth_context_arg, 1);
+  grpc_channel_args_destroy(exec_ctx, tmp_args);
+  // Invoke callback.
+  grpc_exec_ctx_sched(exec_ctx, h->on_handshake_done, GRPC_ERROR_NONE, NULL);
+  // Set shutdown to true so that subsequent calls to
+  // security_handshaker_shutdown() do nothing.
+  h->shutdown = true;
+done:
+  gpr_mu_unlock(&h->mu);
+  security_handshaker_unref(exec_ctx, h);
+}
+
+static grpc_error *check_peer_locked(grpc_exec_ctx *exec_ctx,
+                                     security_handshaker *h) {
+  tsi_peer peer;
+  tsi_result result = tsi_handshaker_extract_peer(h->handshaker, &peer);
+  if (result != TSI_OK) {
+    return grpc_set_tsi_error_result(
+        GRPC_ERROR_CREATE("Peer extraction failed"), result);
+  }
+  grpc_security_connector_check_peer(exec_ctx, h->connector, peer,
+                                     &h->auth_context, &h->on_peer_checked);
+  return GRPC_ERROR_NONE;
+}
+
+static grpc_error *send_handshake_bytes_to_peer_locked(grpc_exec_ctx *exec_ctx,
+                                                       security_handshaker *h) {
+  // Get data to send.
+  tsi_result result = TSI_OK;
+  size_t offset = 0;
+  do {
+    size_t to_send_size = h->handshake_buffer_size - offset;
+    result = tsi_handshaker_get_bytes_to_send_to_peer(
+        h->handshaker, h->handshake_buffer + offset, &to_send_size);
+    offset += to_send_size;
+    if (result == TSI_INCOMPLETE_DATA) {
+      h->handshake_buffer_size *= 2;
+      h->handshake_buffer =
+          gpr_realloc(h->handshake_buffer, h->handshake_buffer_size);
+    }
+  } while (result == TSI_INCOMPLETE_DATA);
+  if (result != TSI_OK) {
+    return grpc_set_tsi_error_result(GRPC_ERROR_CREATE("Handshake failed"),
+                                     result);
+  }
+  // Send data.
+  grpc_slice to_send =
+      grpc_slice_from_copied_buffer((const char *)h->handshake_buffer, offset);
+  grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &h->outgoing);
+  grpc_slice_buffer_add(&h->outgoing, to_send);
+  grpc_endpoint_write(exec_ctx, h->args->endpoint, &h->outgoing,
+                      &h->on_handshake_data_sent_to_peer);
+  return GRPC_ERROR_NONE;
+}
+
+static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
+                                                 void *arg, grpc_error *error) {
+  security_handshaker *h = arg;
+  gpr_mu_lock(&h->mu);
+  if (error != GRPC_ERROR_NONE || h->shutdown) {
+    security_handshake_failed_locked(
+        exec_ctx, h,
+        GRPC_ERROR_CREATE_REFERENCING("Handshake read failed", &error, 1));
+    gpr_mu_unlock(&h->mu);
+    security_handshaker_unref(exec_ctx, h);
+    return;
+  }
+  // Process received data.
+  tsi_result result = TSI_OK;
+  size_t consumed_slice_size = 0;
+  size_t i;
+  for (i = 0; i < h->args->read_buffer->count; i++) {
+    consumed_slice_size = GRPC_SLICE_LENGTH(h->args->read_buffer->slices[i]);
+    result = tsi_handshaker_process_bytes_from_peer(
+        h->handshaker, GRPC_SLICE_START_PTR(h->args->read_buffer->slices[i]),
+        &consumed_slice_size);
+    if (!tsi_handshaker_is_in_progress(h->handshaker)) break;
+  }
+  if (tsi_handshaker_is_in_progress(h->handshaker)) {
+    /* We may need more data. */
+    if (result == TSI_INCOMPLETE_DATA) {
+      grpc_endpoint_read(exec_ctx, h->args->endpoint, h->args->read_buffer,
+                         &h->on_handshake_data_received_from_peer);
+      goto done;
+    } else {
+      error = send_handshake_bytes_to_peer_locked(exec_ctx, h);
+      if (error != GRPC_ERROR_NONE) {
+        security_handshake_failed_locked(exec_ctx, h, error);
+        gpr_mu_unlock(&h->mu);
+        security_handshaker_unref(exec_ctx, h);
+        return;
+      }
+      goto done;
+    }
+  }
+  if (result != TSI_OK) {
+    security_handshake_failed_locked(
+        exec_ctx, h, grpc_set_tsi_error_result(
+                         GRPC_ERROR_CREATE("Handshake failed"), result));
+    gpr_mu_unlock(&h->mu);
+    security_handshaker_unref(exec_ctx, h);
+    return;
+  }
+  /* Handshake is done and successful this point. */
+  bool has_left_overs_in_current_slice =
+      (consumed_slice_size <
+       GRPC_SLICE_LENGTH(h->args->read_buffer->slices[i]));
+  size_t num_left_overs = (has_left_overs_in_current_slice ? 1 : 0) +
+                          h->args->read_buffer->count - i - 1;
+  if (num_left_overs > 0) {
+    /* Put the leftovers in our buffer (ownership transfered). */
+    if (has_left_overs_in_current_slice) {
+      grpc_slice_buffer_add(
+          &h->left_overs,
+          grpc_slice_split_tail(&h->args->read_buffer->slices[i],
+                                consumed_slice_size));
+      /* split_tail above increments refcount. */
+      grpc_slice_unref_internal(exec_ctx, h->args->read_buffer->slices[i]);
+    }
+    grpc_slice_buffer_addn(
+        &h->left_overs, &h->args->read_buffer->slices[i + 1],
+        num_left_overs - (size_t)has_left_overs_in_current_slice);
+  }
+  // Check peer.
+  error = check_peer_locked(exec_ctx, h);
+  if (error != GRPC_ERROR_NONE) {
+    security_handshake_failed_locked(exec_ctx, h, error);
+    gpr_mu_unlock(&h->mu);
+    security_handshaker_unref(exec_ctx, h);
+    return;
+  }
+done:
+  gpr_mu_unlock(&h->mu);
+}
+
+static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx, void *arg,
+                                           grpc_error *error) {
+  security_handshaker *h = arg;
+  gpr_mu_lock(&h->mu);
+  if (error != GRPC_ERROR_NONE || h->shutdown) {
+    security_handshake_failed_locked(
+        exec_ctx, h,
+        GRPC_ERROR_CREATE_REFERENCING("Handshake write failed", &error, 1));
+    gpr_mu_unlock(&h->mu);
+    security_handshaker_unref(exec_ctx, h);
+    return;
+  }
+  /* We may be done. */
+  if (tsi_handshaker_is_in_progress(h->handshaker)) {
+    grpc_endpoint_read(exec_ctx, h->args->endpoint, h->args->read_buffer,
+                       &h->on_handshake_data_received_from_peer);
+  } else {
+    error = check_peer_locked(exec_ctx, h);
+    if (error != GRPC_ERROR_NONE) {
+      security_handshake_failed_locked(exec_ctx, h, error);
+      gpr_mu_unlock(&h->mu);
+      security_handshaker_unref(exec_ctx, h);
+      return;
+    }
+  }
+  gpr_mu_unlock(&h->mu);
+}
+
+//
+// public handshaker API
+//
+
+static void security_handshaker_destroy(grpc_exec_ctx *exec_ctx,
+                                        grpc_handshaker *handshaker) {
+  security_handshaker *h = (security_handshaker *)handshaker;
+  security_handshaker_unref(exec_ctx, h);
+}
+
+static void security_handshaker_shutdown(grpc_exec_ctx *exec_ctx,
+                                         grpc_handshaker *handshaker) {
+  security_handshaker *h = (security_handshaker *)handshaker;
+  gpr_mu_lock(&h->mu);
+  if (!h->shutdown) {
+    h->shutdown = true;
+    grpc_endpoint_shutdown(exec_ctx, h->args->endpoint);
+    cleanup_args_for_failure_locked(exec_ctx, h);
+  }
+  gpr_mu_unlock(&h->mu);
+}
+
+static void security_handshaker_do_handshake(grpc_exec_ctx *exec_ctx,
+                                             grpc_handshaker *handshaker,
+                                             grpc_tcp_server_acceptor *acceptor,
+                                             grpc_closure *on_handshake_done,
+                                             grpc_handshaker_args *args) {
+  security_handshaker *h = (security_handshaker *)handshaker;
+  gpr_mu_lock(&h->mu);
+  h->args = args;
+  h->on_handshake_done = on_handshake_done;
+  gpr_ref(&h->refs);
+  grpc_error *error = send_handshake_bytes_to_peer_locked(exec_ctx, h);
+  if (error != GRPC_ERROR_NONE) {
+    security_handshake_failed_locked(exec_ctx, h, error);
+    gpr_mu_unlock(&h->mu);
+    security_handshaker_unref(exec_ctx, h);
+    return;
+  }
+  gpr_mu_unlock(&h->mu);
+}
+
+static const grpc_handshaker_vtable security_handshaker_vtable = {
+    security_handshaker_destroy, security_handshaker_shutdown,
+    security_handshaker_do_handshake};
+
+static grpc_handshaker *security_handshaker_create(
+    grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker,
+    grpc_security_connector *connector) {
+  security_handshaker *h = gpr_malloc(sizeof(security_handshaker));
+  memset(h, 0, sizeof(security_handshaker));
+  grpc_handshaker_init(&security_handshaker_vtable, &h->base);
+  h->handshaker = handshaker;
+  h->connector = GRPC_SECURITY_CONNECTOR_REF(connector, "handshake");
+  gpr_mu_init(&h->mu);
+  gpr_ref_init(&h->refs, 1);
+  h->handshake_buffer_size = GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE;
+  h->handshake_buffer = gpr_malloc(h->handshake_buffer_size);
+  grpc_closure_init(&h->on_handshake_data_sent_to_peer,
+                    on_handshake_data_sent_to_peer, h);
+  grpc_closure_init(&h->on_handshake_data_received_from_peer,
+                    on_handshake_data_received_from_peer, h);
+  grpc_closure_init(&h->on_peer_checked, on_peer_checked, h);
+  grpc_slice_buffer_init(&h->left_overs);
+  grpc_slice_buffer_init(&h->outgoing);
+  return &h->base;
+}
+
+//
+// fail_handshaker
+//
+
+static void fail_handshaker_destroy(grpc_exec_ctx *exec_ctx,
+                                    grpc_handshaker *handshaker) {
+  gpr_free(handshaker);
+}
+
+static void fail_handshaker_shutdown(grpc_exec_ctx *exec_ctx,
+                                     grpc_handshaker *handshaker) {}
+
+static void fail_handshaker_do_handshake(grpc_exec_ctx *exec_ctx,
+                                         grpc_handshaker *handshaker,
+                                         grpc_tcp_server_acceptor *acceptor,
+                                         grpc_closure *on_handshake_done,
+                                         grpc_handshaker_args *args) {
+  grpc_exec_ctx_sched(exec_ctx, on_handshake_done,
+                      GRPC_ERROR_CREATE("Failed to create security handshaker"),
+                      NULL);
+}
+
+static const grpc_handshaker_vtable fail_handshaker_vtable = {
+    fail_handshaker_destroy, fail_handshaker_shutdown,
+    fail_handshaker_do_handshake};
+
+static grpc_handshaker *fail_handshaker_create() {
+  grpc_handshaker *h = gpr_malloc(sizeof(*h));
+  grpc_handshaker_init(&fail_handshaker_vtable, h);
+  return h;
+}
+
+//
+// exported functions
+//
+
+void grpc_security_create_handshakers(grpc_exec_ctx *exec_ctx,
+                                      tsi_handshaker *handshaker,
+                                      grpc_security_connector *connector,
+                                      grpc_handshake_manager *handshake_mgr) {
+  // If no TSI handshaker was created, add a handshaker that always fails.
+  // Otherwise, add a real security handshaker.
+  if (handshaker == NULL) {
+    grpc_handshake_manager_add(handshake_mgr, fail_handshaker_create());
+  } else {
+    grpc_handshake_manager_add(
+        handshake_mgr,
+        security_handshaker_create(exec_ctx, handshaker, connector));
+  }
+}
diff --git a/src/core/lib/security/transport/handshake.h b/src/core/lib/security/transport/security_handshaker.h
index f894540515..f71f43a359 100644
--- a/src/core/lib/security/transport/handshake.h
+++ b/src/core/lib/security/transport/security_handshaker.h
@@ -31,20 +31,17 @@
  *
  */
 
-#ifndef GRPC_CORE_LIB_SECURITY_TRANSPORT_HANDSHAKE_H
-#define GRPC_CORE_LIB_SECURITY_TRANSPORT_HANDSHAKE_H
+#ifndef GRPC_CORE_LIB_SECURITY_TRANSPORT_SECURITY_HANDSHAKER_H
+#define GRPC_CORE_LIB_SECURITY_TRANSPORT_SECURITY_HANDSHAKER_H
 
 #include "src/core/lib/iomgr/endpoint.h"
 #include "src/core/lib/security/transport/security_connector.h"
 
-/* Calls the callback upon completion. Takes owership of handshaker and
- * read_buffer. */
-void grpc_do_security_handshake(
-    grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker,
-    grpc_security_connector *connector, bool is_client_side,
-    grpc_endpoint *nonsecure_endpoint, grpc_slice_buffer *read_buffer,
-    gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data);
+/// Creates any necessary security handshakers and adds them to
+/// \a handshake_mgr.
+void grpc_security_create_handshakers(grpc_exec_ctx *exec_ctx,
+                                      tsi_handshaker *handshaker,
+                                      grpc_security_connector *connector,
+                                      grpc_handshake_manager *handshake_mgr);
 
-void grpc_security_handshake_shutdown(grpc_exec_ctx *exec_ctx, void *handshake);
-
-#endif /* GRPC_CORE_LIB_SECURITY_TRANSPORT_HANDSHAKE_H */
+#endif /* GRPC_CORE_LIB_SECURITY_TRANSPORT_SECURITY_HANDSHAKER_H */
diff --git a/src/core/lib/support/backoff.c b/src/core/lib/support/backoff.c
index e89ef47220..0612472712 100644
--- a/src/core/lib/support/backoff.c
+++ b/src/core/lib/support/backoff.c
@@ -35,8 +35,10 @@
 
 #include <grpc/support/useful.h>
 
-void gpr_backoff_init(gpr_backoff *backoff, double multiplier, double jitter,
+void gpr_backoff_init(gpr_backoff *backoff, int64_t initial_connect_timeout,
+                      double multiplier, double jitter,
                       int64_t min_timeout_millis, int64_t max_timeout_millis) {
+  backoff->initial_connect_timeout = initial_connect_timeout;
   backoff->multiplier = multiplier;
   backoff->jitter = jitter;
   backoff->min_timeout_millis = min_timeout_millis;
@@ -45,9 +47,10 @@ void gpr_backoff_init(gpr_backoff *backoff, double multiplier, double jitter,
 }
 
 gpr_timespec gpr_backoff_begin(gpr_backoff *backoff, gpr_timespec now) {
-  backoff->current_timeout_millis = backoff->min_timeout_millis;
-  return gpr_time_add(
-      now, gpr_time_from_millis(backoff->current_timeout_millis, GPR_TIMESPAN));
+  backoff->current_timeout_millis = backoff->initial_connect_timeout;
+  const int64_t first_timeout =
+      GPR_MAX(backoff->current_timeout_millis, backoff->min_timeout_millis);
+  return gpr_time_add(now, gpr_time_from_millis(first_timeout, GPR_TIMESPAN));
 }
 
 /* Generate a random number between 0 and 1. */
@@ -57,20 +60,28 @@ static double generate_uniform_random_number(uint32_t *rng_state) {
 }
 
 gpr_timespec gpr_backoff_step(gpr_backoff *backoff, gpr_timespec now) {
-  double new_timeout_millis =
+  const double new_timeout_millis =
       backoff->multiplier * (double)backoff->current_timeout_millis;
-  double jitter_range = backoff->jitter * new_timeout_millis;
-  double jitter =
+  backoff->current_timeout_millis =
+      GPR_MIN((int64_t)new_timeout_millis, backoff->max_timeout_millis);
+
+  const double jitter_range_width = backoff->jitter * new_timeout_millis;
+  const double jitter =
       (2 * generate_uniform_random_number(&backoff->rng_state) - 1) *
-      jitter_range;
+      jitter_range_width;
+
   backoff->current_timeout_millis =
-      GPR_CLAMP((int64_t)(new_timeout_millis + jitter),
-                backoff->min_timeout_millis, backoff->max_timeout_millis);
-  return gpr_time_add(
+      (int64_t)((double)(backoff->current_timeout_millis) + jitter);
+
+  const gpr_timespec current_deadline = gpr_time_add(
       now, gpr_time_from_millis(backoff->current_timeout_millis, GPR_TIMESPAN));
+
+  const gpr_timespec min_deadline = gpr_time_add(
+      now, gpr_time_from_millis(backoff->min_timeout_millis, GPR_TIMESPAN));
+
+  return gpr_time_max(current_deadline, min_deadline);
 }
 
 void gpr_backoff_reset(gpr_backoff *backoff) {
-  // forces step() to return a timeout of min_timeout_millis
-  backoff->current_timeout_millis = 0;
+  backoff->current_timeout_millis = backoff->initial_connect_timeout;
 }
diff --git a/src/core/lib/support/backoff.h b/src/core/lib/support/backoff.h
index 6d40c15546..5e9b740824 100644
--- a/src/core/lib/support/backoff.h
+++ b/src/core/lib/support/backoff.h
@@ -37,7 +37,9 @@
 #include <grpc/support/time.h>
 
 typedef struct {
-  /// const: multiplier between retry attempts
+  /// const:  how long to wait after the first failure before retrying
+  int64_t initial_connect_timeout;
+  /// const: factor with which to multiply backoff after a failed retry
   double multiplier;
   /// const: amount to randomize backoffs
   double jitter;
@@ -54,7 +56,8 @@ typedef struct {
 } gpr_backoff;
 
 /// Initialize backoff machinery - does not need to be destroyed
-void gpr_backoff_init(gpr_backoff *backoff, double multiplier, double jitter,
+void gpr_backoff_init(gpr_backoff *backoff, int64_t initial_connect_timeout,
+                      double multiplier, double jitter,
                       int64_t min_timeout_millis, int64_t max_timeout_millis);
 
 /// Begin retry loop: returns a timespec for the NEXT retry
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index 96cf706778..acfa03384a 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -1490,6 +1490,10 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
           error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
           goto done_with_error;
         }
+        /* IF this is a server, then GRPC_OP_RECV_INITIAL_METADATA *must* come
+           from server.c. In that case, it's coming from accept_stream, and in
+           that case we're not necessarily covered by a poller. */
+        stream_op->covered_by_poller = call->is_client;
         call->received_initial_metadata = 1;
         call->buffered_metadata[0] = op->data.recv_initial_metadata;
         grpc_closure_init(&call->receiving_initial_metadata_ready,