Merge pull request #16773 from yihuazhang/alts_tsi_handshaker_use_pollset_set

Use gRPC thread model (i.e., pollset_set) in ALTS TSI implementation

13 files changed, 740 insertions, 523 deletions

diff --git a/src/core/lib/security/transport/security_handshaker.cc b/src/core/lib/security/transport/security_handshaker.cc
index 4d6b133809..854a1c4af9 100644
--- a/src/core/lib/security/transport/security_handshaker.cc
+++ b/src/core/lib/security/transport/security_handshaker.cc
@@ -275,9 +275,6 @@ static void on_handshake_next_done_grpc_wrapper(
     tsi_result result, void* user_data, const unsigned char* bytes_to_send,
     size_t bytes_to_send_size, tsi_handshaker_result* handshaker_result) {
   security_handshaker* h = static_cast<security_handshaker*>(user_data);
-  // This callback will be invoked by TSI in a non-grpc thread, so it's
-  // safe to create our own exec_ctx here.
-  grpc_core::ExecCtx exec_ctx;
   gpr_mu_lock(&h->mu);
   grpc_error* error = on_handshake_next_done_locked(
       h, result, bytes_to_send, bytes_to_send_size, handshaker_result);
diff --git a/src/core/tsi/alts/handshaker/alts_handshaker_client.cc b/src/core/tsi/alts/handshaker/alts_handshaker_client.cc
index 17e8026096..941ca13114 100644
--- a/src/core/tsi/alts/handshaker/alts_handshaker_client.cc
+++ b/src/core/tsi/alts/handshaker/alts_handshaker_client.cc
@@ -25,30 +25,165 @@
 #include <grpc/support/log.h>
 
 #include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/surface/call.h"
+#include "src/core/lib/surface/channel.h"
 #include "src/core/tsi/alts/handshaker/alts_handshaker_service_api.h"
+#include "src/core/tsi/alts/handshaker/alts_shared_resource.h"
+#include "src/core/tsi/alts/handshaker/alts_tsi_handshaker_private.h"
+#include "src/core/tsi/alts/handshaker/alts_tsi_utils.h"
+
+#define TSI_ALTS_INITIAL_BUFFER_SIZE 256
 
 const int kHandshakerClientOpNum = 4;
 
+struct alts_handshaker_client {
+  const alts_handshaker_client_vtable* vtable;
+};
+
 typedef struct alts_grpc_handshaker_client {
   alts_handshaker_client base;
+  alts_tsi_handshaker* handshaker;
   grpc_call* call;
+  /* A pointer to a function handling the interaction with handshaker service.
+   * That is, it points to grpc_call_start_batch_and_execute when the handshaker
+   * client is used in a non-testing use case and points to a custom function
+   * that validates the data to be sent to handshaker service in a testing use
+   * case. */
   alts_grpc_caller grpc_caller;
+  /* A callback function provided by gRPC to handle the response returned from
+   * handshaker service. It also serves to bring the control safely back to
+   * application when dedicated CQ and thread are used. */
+  grpc_iomgr_cb_func grpc_cb;
+  /* A gRPC closure to be scheduled when the response from handshaker service
+   * is received. It will be initialized with grpc_cb. */
+  grpc_closure on_handshaker_service_resp_recv;
+  /* Buffers containing information to be sent (or received) to (or from) the
+   * handshaker service. */
+  grpc_byte_buffer* send_buffer;
+  grpc_byte_buffer* recv_buffer;
+  grpc_status_code status;
+  /* Initial metadata to be received from handshaker service. */
+  grpc_metadata_array recv_initial_metadata;
+  /* A callback function provided by an application to be invoked when response
+   * is received from handshaker service. */
+  tsi_handshaker_on_next_done_cb cb;
+  void* user_data;
+  /* ALTS credential options passed in from the caller. */
+  grpc_alts_credentials_options* options;
+  /* target name information to be passed to handshaker service for server
+   * authorization check. */
+  grpc_slice target_name;
+  /* boolean flag indicating if the handshaker client is used at client
+   * (is_client = true) or server (is_client = false) side. */
+  bool is_client;
+  /* a temporary store for data received from handshaker service used to extract
+   * unused data. */
+  grpc_slice recv_bytes;
+  /* a buffer containing data to be sent to the grpc client or server's peer. */
+  unsigned char* buffer;
+  size_t buffer_size;
 } alts_grpc_handshaker_client;
 
-static grpc_call_error grpc_start_batch(grpc_call* call, const grpc_op* ops,
-                                        size_t nops, void* tag) {
-  return grpc_call_start_batch(call, ops, nops, tag, nullptr);
+static void handshaker_client_send_buffer_destroy(
+    alts_grpc_handshaker_client* client) {
+  GPR_ASSERT(client != nullptr);
+  grpc_byte_buffer_destroy(client->send_buffer);
+  client->send_buffer = nullptr;
+}
+
+static bool is_handshake_finished_properly(grpc_gcp_handshaker_resp* resp) {
+  GPR_ASSERT(resp != nullptr);
+  if (resp->has_result) {
+    return true;
+  }
+  return false;
+}
+
+void alts_handshaker_client_handle_response(alts_handshaker_client* c,
+                                            bool is_ok) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  grpc_byte_buffer* recv_buffer = client->recv_buffer;
+  grpc_status_code status = client->status;
+  tsi_handshaker_on_next_done_cb cb = client->cb;
+  void* user_data = client->user_data;
+  alts_tsi_handshaker* handshaker = client->handshaker;
+
+  /* Invalid input check. */
+  if (cb == nullptr) {
+    gpr_log(GPR_ERROR,
+            "cb is nullptr in alts_tsi_handshaker_handle_response()");
+    return;
+  }
+  if (handshaker == nullptr || recv_buffer == nullptr) {
+    gpr_log(GPR_ERROR,
+            "Invalid arguments to alts_tsi_handshaker_handle_response()");
+    cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr);
+    return;
+  }
+  if (alts_tsi_handshaker_has_shutdown(handshaker)) {
+    gpr_log(GPR_ERROR, "TSI handshake shutdown");
+    cb(TSI_HANDSHAKE_SHUTDOWN, user_data, nullptr, 0, nullptr);
+    return;
+  }
+  /* Failed grpc call check. */
+  if (!is_ok || status != GRPC_STATUS_OK) {
+    gpr_log(GPR_ERROR, "grpc call made to handshaker service failed");
+    cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr);
+    return;
+  }
+  grpc_gcp_handshaker_resp* resp =
+      alts_tsi_utils_deserialize_response(recv_buffer);
+  grpc_byte_buffer_destroy(client->recv_buffer);
+  client->recv_buffer = nullptr;
+  /* Invalid handshaker response check. */
+  if (resp == nullptr) {
+    gpr_log(GPR_ERROR, "alts_tsi_utils_deserialize_response() failed");
+    cb(TSI_DATA_CORRUPTED, user_data, nullptr, 0, nullptr);
+    return;
+  }
+  grpc_slice* slice = static_cast<grpc_slice*>(resp->out_frames.arg);
+  unsigned char* bytes_to_send = nullptr;
+  size_t bytes_to_send_size = 0;
+  if (slice != nullptr) {
+    bytes_to_send_size = GRPC_SLICE_LENGTH(*slice);
+    while (bytes_to_send_size > client->buffer_size) {
+      client->buffer_size *= 2;
+      client->buffer = static_cast<unsigned char*>(
+          gpr_realloc(client->buffer, client->buffer_size));
+    }
+    memcpy(client->buffer, GRPC_SLICE_START_PTR(*slice), bytes_to_send_size);
+    bytes_to_send = client->buffer;
+  }
+  tsi_handshaker_result* result = nullptr;
+  if (is_handshake_finished_properly(resp)) {
+    alts_tsi_handshaker_result_create(resp, client->is_client, &result);
+    alts_tsi_handshaker_result_set_unused_bytes(result, &client->recv_bytes,
+                                                resp->bytes_consumed);
+  }
+  grpc_status_code code = static_cast<grpc_status_code>(resp->status.code);
+  if (code != GRPC_STATUS_OK) {
+    grpc_slice* details = static_cast<grpc_slice*>(resp->status.details.arg);
+    if (details != nullptr) {
+      char* error_details = grpc_slice_to_c_string(*details);
+      gpr_log(GPR_ERROR, "Error from handshaker service:%s", error_details);
+      gpr_free(error_details);
+    }
+  }
+  grpc_gcp_handshaker_resp_destroy(resp);
+  cb(alts_tsi_utils_convert_to_tsi_result(code), user_data, bytes_to_send,
+     bytes_to_send_size, result);
 }
 
 /**
- * Populate grpc operation data with the fields of ALTS TSI event and make a
- * grpc call.
+ * Populate grpc operation data with the fields of ALTS handshaker client and
+ * make a grpc call.
  */
-static tsi_result make_grpc_call(alts_handshaker_client* client,
-                                 alts_tsi_event* event, bool is_start) {
-  GPR_ASSERT(client != nullptr && event != nullptr);
-  alts_grpc_handshaker_client* grpc_client =
-      reinterpret_cast<alts_grpc_handshaker_client*>(client);
+static tsi_result make_grpc_call(alts_handshaker_client* c, bool is_start) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
   grpc_op ops[kHandshakerClientOpNum];
   memset(ops, 0, sizeof(ops));
   grpc_op* op = ops;
@@ -59,22 +194,22 @@ static tsi_result make_grpc_call(alts_handshaker_client* client,
     GPR_ASSERT(op - ops <= kHandshakerClientOpNum);
     op->op = GRPC_OP_RECV_INITIAL_METADATA;
     op->data.recv_initial_metadata.recv_initial_metadata =
-        &event->initial_metadata;
+        &client->recv_initial_metadata;
     op++;
     GPR_ASSERT(op - ops <= kHandshakerClientOpNum);
   }
   op->op = GRPC_OP_SEND_MESSAGE;
-  op->data.send_message.send_message = event->send_buffer;
+  op->data.send_message.send_message = client->send_buffer;
   op++;
   GPR_ASSERT(op - ops <= kHandshakerClientOpNum);
   op->op = GRPC_OP_RECV_MESSAGE;
-  op->data.recv_message.recv_message = &event->recv_buffer;
+  op->data.recv_message.recv_message = &client->recv_buffer;
   op++;
   GPR_ASSERT(op - ops <= kHandshakerClientOpNum);
-  GPR_ASSERT(grpc_client->grpc_caller != nullptr);
-  if (grpc_client->grpc_caller(grpc_client->call, ops,
-                               static_cast<size_t>(op - ops),
-                               (void*)event) != GRPC_CALL_OK) {
+  GPR_ASSERT(client->grpc_caller != nullptr);
+  if (client->grpc_caller(client->call, ops, static_cast<size_t>(op - ops),
+                          &client->on_handshaker_service_resp_recv) !=
+      GRPC_CALL_OK) {
     gpr_log(GPR_ERROR, "Start batch operation failed");
     return TSI_INTERNAL_ERROR;
   }
@@ -82,7 +217,11 @@ static tsi_result make_grpc_call(alts_handshaker_client* client,
 }
 
 /* Create and populate a client_start handshaker request, then serialize it. */
-static grpc_byte_buffer* get_serialized_start_client(alts_tsi_event* event) {
+static grpc_byte_buffer* get_serialized_start_client(
+    alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
   bool ok = true;
   grpc_gcp_handshaker_req* req =
       grpc_gcp_handshaker_req_create(CLIENT_START_REQ);
@@ -91,14 +230,14 @@ static grpc_byte_buffer* get_serialized_start_client(alts_tsi_event* event) {
   ok &= grpc_gcp_handshaker_req_add_application_protocol(
       req, ALTS_APPLICATION_PROTOCOL);
   ok &= grpc_gcp_handshaker_req_add_record_protocol(req, ALTS_RECORD_PROTOCOL);
-  grpc_gcp_rpc_protocol_versions* versions = &event->options->rpc_versions;
+  grpc_gcp_rpc_protocol_versions* versions = &client->options->rpc_versions;
   ok &= grpc_gcp_handshaker_req_set_rpc_versions(
       req, versions->max_rpc_version.major, versions->max_rpc_version.minor,
       versions->min_rpc_version.major, versions->min_rpc_version.minor);
-  char* target_name = grpc_slice_to_c_string(event->target_name);
+  char* target_name = grpc_slice_to_c_string(client->target_name);
   ok &= grpc_gcp_handshaker_req_set_target_name(req, target_name);
   target_service_account* ptr =
-      (reinterpret_cast<grpc_alts_credentials_client_options*>(event->options))
+      (reinterpret_cast<grpc_alts_credentials_client_options*>(client->options))
           ->target_account_list_head;
   while (ptr != nullptr) {
     grpc_gcp_handshaker_req_add_target_identity_service_account(req, ptr->data);
@@ -116,19 +255,21 @@ static grpc_byte_buffer* get_serialized_start_client(alts_tsi_event* event) {
   return buffer;
 }
 
-static tsi_result handshaker_client_start_client(alts_handshaker_client* client,
-                                                 alts_tsi_event* event) {
-  if (client == nullptr || event == nullptr) {
-    gpr_log(GPR_ERROR, "Invalid arguments to handshaker_client_start_client()");
+static tsi_result handshaker_client_start_client(alts_handshaker_client* c) {
+  if (c == nullptr) {
+    gpr_log(GPR_ERROR, "client is nullptr in handshaker_client_start_client()");
     return TSI_INVALID_ARGUMENT;
   }
-  grpc_byte_buffer* buffer = get_serialized_start_client(event);
+  grpc_byte_buffer* buffer = get_serialized_start_client(c);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
   if (buffer == nullptr) {
     gpr_log(GPR_ERROR, "get_serialized_start_client() failed");
     return TSI_INTERNAL_ERROR;
   }
-  event->send_buffer = buffer;
-  tsi_result result = make_grpc_call(client, event, true /* is_start */);
+  handshaker_client_send_buffer_destroy(client);
+  client->send_buffer = buffer;
+  tsi_result result = make_grpc_call(&client->base, true /* is_start */);
   if (result != TSI_OK) {
     gpr_log(GPR_ERROR, "make_grpc_call() failed");
   }
@@ -137,8 +278,11 @@ static tsi_result handshaker_client_start_client(alts_handshaker_client* client,
 
 /* Create and populate a start_server handshaker request, then serialize it. */
 static grpc_byte_buffer* get_serialized_start_server(
-    alts_tsi_event* event, grpc_slice* bytes_received) {
+    alts_handshaker_client* c, grpc_slice* bytes_received) {
+  GPR_ASSERT(c != nullptr);
   GPR_ASSERT(bytes_received != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
   grpc_gcp_handshaker_req* req =
       grpc_gcp_handshaker_req_create(SERVER_START_REQ);
   bool ok = grpc_gcp_handshaker_req_add_application_protocol(
@@ -148,7 +292,7 @@ static grpc_byte_buffer* get_serialized_start_server(
   ok &= grpc_gcp_handshaker_req_set_in_bytes(
       req, reinterpret_cast<const char*> GRPC_SLICE_START_PTR(*bytes_received),
       GRPC_SLICE_LENGTH(*bytes_received));
-  grpc_gcp_rpc_protocol_versions* versions = &event->options->rpc_versions;
+  grpc_gcp_rpc_protocol_versions* versions = &client->options->rpc_versions;
   ok &= grpc_gcp_handshaker_req_set_rpc_versions(
       req, versions->max_rpc_version.major, versions->max_rpc_version.minor,
       versions->min_rpc_version.major, versions->min_rpc_version.minor);
@@ -163,20 +307,22 @@ static grpc_byte_buffer* get_serialized_start_server(
   return buffer;
 }
 
-static tsi_result handshaker_client_start_server(alts_handshaker_client* client,
-                                                 alts_tsi_event* event,
+static tsi_result handshaker_client_start_server(alts_handshaker_client* c,
                                                  grpc_slice* bytes_received) {
-  if (client == nullptr || event == nullptr || bytes_received == nullptr) {
+  if (c == nullptr || bytes_received == nullptr) {
     gpr_log(GPR_ERROR, "Invalid arguments to handshaker_client_start_server()");
     return TSI_INVALID_ARGUMENT;
   }
-  grpc_byte_buffer* buffer = get_serialized_start_server(event, bytes_received);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  grpc_byte_buffer* buffer = get_serialized_start_server(c, bytes_received);
   if (buffer == nullptr) {
     gpr_log(GPR_ERROR, "get_serialized_start_server() failed");
     return TSI_INTERNAL_ERROR;
   }
-  event->send_buffer = buffer;
-  tsi_result result = make_grpc_call(client, event, true /* is_start */);
+  handshaker_client_send_buffer_destroy(client);
+  client->send_buffer = buffer;
+  tsi_result result = make_grpc_call(&client->base, true /* is_start */);
   if (result != TSI_OK) {
     gpr_log(GPR_ERROR, "make_grpc_call() failed");
   }
@@ -201,40 +347,48 @@ static grpc_byte_buffer* get_serialized_next(grpc_slice* bytes_received) {
   return buffer;
 }
 
-static tsi_result handshaker_client_next(alts_handshaker_client* client,
-                                         alts_tsi_event* event,
+static tsi_result handshaker_client_next(alts_handshaker_client* c,
                                          grpc_slice* bytes_received) {
-  if (client == nullptr || event == nullptr || bytes_received == nullptr) {
+  if (c == nullptr || bytes_received == nullptr) {
     gpr_log(GPR_ERROR, "Invalid arguments to handshaker_client_next()");
     return TSI_INVALID_ARGUMENT;
   }
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  grpc_slice_unref_internal(client->recv_bytes);
+  client->recv_bytes = grpc_slice_ref(*bytes_received);
   grpc_byte_buffer* buffer = get_serialized_next(bytes_received);
   if (buffer == nullptr) {
     gpr_log(GPR_ERROR, "get_serialized_next() failed");
     return TSI_INTERNAL_ERROR;
   }
-  event->send_buffer = buffer;
-  tsi_result result = make_grpc_call(client, event, false /* is_start */);
+  handshaker_client_send_buffer_destroy(client);
+  client->send_buffer = buffer;
+  tsi_result result = make_grpc_call(&client->base, false /* is_start */);
   if (result != TSI_OK) {
     gpr_log(GPR_ERROR, "make_grpc_call() failed");
   }
   return result;
 }
 
-static void handshaker_client_shutdown(alts_handshaker_client* client) {
-  GPR_ASSERT(client != nullptr);
-  alts_grpc_handshaker_client* grpc_client =
-      reinterpret_cast<alts_grpc_handshaker_client*>(client);
-  GPR_ASSERT(grpc_call_cancel(grpc_client->call, nullptr) == GRPC_CALL_OK);
+static void handshaker_client_shutdown(alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  if (client->call != nullptr) {
+    GPR_ASSERT(grpc_call_cancel(client->call, nullptr) == GRPC_CALL_OK);
+  }
 }
 
-static void handshaker_client_destruct(alts_handshaker_client* client) {
-  if (client == nullptr) {
+static void handshaker_client_destruct(alts_handshaker_client* c) {
+  if (c == nullptr) {
     return;
   }
-  alts_grpc_handshaker_client* grpc_client =
-      reinterpret_cast<alts_grpc_handshaker_client*>(client);
-  grpc_call_unref(grpc_client->call);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  if (client->call != nullptr) {
+    grpc_call_unref(client->call);
+  }
 }
 
 static const alts_handshaker_client_vtable vtable = {
@@ -243,22 +397,45 @@ static const alts_handshaker_client_vtable vtable = {
     handshaker_client_destruct};
 
 alts_handshaker_client* alts_grpc_handshaker_client_create(
-    grpc_channel* channel, grpc_completion_queue* queue,
-    const char* handshaker_service_url) {
-  if (channel == nullptr || queue == nullptr ||
-      handshaker_service_url == nullptr) {
+    alts_tsi_handshaker* handshaker, grpc_channel* channel,
+    const char* handshaker_service_url, grpc_pollset_set* interested_parties,
+    grpc_alts_credentials_options* options, grpc_slice target_name,
+    grpc_iomgr_cb_func grpc_cb, tsi_handshaker_on_next_done_cb cb,
+    void* user_data, alts_handshaker_client_vtable* vtable_for_testing,
+    bool is_client) {
+  if (channel == nullptr || handshaker_service_url == nullptr) {
     gpr_log(GPR_ERROR, "Invalid arguments to alts_handshaker_client_create()");
     return nullptr;
   }
   alts_grpc_handshaker_client* client =
       static_cast<alts_grpc_handshaker_client*>(gpr_zalloc(sizeof(*client)));
-  client->grpc_caller = grpc_start_batch;
+  client->grpc_caller = grpc_call_start_batch_and_execute;
+  client->handshaker = handshaker;
+  client->cb = cb;
+  client->user_data = user_data;
+  client->send_buffer = nullptr;
+  client->recv_buffer = nullptr;
+  client->options = grpc_alts_credentials_options_copy(options);
+  client->target_name = grpc_slice_copy(target_name);
+  client->recv_bytes = grpc_empty_slice();
+  grpc_metadata_array_init(&client->recv_initial_metadata);
+  client->grpc_cb = grpc_cb;
+  client->is_client = is_client;
+  client->buffer_size = TSI_ALTS_INITIAL_BUFFER_SIZE;
+  client->buffer = static_cast<unsigned char*>(gpr_zalloc(client->buffer_size));
   grpc_slice slice = grpc_slice_from_copied_string(handshaker_service_url);
-  client->call = grpc_channel_create_call(
-      channel, nullptr, GRPC_PROPAGATE_DEFAULTS, queue,
-      grpc_slice_from_static_string(ALTS_SERVICE_METHOD), &slice,
-      gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
-  client->base.vtable = &vtable;
+  client->call =
+      strcmp(handshaker_service_url, ALTS_HANDSHAKER_SERVICE_URL_FOR_TESTING) ==
+              0
+          ? nullptr
+          : grpc_channel_create_pollset_set_call(
+                channel, nullptr, GRPC_PROPAGATE_DEFAULTS, interested_parties,
+                grpc_slice_from_static_string(ALTS_SERVICE_METHOD), &slice,
+                GRPC_MILLIS_INF_FUTURE, nullptr);
+  client->base.vtable =
+      vtable_for_testing == nullptr ? &vtable : vtable_for_testing;
+  GRPC_CLOSURE_INIT(&client->on_handshaker_service_resp_recv, client->grpc_cb,
+                    client, grpc_schedule_on_exec_ctx);
   grpc_slice_unref_internal(slice);
   return &client->base;
 }
@@ -267,21 +444,114 @@ namespace grpc_core {
 namespace internal {
 
 void alts_handshaker_client_set_grpc_caller_for_testing(
-    alts_handshaker_client* client, alts_grpc_caller caller) {
-  GPR_ASSERT(client != nullptr && caller != nullptr);
-  alts_grpc_handshaker_client* grpc_client =
-      reinterpret_cast<alts_grpc_handshaker_client*>(client);
-  grpc_client->grpc_caller = caller;
+    alts_handshaker_client* c, alts_grpc_caller caller) {
+  GPR_ASSERT(c != nullptr && caller != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  client->grpc_caller = caller;
+}
+
+grpc_byte_buffer* alts_handshaker_client_get_send_buffer_for_testing(
+    alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  return client->send_buffer;
+}
+
+grpc_byte_buffer** alts_handshaker_client_get_recv_buffer_addr_for_testing(
+    alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  return &client->recv_buffer;
+}
+
+grpc_metadata_array* alts_handshaker_client_get_initial_metadata_for_testing(
+    alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  return &client->recv_initial_metadata;
+}
+
+void alts_handshaker_client_set_recv_bytes_for_testing(
+    alts_handshaker_client* c, grpc_slice* recv_bytes) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  client->recv_bytes = grpc_slice_ref(*recv_bytes);
+}
+
+void alts_handshaker_client_set_fields_for_testing(
+    alts_handshaker_client* c, alts_tsi_handshaker* handshaker,
+    tsi_handshaker_on_next_done_cb cb, void* user_data,
+    grpc_byte_buffer* recv_buffer, grpc_status_code status) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  client->handshaker = handshaker;
+  client->cb = cb;
+  client->user_data = user_data;
+  client->recv_buffer = recv_buffer;
+  client->status = status;
+}
+
+void alts_handshaker_client_check_fields_for_testing(
+    alts_handshaker_client* c, tsi_handshaker_on_next_done_cb cb,
+    void* user_data, bool has_sent_start_message, grpc_slice* recv_bytes) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  GPR_ASSERT(client->cb == cb);
+  GPR_ASSERT(client->user_data == user_data);
+  if (recv_bytes != nullptr) {
+    GPR_ASSERT(grpc_slice_cmp(client->recv_bytes, *recv_bytes) == 0);
+  }
+  GPR_ASSERT(alts_tsi_handshaker_get_has_sent_start_message_for_testing(
+                 client->handshaker) == has_sent_start_message);
+}
+
+void alts_handshaker_client_set_vtable_for_testing(
+    alts_handshaker_client* c, alts_handshaker_client_vtable* vtable) {
+  GPR_ASSERT(c != nullptr);
+  GPR_ASSERT(vtable != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  client->base.vtable = vtable;
+}
+
+alts_tsi_handshaker* alts_handshaker_client_get_handshaker_for_testing(
+    alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  return client->handshaker;
+}
+
+void alts_handshaker_client_set_cb_for_testing(
+    alts_handshaker_client* c, tsi_handshaker_on_next_done_cb cb) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  client->cb = cb;
+}
+
+grpc_closure* alts_handshaker_client_get_closure_for_testing(
+    alts_handshaker_client* c) {
+  GPR_ASSERT(c != nullptr);
+  alts_grpc_handshaker_client* client =
+      reinterpret_cast<alts_grpc_handshaker_client*>(c);
+  return &client->on_handshaker_service_resp_recv;
 }
 
 }  // namespace internal
 }  // namespace grpc_core
 
-tsi_result alts_handshaker_client_start_client(alts_handshaker_client* client,
-                                               alts_tsi_event* event) {
+tsi_result alts_handshaker_client_start_client(alts_handshaker_client* client) {
   if (client != nullptr && client->vtable != nullptr &&
       client->vtable->client_start != nullptr) {
-    return client->vtable->client_start(client, event);
+    return client->vtable->client_start(client);
   }
   gpr_log(GPR_ERROR,
           "client or client->vtable has not been initialized properly");
@@ -289,11 +559,10 @@ tsi_result alts_handshaker_client_start_client(alts_handshaker_client* client,
 }
 
 tsi_result alts_handshaker_client_start_server(alts_handshaker_client* client,
-                                               alts_tsi_event* event,
                                                grpc_slice* bytes_received) {
   if (client != nullptr && client->vtable != nullptr &&
       client->vtable->server_start != nullptr) {
-    return client->vtable->server_start(client, event, bytes_received);
+    return client->vtable->server_start(client, bytes_received);
   }
   gpr_log(GPR_ERROR,
           "client or client->vtable has not been initialized properly");
@@ -301,11 +570,10 @@ tsi_result alts_handshaker_client_start_server(alts_handshaker_client* client,
 }
 
 tsi_result alts_handshaker_client_next(alts_handshaker_client* client,
-                                       alts_tsi_event* event,
                                        grpc_slice* bytes_received) {
   if (client != nullptr && client->vtable != nullptr &&
       client->vtable->next != nullptr) {
-    return client->vtable->next(client, event, bytes_received);
+    return client->vtable->next(client, bytes_received);
   }
   gpr_log(GPR_ERROR,
           "client or client->vtable has not been initialized properly");
@@ -319,11 +587,22 @@ void alts_handshaker_client_shutdown(alts_handshaker_client* client) {
   }
 }
 
-void alts_handshaker_client_destroy(alts_handshaker_client* client) {
-  if (client != nullptr) {
-    if (client->vtable != nullptr && client->vtable->destruct != nullptr) {
-      client->vtable->destruct(client);
+void alts_handshaker_client_destroy(alts_handshaker_client* c) {
+  if (c != nullptr) {
+    if (c->vtable != nullptr && c->vtable->destruct != nullptr) {
+      c->vtable->destruct(c);
     }
+    alts_grpc_handshaker_client* client =
+        reinterpret_cast<alts_grpc_handshaker_client*>(c);
+    grpc_byte_buffer_destroy(client->send_buffer);
+    grpc_byte_buffer_destroy(client->recv_buffer);
+    client->send_buffer = nullptr;
+    client->recv_buffer = nullptr;
+    grpc_metadata_array_destroy(&client->recv_initial_metadata);
+    grpc_slice_unref_internal(client->recv_bytes);
+    grpc_slice_unref_internal(client->target_name);
+    grpc_alts_credentials_options_destroy(client->options);
+    gpr_free(client->buffer);
     gpr_free(client);
   }
 }
diff --git a/src/core/tsi/alts/handshaker/alts_handshaker_client.h b/src/core/tsi/alts/handshaker/alts_handshaker_client.h
index 8dd8fe440d..4b489875f3 100644
--- a/src/core/tsi/alts/handshaker/alts_handshaker_client.h
+++ b/src/core/tsi/alts/handshaker/alts_handshaker_client.h
@@ -21,16 +21,24 @@
 
 #include <grpc/support/port_platform.h>
 
+#include <grpc/byte_buffer.h>
+#include <grpc/byte_buffer_reader.h>
 #include <grpc/grpc.h>
 
-#include "src/core/tsi/alts/handshaker/alts_tsi_event.h"
+#include "src/core/tsi/alts/handshaker/alts_tsi_handshaker.h"
+#include "src/core/tsi/transport_security_interface.h"
+
+#include "src/core/lib/iomgr/closure.h"
+#include "src/core/lib/iomgr/pollset_set.h"
 
 #define ALTS_SERVICE_METHOD "/grpc.gcp.HandshakerService/DoHandshake"
 #define ALTS_APPLICATION_PROTOCOL "grpc"
 #define ALTS_RECORD_PROTOCOL "ALTSRP_GCM_AES128_REKEY"
+#define ALTS_HANDSHAKER_SERVICE_URL_FOR_TESTING "lame"
 
 const size_t kAltsAes128GcmRekeyKeyLength = 44;
 
+typedef struct alts_tsi_handshaker alts_tsi_handshaker;
 /**
  * A ALTS handshaker client interface. It is used to communicate with
  * ALTS handshaker service by scheduling a handshaker request that could be one
@@ -41,63 +49,52 @@ typedef struct alts_handshaker_client alts_handshaker_client;
 
 /* A function that makes the grpc call to the handshaker service. */
 typedef grpc_call_error (*alts_grpc_caller)(grpc_call* call, const grpc_op* ops,
-                                            size_t nops, void* tag);
+                                            size_t nops, grpc_closure* tag);
 
 /* V-table for ALTS handshaker client operations. */
 typedef struct alts_handshaker_client_vtable {
-  tsi_result (*client_start)(alts_handshaker_client* client,
-                             alts_tsi_event* event);
+  tsi_result (*client_start)(alts_handshaker_client* client);
   tsi_result (*server_start)(alts_handshaker_client* client,
-                             alts_tsi_event* event, grpc_slice* bytes_received);
-  tsi_result (*next)(alts_handshaker_client* client, alts_tsi_event* event,
+                             grpc_slice* bytes_received);
+  tsi_result (*next)(alts_handshaker_client* client,
                      grpc_slice* bytes_received);
   void (*shutdown)(alts_handshaker_client* client);
   void (*destruct)(alts_handshaker_client* client);
 } alts_handshaker_client_vtable;
 
-struct alts_handshaker_client {
-  const alts_handshaker_client_vtable* vtable;
-};
-
 /**
  * This method schedules a client_start handshaker request to ALTS handshaker
  * service.
  *
  * - client: ALTS handshaker client instance.
- * - event: ALTS TSI event instance.
  *
  * It returns TSI_OK on success and an error status code on failure.
  */
-tsi_result alts_handshaker_client_start_client(alts_handshaker_client* client,
-                                               alts_tsi_event* event);
+tsi_result alts_handshaker_client_start_client(alts_handshaker_client* client);
 
 /**
  * This method schedules a server_start handshaker request to ALTS handshaker
  * service.
  *
  * - client: ALTS handshaker client instance.
- * - event: ALTS TSI event instance.
  * - bytes_received: bytes in out_frames returned from the peer's handshaker
  *   response.
  *
  * It returns TSI_OK on success and an error status code on failure.
  */
 tsi_result alts_handshaker_client_start_server(alts_handshaker_client* client,
-                                               alts_tsi_event* event,
                                                grpc_slice* bytes_received);
 
 /**
  * This method schedules a next handshaker request to ALTS handshaker service.
  *
  * - client: ALTS handshaker client instance.
- * - event: ALTS TSI event instance.
  * - bytes_received: bytes in out_frames returned from the peer's handshaker
  *   response.
  *
  * It returns TSI_OK on success and an error status code on failure.
  */
 tsi_result alts_handshaker_client_next(alts_handshaker_client* client,
-                                       alts_tsi_event* event,
                                        grpc_slice* bytes_received);
 
 /**
@@ -110,38 +107,51 @@ tsi_result alts_handshaker_client_next(alts_handshaker_client* client,
 void alts_handshaker_client_shutdown(alts_handshaker_client* client);
 
 /**
- * This method destroys a ALTS handshaker client.
+ * This method destroys an ALTS handshaker client.
  *
- * - client: a ALTS handshaker client instance.
+ * - client: an ALTS handshaker client instance.
  */
 void alts_handshaker_client_destroy(alts_handshaker_client* client);
 
 /**
- * This method creates a ALTS handshaker client.
+ * This method creates an ALTS handshaker client.
  *
+ * - handshaker: ALTS TSI handshaker to which the created handshaker client
+ * belongs to.
  * - channel: grpc channel to ALTS handshaker service.
- * - queue: grpc completion queue.
  * - handshaker_service_url: address of ALTS handshaker service in the format of
  *   "host:port".
- *
- * It returns the created ALTS handshaker client on success, and NULL on
- * failure.
+ * - interested_parties: set of pollsets interested in this connection.
+ * - options: ALTS credentials options containing information passed from TSI
+ *   caller (e.g., rpc protocol versions)
+ * - target_name: the name of the endpoint that the channel is connecting to,
+ *   and will be used for secure naming check
+ * - grpc_cb: gRPC provided callbacks passed from TSI handshaker.
+ * - cb: callback to be executed when tsi_handshaker_next API compltes.
+ * - user_data: argument passed to cb.
+ * - vtable_for_testing: ALTS handshaker client vtable instance used for
+ *   testing purpose.
+ * - is_client: a boolean value indicating if the created handshaker client is
+ * used at the client (is_client = true) or server (is_client = false) side. It
+ * returns the created ALTS handshaker client on success, and NULL on failure.
  */
 alts_handshaker_client* alts_grpc_handshaker_client_create(
-    grpc_channel* channel, grpc_completion_queue* queue,
-    const char* handshaker_service_url);
-
-namespace grpc_core {
-namespace internal {
+    alts_tsi_handshaker* handshaker, grpc_channel* channel,
+    const char* handshaker_service_url, grpc_pollset_set* interested_parties,
+    grpc_alts_credentials_options* options, grpc_slice target_name,
+    grpc_iomgr_cb_func grpc_cb, tsi_handshaker_on_next_done_cb cb,
+    void* user_data, alts_handshaker_client_vtable* vtable_for_testing,
+    bool is_client);
 
 /**
- * Unsafe, use for testing only. It allows the caller to change the way that
- * GRPC calls are made to the handshaker service.
+ * This method handles handshaker response returned from ALTS handshaker
+ * service. Note that the only reason the API is exposed is that it is used in
+ * alts_shared_resources.cc.
+ *
+ * - client: an ALTS handshaker client instance.
+ * - is_ok: a boolean value indicating if the handshaker response is ok to read.
  */
-void alts_handshaker_client_set_grpc_caller_for_testing(
-    alts_handshaker_client* client, alts_grpc_caller caller);
-
-}  // namespace internal
-}  // namespace grpc_core
+void alts_handshaker_client_handle_response(alts_handshaker_client* client,
+                                            bool is_ok);
 
 #endif /* GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_HANDSHAKER_CLIENT_H */
diff --git a/src/core/tsi/alts/handshaker/alts_shared_resource.cc b/src/core/tsi/alts/handshaker/alts_shared_resource.cc
new file mode 100644
index 0000000000..ffb5e1c655
--- /dev/null
+++ b/src/core/tsi/alts/handshaker/alts_shared_resource.cc
@@ -0,0 +1,73 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/tsi/alts/handshaker/alts_shared_resource.h"
+
+#include <grpc/support/log.h>
+
+#include "src/core/tsi/alts/handshaker/alts_handshaker_client.h"
+
+static alts_shared_resource_dedicated g_alts_resource_dedicated;
+static alts_shared_resource* g_shared_resources = alts_get_shared_resource();
+
+alts_shared_resource_dedicated* grpc_alts_get_shared_resource_dedicated(void) {
+  return &g_alts_resource_dedicated;
+}
+
+static void thread_worker(void* arg) {
+  while (true) {
+    grpc_event event =
+        grpc_completion_queue_next(g_alts_resource_dedicated.cq,
+                                   gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
+    GPR_ASSERT(event.type != GRPC_QUEUE_TIMEOUT);
+    if (event.type == GRPC_QUEUE_SHUTDOWN) {
+      break;
+    }
+    GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
+    alts_handshaker_client* client =
+        static_cast<alts_handshaker_client*>(event.tag);
+    alts_handshaker_client_handle_response(client, event.success);
+  }
+}
+
+void grpc_alts_shared_resource_dedicated_init() {
+  g_alts_resource_dedicated.cq = nullptr;
+}
+
+void grpc_alts_shared_resource_dedicated_start() {
+  g_alts_resource_dedicated.cq = grpc_completion_queue_create_for_next(nullptr);
+  g_alts_resource_dedicated.thread =
+      grpc_core::Thread("alts_tsi_handshaker", &thread_worker, nullptr);
+  g_alts_resource_dedicated.interested_parties = grpc_pollset_set_create();
+  grpc_pollset_set_add_pollset(g_alts_resource_dedicated.interested_parties,
+                               grpc_cq_pollset(g_alts_resource_dedicated.cq));
+  g_alts_resource_dedicated.thread.Start();
+}
+
+void grpc_alts_shared_resource_dedicated_shutdown() {
+  if (g_alts_resource_dedicated.cq != nullptr) {
+    grpc_pollset_set_del_pollset(g_alts_resource_dedicated.interested_parties,
+                                 grpc_cq_pollset(g_alts_resource_dedicated.cq));
+    grpc_completion_queue_shutdown(g_alts_resource_dedicated.cq);
+    g_alts_resource_dedicated.thread.Join();
+    grpc_pollset_set_destroy(g_alts_resource_dedicated.interested_parties);
+    grpc_completion_queue_destroy(g_alts_resource_dedicated.cq);
+  }
+}
diff --git a/src/core/tsi/alts/handshaker/alts_shared_resource.h b/src/core/tsi/alts/handshaker/alts_shared_resource.h
new file mode 100644
index 0000000000..a07da305a8
--- /dev/null
+++ b/src/core/tsi/alts/handshaker/alts_shared_resource.h
@@ -0,0 +1,70 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_SHARED_RESOURCE_H
+#define GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_SHARED_RESOURCE_H
+
+#include <grpc/support/port_platform.h>
+
+#include <grpc/grpc.h>
+#include <grpc/support/sync.h>
+
+#include "src/core/lib/gprpp/thd.h"
+#include "src/core/lib/iomgr/pollset_set.h"
+#include "src/core/lib/surface/completion_queue.h"
+
+/**
+ * Main struct containing ALTS shared resources used when
+ * employing the dedicated completion queue and thread.
+ */
+typedef struct alts_shared_resource_dedicated {
+  grpc_core::Thread thread;
+  grpc_completion_queue* cq;
+  grpc_pollset_set* interested_parties;
+  grpc_cq_completion storage;
+} alts_shared_resource_dedicated;
+
+/* This method returns the address of alts_shared_resource_dedicated
+ * object shared by all TSI handshakes.
+ */
+alts_shared_resource_dedicated* grpc_alts_get_shared_resource_dedicated(void);
+
+/**
+ * This method destroys the alts_shared_resource_dedicated object
+ * shared by all TSI handshakes. The applicaiton is responsible for
+ * invoking the API before calling grpc_shutdown().
+ */
+void grpc_alts_shared_resource_dedicated_shutdown();
+
+/**
+ * This method initializes the alts_shared_resource_dedicated object
+ * shared by all TSI handshakes. The application is responsible for
+ * invoking the API after calling grpc_init();
+ */
+void grpc_alts_shared_resource_dedicated_init();
+
+/**
+ * This method populates various fields of the alts_shared_resource_dedicated
+ * object shared by all TSI handshakes and start the dedicated thread.
+ * The API will be invoked by the caller in a lazy manner. That is,
+ * it will get invoked when ALTS TSI handshake occurs for the first time.
+ */
+void grpc_alts_shared_resource_dedicated_start();
+
+#endif /* GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_SHARED_RESOURCE_H \
+        */
diff --git a/src/core/tsi/alts/handshaker/alts_tsi_event.cc b/src/core/tsi/alts/handshaker/alts_tsi_event.cc
deleted file mode 100644
index cb36d5ebd1..0000000000
--- a/src/core/tsi/alts/handshaker/alts_tsi_event.cc
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- *
- * Copyright 2018 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#include <grpc/support/port_platform.h>
-
-#include "src/core/tsi/alts/handshaker/alts_tsi_event.h"
-
-#include <grpc/grpc.h>
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-
-#include "src/core/lib/slice/slice_internal.h"
-
-tsi_result alts_tsi_event_create(alts_tsi_handshaker* handshaker,
-                                 tsi_handshaker_on_next_done_cb cb,
-                                 void* user_data,
-                                 grpc_alts_credentials_options* options,
-                                 grpc_slice target_name,
-                                 alts_tsi_event** event) {
-  if (event == nullptr || handshaker == nullptr || cb == nullptr) {
-    gpr_log(GPR_ERROR, "Invalid arguments to alts_tsi_event_create()");
-    return TSI_INVALID_ARGUMENT;
-  }
-  alts_tsi_event* e = static_cast<alts_tsi_event*>(gpr_zalloc(sizeof(*e)));
-  e->handshaker = handshaker;
-  e->cb = cb;
-  e->user_data = user_data;
-  e->options = grpc_alts_credentials_options_copy(options);
-  e->target_name = grpc_slice_copy(target_name);
-  grpc_metadata_array_init(&e->initial_metadata);
-  grpc_metadata_array_init(&e->trailing_metadata);
-  *event = e;
-  return TSI_OK;
-}
-
-void alts_tsi_event_dispatch_to_handshaker(alts_tsi_event* event, bool is_ok) {
-  if (event == nullptr) {
-    gpr_log(
-        GPR_ERROR,
-        "ALTS TSI event is nullptr in alts_tsi_event_dispatch_to_handshaker()");
-    return;
-  }
-  alts_tsi_handshaker_handle_response(event->handshaker, event->recv_buffer,
-                                      event->status, &event->details, event->cb,
-                                      event->user_data, is_ok);
-}
-
-void alts_tsi_event_destroy(alts_tsi_event* event) {
-  if (event == nullptr) {
-    return;
-  }
-  grpc_byte_buffer_destroy(event->send_buffer);
-  grpc_byte_buffer_destroy(event->recv_buffer);
-  grpc_metadata_array_destroy(&event->initial_metadata);
-  grpc_metadata_array_destroy(&event->trailing_metadata);
-  grpc_slice_unref_internal(event->details);
-  grpc_slice_unref_internal(event->target_name);
-  grpc_alts_credentials_options_destroy(event->options);
-  gpr_free(event);
-}
diff --git a/src/core/tsi/alts/handshaker/alts_tsi_event.h b/src/core/tsi/alts/handshaker/alts_tsi_event.h
deleted file mode 100644
index 043e75d4a9..0000000000
--- a/src/core/tsi/alts/handshaker/alts_tsi_event.h
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- *
- * Copyright 2018 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-#ifndef GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_TSI_EVENT_H
-#define GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_TSI_EVENT_H
-
-#include <grpc/support/port_platform.h>
-
-#include <grpc/byte_buffer.h>
-#include <grpc/byte_buffer_reader.h>
-
-#include "src/core/tsi/alts/handshaker/alts_tsi_handshaker.h"
-#include "src/core/tsi/transport_security_interface.h"
-
-/**
- * A ALTS TSI event interface. In asynchronous implementation of
- * tsi_handshaker_next(), the function will exit after scheduling a handshaker
- * request to ALTS handshaker service without waiting for response to return.
- * The event is used to link the scheduled handshaker request with the
- * corresponding response so that enough context information can be inferred
- * from it to handle the response. All APIs in the header are thread-compatible.
- */
-
-/**
- * Main struct for ALTS TSI event. It retains ownership on send_buffer and
- * recv_buffer, but not on handshaker.
- */
-typedef struct alts_tsi_event {
-  alts_tsi_handshaker* handshaker;
-  grpc_byte_buffer* send_buffer;
-  grpc_byte_buffer* recv_buffer;
-  grpc_status_code status;
-  grpc_slice details;
-  grpc_metadata_array initial_metadata;
-  grpc_metadata_array trailing_metadata;
-  tsi_handshaker_on_next_done_cb cb;
-  void* user_data;
-  grpc_alts_credentials_options* options;
-  grpc_slice target_name;
-} alts_tsi_event;
-
-/**
- * This method creates a ALTS TSI event.
- *
- * - handshaker: ALTS TSI handshaker instance associated with the event to be
- *   created. The created event does not own the handshaker instance.
- * - cb: callback function to be called when handling data received from ALTS
- *   handshaker service.
- * - user_data: argument to callback function.
- * - options: ALTS credentials options.
- * - target_name: name of endpoint used for secure naming check.
- * - event: address of ALTS TSI event instance to be returned from the method.
- *
- * It returns TSI_OK on success and an error status code on failure.
- */
-tsi_result alts_tsi_event_create(alts_tsi_handshaker* handshaker,
-                                 tsi_handshaker_on_next_done_cb cb,
-                                 void* user_data,
-                                 grpc_alts_credentials_options* options,
-                                 grpc_slice target_name,
-                                 alts_tsi_event** event);
-
-/**
- * This method dispatches a ALTS TSI event received from the handshaker service,
- * and a boolean flag indicating if the event is valid to read to ALTS TSI
- * handshaker to process. It is called by TSI thread.
- *
- * - event: ALTS TSI event instance.
- * - is_ok: a boolean value indicating if the event is valid to read.
- */
-void alts_tsi_event_dispatch_to_handshaker(alts_tsi_event* event, bool is_ok);
-
-/**
- * This method destroys the ALTS TSI event.
- */
-void alts_tsi_event_destroy(alts_tsi_event* event);
-
-#endif /* GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_TSI_EVENT_H */
diff --git a/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc b/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc
index dfdd659b87..1639252883 100644
--- a/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc
+++ b/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc
@@ -26,34 +26,37 @@
 
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
 #include <grpc/support/sync.h>
 #include <grpc/support/thd_id.h>
 
 #include "src/core/lib/gpr/host_port.h"
 #include "src/core/lib/gprpp/thd.h"
+#include "src/core/lib/iomgr/closure.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/tsi/alts/frame_protector/alts_frame_protector.h"
 #include "src/core/tsi/alts/handshaker/alts_handshaker_client.h"
+#include "src/core/tsi/alts/handshaker/alts_shared_resource.h"
 #include "src/core/tsi/alts/handshaker/alts_tsi_utils.h"
 #include "src/core/tsi/alts/zero_copy_frame_protector/alts_zero_copy_grpc_protector.h"
 #include "src/core/tsi/alts_transport_security.h"
 
-#define TSI_ALTS_INITIAL_BUFFER_SIZE 256
-
-static alts_shared_resource* kSharedResource = alts_get_shared_resource();
+static alts_shared_resource* g_shared_resources = alts_get_shared_resource();
 
 /* Main struct for ALTS TSI handshaker. */
-typedef struct alts_tsi_handshaker {
+struct alts_tsi_handshaker {
   tsi_handshaker base;
   alts_handshaker_client* client;
-  grpc_slice recv_bytes;
   grpc_slice target_name;
-  unsigned char* buffer;
-  size_t buffer_size;
   bool is_client;
   bool has_sent_start_message;
+  bool has_created_handshaker_client;
+  bool use_dedicated_cq;
+  char* handshaker_service_url;
+  grpc_pollset_set* interested_parties;
   grpc_alts_credentials_options* options;
-} alts_tsi_handshaker;
+  alts_handshaker_client_vtable* client_vtable_for_testing;
+};
 
 /* Main struct for ALTS TSI handshaker result. */
 typedef struct alts_tsi_handshaker_result {
@@ -193,9 +196,9 @@ static const tsi_handshaker_result_vtable result_vtable = {
     handshaker_result_create_frame_protector,
     handshaker_result_get_unused_bytes, handshaker_result_destroy};
 
-static tsi_result create_handshaker_result(grpc_gcp_handshaker_resp* resp,
-                                           bool is_client,
-                                           tsi_handshaker_result** self) {
+tsi_result alts_tsi_handshaker_result_create(grpc_gcp_handshaker_resp* resp,
+                                             bool is_client,
+                                             tsi_handshaker_result** self) {
   if (self == nullptr || resp == nullptr) {
     gpr_log(GPR_ERROR, "Invalid arguments to create_handshaker_result()");
     return TSI_INVALID_ARGUMENT;
@@ -234,6 +237,41 @@ static tsi_result create_handshaker_result(grpc_gcp_handshaker_resp* resp,
   return TSI_OK;
 }
 
+static void init_shared_resources(const char* handshaker_service_url,
+                                  bool use_dedicated_cq) {
+  GPR_ASSERT(handshaker_service_url != nullptr);
+  gpr_mu_lock(&g_shared_resources->mu);
+  if (g_shared_resources->channel == nullptr) {
+    g_shared_resources->channel =
+        grpc_insecure_channel_create(handshaker_service_url, nullptr, nullptr);
+    if (use_dedicated_cq) {
+      grpc_alts_shared_resource_dedicated_start();
+    }
+  }
+  gpr_mu_unlock(&g_shared_resources->mu);
+}
+
+/* gRPC provided callback used when gRPC thread model is applied. */
+static void on_handshaker_service_resp_recv(void* arg, grpc_error* error) {
+  alts_handshaker_client* client = static_cast<alts_handshaker_client*>(arg);
+  if (client == nullptr) {
+    gpr_log(GPR_ERROR, "ALTS handshaker client is nullptr");
+    return;
+  }
+  alts_handshaker_client_handle_response(client, true);
+}
+
+/* gRPC provided callback used when dedicatd CQ and thread are used.
+ * It serves to safely bring the control back to application. */
+static void on_handshaker_service_resp_recv_dedicated(void* arg,
+                                                      grpc_error* error) {
+  alts_shared_resource_dedicated* resource =
+      grpc_alts_get_shared_resource_dedicated();
+  grpc_cq_end_op(resource->cq, arg, GRPC_ERROR_NONE,
+                 [](void* done_arg, grpc_cq_completion* storage) {}, nullptr,
+                 &resource->storage);
+}
+
 static tsi_result handshaker_next(
     tsi_handshaker* self, const unsigned char* received_bytes,
     size_t received_bytes_size, const unsigned char** bytes_to_send,
@@ -250,12 +288,32 @@ static tsi_result handshaker_next(
   alts_tsi_handshaker* handshaker =
       reinterpret_cast<alts_tsi_handshaker*>(self);
   tsi_result ok = TSI_OK;
-  alts_tsi_event* event = nullptr;
-  ok = alts_tsi_event_create(handshaker, cb, user_data, handshaker->options,
-                             handshaker->target_name, &event);
-  if (ok != TSI_OK) {
-    gpr_log(GPR_ERROR, "Failed to create ALTS TSI event");
-    return ok;
+  if (!handshaker->has_created_handshaker_client) {
+    init_shared_resources(handshaker->handshaker_service_url,
+                          handshaker->use_dedicated_cq);
+    if (handshaker->use_dedicated_cq) {
+      handshaker->interested_parties =
+          grpc_alts_get_shared_resource_dedicated()->interested_parties;
+      GPR_ASSERT(handshaker->interested_parties != nullptr);
+    }
+    grpc_iomgr_cb_func grpc_cb = handshaker->use_dedicated_cq
+                                     ? on_handshaker_service_resp_recv_dedicated
+                                     : on_handshaker_service_resp_recv;
+    handshaker->client = alts_grpc_handshaker_client_create(
+        handshaker, g_shared_resources->channel,
+        handshaker->handshaker_service_url, handshaker->interested_parties,
+        handshaker->options, handshaker->target_name, grpc_cb, cb, user_data,
+        handshaker->client_vtable_for_testing, handshaker->is_client);
+    if (handshaker->client == nullptr) {
+      gpr_log(GPR_ERROR, "Failed to create ALTS handshaker client");
+      return TSI_FAILED_PRECONDITION;
+    }
+    handshaker->has_created_handshaker_client = true;
+  }
+  if (handshaker->use_dedicated_cq &&
+      handshaker->client_vtable_for_testing == nullptr) {
+    GPR_ASSERT(grpc_cq_begin_op(grpc_alts_get_shared_resource_dedicated()->cq,
+                                handshaker->client));
   }
   grpc_slice slice = (received_bytes == nullptr || received_bytes_size == 0)
                          ? grpc_empty_slice()
@@ -264,16 +322,11 @@ static tsi_result handshaker_next(
                                received_bytes_size);
   if (!handshaker->has_sent_start_message) {
     ok = handshaker->is_client
-             ? alts_handshaker_client_start_client(handshaker->client, event)
-             : alts_handshaker_client_start_server(handshaker->client, event,
-                                                   &slice);
+             ? alts_handshaker_client_start_client(handshaker->client)
+             : alts_handshaker_client_start_server(handshaker->client, &slice);
     handshaker->has_sent_start_message = true;
   } else {
-    if (!GRPC_SLICE_IS_EMPTY(handshaker->recv_bytes)) {
-      grpc_slice_unref_internal(handshaker->recv_bytes);
-    }
-    handshaker->recv_bytes = grpc_slice_ref(slice);
-    ok = alts_handshaker_client_next(handshaker->client, event, &slice);
+    ok = alts_handshaker_client_next(handshaker->client, &slice);
   }
   grpc_slice_unref_internal(slice);
   if (ok != TSI_OK) {
@@ -283,6 +336,22 @@ static tsi_result handshaker_next(
   return TSI_ASYNC;
 }
 
+/*
+ * This API will be invoked by a non-gRPC application, and an ExecCtx needs
+ * to be explicitly created in order to invoke ALTS handshaker client API's
+ * that assumes the caller is inside gRPC core.
+ */
+static tsi_result handshaker_next_dedicated(
+    tsi_handshaker* self, const unsigned char* received_bytes,
+    size_t received_bytes_size, const unsigned char** bytes_to_send,
+    size_t* bytes_to_send_size, tsi_handshaker_result** result,
+    tsi_handshaker_on_next_done_cb cb, void* user_data) {
+  grpc_core::ExecCtx exec_ctx;
+  return handshaker_next(self, received_bytes, received_bytes_size,
+                         bytes_to_send, bytes_to_send_size, result, cb,
+                         user_data);
+}
+
 static void handshaker_shutdown(tsi_handshaker* self) {
   GPR_ASSERT(self != nullptr);
   if (self->handshake_shutdown) {
@@ -300,10 +369,9 @@ static void handshaker_destroy(tsi_handshaker* self) {
   alts_tsi_handshaker* handshaker =
       reinterpret_cast<alts_tsi_handshaker*>(self);
   alts_handshaker_client_destroy(handshaker->client);
-  grpc_slice_unref_internal(handshaker->recv_bytes);
   grpc_slice_unref_internal(handshaker->target_name);
   grpc_alts_credentials_options_destroy(handshaker->options);
-  gpr_free(handshaker->buffer);
+  gpr_free(handshaker->handshaker_service_url);
   gpr_free(handshaker);
 }
 
@@ -313,36 +381,19 @@ static const tsi_handshaker_vtable handshaker_vtable = {
     nullptr,         handshaker_destroy,
     handshaker_next, handshaker_shutdown};
 
-static void thread_worker(void* arg) {
-  while (true) {
-    grpc_event event = grpc_completion_queue_next(
-        kSharedResource->cq, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr);
-    GPR_ASSERT(event.type != GRPC_QUEUE_TIMEOUT);
-    if (event.type == GRPC_QUEUE_SHUTDOWN) {
-      /* signal alts_tsi_shutdown() to destroy completion queue. */
-      grpc_tsi_alts_signal_for_cq_destroy();
-      break;
-    }
-    /* event.type == GRPC_OP_COMPLETE. */
-    alts_tsi_event* alts_event = static_cast<alts_tsi_event*>(event.tag);
-    alts_tsi_event_dispatch_to_handshaker(alts_event, event.success);
-    alts_tsi_event_destroy(alts_event);
-  }
-}
-
-static void init_shared_resources(const char* handshaker_service_url) {
-  GPR_ASSERT(handshaker_service_url != nullptr);
-  gpr_mu_lock(&kSharedResource->mu);
-  if (kSharedResource->channel == nullptr) {
-    gpr_cv_init(&kSharedResource->cv);
-    kSharedResource->channel =
-        grpc_insecure_channel_create(handshaker_service_url, nullptr, nullptr);
-    kSharedResource->cq = grpc_completion_queue_create_for_next(nullptr);
-    kSharedResource->thread =
-        grpc_core::Thread("alts_tsi_handshaker", &thread_worker, nullptr);
-    kSharedResource->thread.Start();
-  }
-  gpr_mu_unlock(&kSharedResource->mu);
+static const tsi_handshaker_vtable handshaker_vtable_dedicated = {
+    nullptr,
+    nullptr,
+    nullptr,
+    nullptr,
+    nullptr,
+    handshaker_destroy,
+    handshaker_next_dedicated,
+    handshaker_shutdown};
+
+bool alts_tsi_handshaker_has_shutdown(alts_tsi_handshaker* handshaker) {
+  GPR_ASSERT(handshaker != nullptr);
+  return handshaker->base.handshake_shutdown;
 }
 
 tsi_result alts_tsi_handshaker_create(
@@ -354,40 +405,29 @@ tsi_result alts_tsi_handshaker_create(
     gpr_log(GPR_ERROR, "Invalid arguments to alts_tsi_handshaker_create()");
     return TSI_INVALID_ARGUMENT;
   }
-  init_shared_resources(handshaker_service_url);
-  alts_handshaker_client* client = alts_grpc_handshaker_client_create(
-      kSharedResource->channel, kSharedResource->cq, handshaker_service_url);
-  if (client == nullptr) {
-    gpr_log(GPR_ERROR, "Failed to create ALTS handshaker client");
-    return TSI_FAILED_PRECONDITION;
-  }
   alts_tsi_handshaker* handshaker =
       static_cast<alts_tsi_handshaker*>(gpr_zalloc(sizeof(*handshaker)));
-  handshaker->client = client;
-  handshaker->buffer_size = TSI_ALTS_INITIAL_BUFFER_SIZE;
-  handshaker->buffer =
-      static_cast<unsigned char*>(gpr_zalloc(handshaker->buffer_size));
+  handshaker->use_dedicated_cq = interested_parties == nullptr;
+  handshaker->client = nullptr;
   handshaker->is_client = is_client;
   handshaker->has_sent_start_message = false;
   handshaker->target_name = target_name == nullptr
                                 ? grpc_empty_slice()
                                 : grpc_slice_from_static_string(target_name);
+  handshaker->interested_parties = interested_parties;
+  handshaker->has_created_handshaker_client = false;
+  handshaker->handshaker_service_url = gpr_strdup(handshaker_service_url);
   handshaker->options = grpc_alts_credentials_options_copy(options);
-  handshaker->base.vtable = &handshaker_vtable;
+  handshaker->base.vtable = handshaker->use_dedicated_cq
+                                ? &handshaker_vtable_dedicated
+                                : &handshaker_vtable;
   *self = &handshaker->base;
   return TSI_OK;
 }
 
-static bool is_handshake_finished_properly(grpc_gcp_handshaker_resp* resp) {
-  GPR_ASSERT(resp != nullptr);
-  if (resp->has_result) {
-    return true;
-  }
-  return false;
-}
-
-static void set_unused_bytes(tsi_handshaker_result* self,
-                             grpc_slice* recv_bytes, size_t bytes_consumed) {
+void alts_tsi_handshaker_result_set_unused_bytes(tsi_handshaker_result* self,
+                                                 grpc_slice* recv_bytes,
+                                                 size_t bytes_consumed) {
   GPR_ASSERT(recv_bytes != nullptr && self != nullptr);
   if (GRPC_SLICE_LENGTH(*recv_bytes) == bytes_consumed) {
     return;
@@ -402,81 +442,6 @@ static void set_unused_bytes(tsi_handshaker_result* self,
          result->unused_bytes_size);
 }
 
-void alts_tsi_handshaker_handle_response(alts_tsi_handshaker* handshaker,
-                                         grpc_byte_buffer* recv_buffer,
-                                         grpc_status_code status,
-                                         grpc_slice* details,
-                                         tsi_handshaker_on_next_done_cb cb,
-                                         void* user_data, bool is_ok) {
-  /* Invalid input check. */
-  if (cb == nullptr) {
-    gpr_log(GPR_ERROR,
-            "cb is nullptr in alts_tsi_handshaker_handle_response()");
-    return;
-  }
-  if (handshaker == nullptr || recv_buffer == nullptr) {
-    gpr_log(GPR_ERROR,
-            "Invalid arguments to alts_tsi_handshaker_handle_response()");
-    cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr);
-    return;
-  }
-  if (handshaker->base.handshake_shutdown) {
-    gpr_log(GPR_ERROR, "TSI handshake shutdown");
-    cb(TSI_HANDSHAKE_SHUTDOWN, user_data, nullptr, 0, nullptr);
-    return;
-  }
-  /* Failed grpc call check. */
-  if (!is_ok || status != GRPC_STATUS_OK) {
-    gpr_log(GPR_ERROR, "grpc call made to handshaker service failed");
-    if (details != nullptr) {
-      char* error_details = grpc_slice_to_c_string(*details);
-      gpr_log(GPR_ERROR, "error details:%s", error_details);
-      gpr_free(error_details);
-    }
-    cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr);
-    return;
-  }
-  grpc_gcp_handshaker_resp* resp =
-      alts_tsi_utils_deserialize_response(recv_buffer);
-  /* Invalid handshaker response check. */
-  if (resp == nullptr) {
-    gpr_log(GPR_ERROR, "alts_tsi_utils_deserialize_response() failed");
-    cb(TSI_DATA_CORRUPTED, user_data, nullptr, 0, nullptr);
-    return;
-  }
-  grpc_slice* slice = static_cast<grpc_slice*>(resp->out_frames.arg);
-  unsigned char* bytes_to_send = nullptr;
-  size_t bytes_to_send_size = 0;
-  if (slice != nullptr) {
-    bytes_to_send_size = GRPC_SLICE_LENGTH(*slice);
-    while (bytes_to_send_size > handshaker->buffer_size) {
-      handshaker->buffer_size *= 2;
-      handshaker->buffer = static_cast<unsigned char*>(
-          gpr_realloc(handshaker->buffer, handshaker->buffer_size));
-    }
-    memcpy(handshaker->buffer, GRPC_SLICE_START_PTR(*slice),
-           bytes_to_send_size);
-    bytes_to_send = handshaker->buffer;
-  }
-  tsi_handshaker_result* result = nullptr;
-  if (is_handshake_finished_properly(resp)) {
-    create_handshaker_result(resp, handshaker->is_client, &result);
-    set_unused_bytes(result, &handshaker->recv_bytes, resp->bytes_consumed);
-  }
-  grpc_status_code code = static_cast<grpc_status_code>(resp->status.code);
-  if (code != GRPC_STATUS_OK) {
-    grpc_slice* details = static_cast<grpc_slice*>(resp->status.details.arg);
-    if (details != nullptr) {
-      char* error_details = grpc_slice_to_c_string(*details);
-      gpr_log(GPR_ERROR, "Error from handshaker service:%s", error_details);
-      gpr_free(error_details);
-    }
-  }
-  grpc_gcp_handshaker_resp_destroy(resp);
-  cb(alts_tsi_utils_convert_to_tsi_result(code), user_data, bytes_to_send,
-     bytes_to_send_size, result);
-}
-
 namespace grpc_core {
 namespace internal {
 
@@ -486,29 +451,16 @@ bool alts_tsi_handshaker_get_has_sent_start_message_for_testing(
   return handshaker->has_sent_start_message;
 }
 
-bool alts_tsi_handshaker_get_is_client_for_testing(
-    alts_tsi_handshaker* handshaker) {
+void alts_tsi_handshaker_set_client_vtable_for_testing(
+    alts_tsi_handshaker* handshaker, alts_handshaker_client_vtable* vtable) {
   GPR_ASSERT(handshaker != nullptr);
-  return handshaker->is_client;
+  handshaker->client_vtable_for_testing = vtable;
 }
 
-void alts_tsi_handshaker_set_recv_bytes_for_testing(
-    alts_tsi_handshaker* handshaker, grpc_slice* slice) {
-  GPR_ASSERT(handshaker != nullptr && slice != nullptr);
-  handshaker->recv_bytes = grpc_slice_ref(*slice);
-}
-
-grpc_slice alts_tsi_handshaker_get_recv_bytes_for_testing(
+bool alts_tsi_handshaker_get_is_client_for_testing(
     alts_tsi_handshaker* handshaker) {
   GPR_ASSERT(handshaker != nullptr);
-  return handshaker->recv_bytes;
-}
-
-void alts_tsi_handshaker_set_client_for_testing(
-    alts_tsi_handshaker* handshaker, alts_handshaker_client* client) {
-  GPR_ASSERT(handshaker != nullptr && client != nullptr);
-  alts_handshaker_client_destroy(handshaker->client);
-  handshaker->client = client;
+  return handshaker->is_client;
 }
 
 alts_handshaker_client* alts_tsi_handshaker_get_client_for_testing(
diff --git a/src/core/tsi/alts/handshaker/alts_tsi_handshaker.h b/src/core/tsi/alts/handshaker/alts_tsi_handshaker.h
index 48ce69b1da..c961eae498 100644
--- a/src/core/tsi/alts/handshaker/alts_tsi_handshaker.h
+++ b/src/core/tsi/alts/handshaker/alts_tsi_handshaker.h
@@ -25,6 +25,8 @@
 
 #include "src/core/lib/iomgr/pollset_set.h"
 #include "src/core/lib/security/credentials/alts/grpc_alts_credentials_options.h"
+#include "src/core/tsi/alts/handshaker/alts_handshaker_client.h"
+#include "src/core/tsi/alts/handshaker/alts_handshaker_service_api_util.h"
 #include "src/core/tsi/alts_transport_security.h"
 #include "src/core/tsi/transport_security.h"
 #include "src/core/tsi/transport_security_interface.h"
@@ -35,10 +37,6 @@
 
 const size_t kTsiAltsNumOfPeerProperties = 3;
 
-/**
- * Main struct for ALTS TSI handshaker. All APIs in the header are
- * thread-comptabile.
- */
 typedef struct alts_tsi_handshaker alts_tsi_handshaker;
 
 /**
@@ -56,7 +54,9 @@ typedef struct alts_tsi_handshaker alts_tsi_handshaker;
  * - self: address of ALTS TSI handshaker instance to be returned from the
  *   method.
  *
- * It returns TSI_OK on success and an error status code on failure.
+ * It returns TSI_OK on success and an error status code on failure. Note that
+ * if interested_parties is nullptr, a dedicated TSI thread will be created and
+ * used.
  */
 tsi_result alts_tsi_handshaker_create(
     const grpc_alts_credentials_options* options, const char* target_name,
@@ -64,23 +64,32 @@ tsi_result alts_tsi_handshaker_create(
     grpc_pollset_set* interested_parties, tsi_handshaker** self);
 
 /**
- * This method handles handshaker response returned from ALTS handshaker
- * service.
+ * This method creates an ALTS TSI handshaker result instance.
  *
- * - handshaker: ALTS TSI handshaker instance.
- * - recv_buffer: buffer holding data received from the handshaker service.
- * - status: status of the grpc call made to the handshaker service.
- * - details: error details of the grpc call made to the handshaker service.
- * - cb: callback function of ALTS TSI event.
- * - user_data: argument of callback function.
- * - is_ok: a boolean value indicating if the handshaker response is ok to read.
+ * - resp: data received from the handshaker service.
+ * - is_client: a boolean value indicating if the result belongs to a
+ *   client or not.
+ * - result: address of ALTS TSI handshaker result instance.
+ */
+tsi_result alts_tsi_handshaker_result_create(grpc_gcp_handshaker_resp* resp,
+                                             bool is_client,
+                                             tsi_handshaker_result** result);
+
+/**
+ * This method sets unused bytes of ALTS TSI handshaker result instance.
  *
+ * - result: an ALTS TSI handshaker result instance.
+ * - recv_bytes: data received from the handshaker service.
+ * - bytes_consumed: size of data consumed by the handshaker service.
+ */
+void alts_tsi_handshaker_result_set_unused_bytes(tsi_handshaker_result* result,
+                                                 grpc_slice* recv_bytes,
+                                                 size_t bytes_consumed);
+
+/**
+ * This method returns a boolean value indicating if an ALTS TSI handshaker
+ * has been shutdown or not.
  */
-void alts_tsi_handshaker_handle_response(alts_tsi_handshaker* handshaker,
-                                         grpc_byte_buffer* recv_buffer,
-                                         grpc_status_code status,
-                                         grpc_slice* details,
-                                         tsi_handshaker_on_next_done_cb cb,
-                                         void* user_data, bool is_ok);
+bool alts_tsi_handshaker_has_shutdown(alts_tsi_handshaker* handshaker);
 
 #endif /* GRPC_CORE_TSI_ALTS_HANDSHAKER_ALTS_TSI_HANDSHAKER_H */
diff --git a/src/core/tsi/alts/handshaker/alts_tsi_handshaker_private.h b/src/core/tsi/alts/handshaker/alts_tsi_handshaker_private.h
index 9612071407..ec2616e95f 100644
--- a/src/core/tsi/alts/handshaker/alts_tsi_handshaker_private.h
+++ b/src/core/tsi/alts/handshaker/alts_tsi_handshaker_private.h
@@ -27,27 +27,55 @@ namespace grpc_core {
 namespace internal {
 
 /**
- * Unsafe, use for testing only. It allows the caller to change the way the
- * ALTS TSI handshaker schedules handshaker requests.
- */
-void alts_tsi_handshaker_set_client_for_testing(alts_tsi_handshaker* handshaker,
-                                                alts_handshaker_client* client);
+ * Unsafe, use for testing only. */
 
 alts_handshaker_client* alts_tsi_handshaker_get_client_for_testing(
     alts_tsi_handshaker* handshaker);
 
-/* For testing only. */
 bool alts_tsi_handshaker_get_has_sent_start_message_for_testing(
     alts_tsi_handshaker* handshaker);
 
+void alts_tsi_handshaker_set_client_vtable_for_testing(
+    alts_tsi_handshaker* handshaker, alts_handshaker_client_vtable* vtable);
+
 bool alts_tsi_handshaker_get_is_client_for_testing(
     alts_tsi_handshaker* handshaker);
 
-void alts_tsi_handshaker_set_recv_bytes_for_testing(
-    alts_tsi_handshaker* handshaker, grpc_slice* slice);
+void alts_handshaker_client_set_grpc_caller_for_testing(
+    alts_handshaker_client* client, alts_grpc_caller caller);
 
-grpc_slice alts_tsi_handshaker_get_recv_bytes_for_testing(
-    alts_tsi_handshaker* handshaker);
+grpc_byte_buffer* alts_handshaker_client_get_send_buffer_for_testing(
+    alts_handshaker_client* client);
+
+grpc_byte_buffer** alts_handshaker_client_get_recv_buffer_addr_for_testing(
+    alts_handshaker_client* client);
+
+grpc_metadata_array* alts_handshaker_client_get_initial_metadata_for_testing(
+    alts_handshaker_client* client);
+
+void alts_handshaker_client_set_recv_bytes_for_testing(
+    alts_handshaker_client* client, grpc_slice* recv_bytes);
+
+void alts_handshaker_client_check_fields_for_testing(
+    alts_handshaker_client* client, tsi_handshaker_on_next_done_cb cb,
+    void* user_data, bool has_sent_start_message, grpc_slice* recv_bytes);
+
+void alts_handshaker_client_set_fields_for_testing(
+    alts_handshaker_client* client, alts_tsi_handshaker* handshaker,
+    tsi_handshaker_on_next_done_cb cb, void* user_data,
+    grpc_byte_buffer* recv_buffer, grpc_status_code status);
+
+void alts_handshaker_client_set_vtable_for_testing(
+    alts_handshaker_client* client, alts_handshaker_client_vtable* vtable);
+
+alts_tsi_handshaker* alts_handshaker_client_get_handshaker_for_testing(
+    alts_handshaker_client* client);
+
+void alts_handshaker_client_set_cb_for_testing(
+    alts_handshaker_client* client, tsi_handshaker_on_next_done_cb cb);
+
+grpc_closure* alts_handshaker_client_get_closure_for_testing(
+    alts_handshaker_client* client);
 
 }  // namespace internal
 }  // namespace grpc_core
diff --git a/src/core/tsi/alts_transport_security.cc b/src/core/tsi/alts_transport_security.cc
index dac23bbf7a..5a1494ae5c 100644
--- a/src/core/tsi/alts_transport_security.cc
+++ b/src/core/tsi/alts_transport_security.cc
@@ -18,48 +18,24 @@
 
 #include <grpc/support/port_platform.h>
 
-#include "src/core/tsi/alts_transport_security.h"
-
 #include <string.h>
 
+#include "src/core/tsi/alts_transport_security.h"
+
 static alts_shared_resource g_alts_resource;
 
 alts_shared_resource* alts_get_shared_resource(void) {
   return &g_alts_resource;
 }
 
-static void grpc_tsi_alts_wait_for_cq_drain() {
-  gpr_mu_lock(&g_alts_resource.mu);
-  while (!g_alts_resource.is_cq_drained) {
-    gpr_cv_wait(&g_alts_resource.cv, &g_alts_resource.mu,
-                gpr_inf_future(GPR_CLOCK_REALTIME));
-  }
-  gpr_mu_unlock(&g_alts_resource.mu);
-}
-
-void grpc_tsi_alts_signal_for_cq_destroy() {
-  gpr_mu_lock(&g_alts_resource.mu);
-  g_alts_resource.is_cq_drained = true;
-  gpr_cv_signal(&g_alts_resource.cv);
-  gpr_mu_unlock(&g_alts_resource.mu);
-}
-
 void grpc_tsi_alts_init() {
   g_alts_resource.channel = nullptr;
-  g_alts_resource.cq = nullptr;
-  g_alts_resource.is_cq_drained = false;
   gpr_mu_init(&g_alts_resource.mu);
-  gpr_cv_init(&g_alts_resource.cv);
 }
 
 void grpc_tsi_alts_shutdown() {
-  if (g_alts_resource.cq != nullptr) {
-    grpc_completion_queue_shutdown(g_alts_resource.cq);
-    grpc_tsi_alts_wait_for_cq_drain();
-    grpc_completion_queue_destroy(g_alts_resource.cq);
+  if (g_alts_resource.channel != nullptr) {
     grpc_channel_destroy(g_alts_resource.channel);
-    g_alts_resource.thread.Join();
   }
-  gpr_cv_destroy(&g_alts_resource.cv);
   gpr_mu_destroy(&g_alts_resource.mu);
 }
diff --git a/src/core/tsi/alts_transport_security.h b/src/core/tsi/alts_transport_security.h
index d6b8e11137..f4319d10d2 100644
--- a/src/core/tsi/alts_transport_security.h
+++ b/src/core/tsi/alts_transport_security.h
@@ -27,21 +27,12 @@
 #include "src/core/lib/gprpp/thd.h"
 
 typedef struct alts_shared_resource {
-  grpc_core::Thread thread;
   grpc_channel* channel;
-  grpc_completion_queue* cq;
   gpr_mu mu;
-  gpr_cv cv;
-  bool is_cq_drained;
 } alts_shared_resource;
 
 /* This method returns the address of alts_shared_resource object shared by all
  *    TSI handshakes. */
 alts_shared_resource* alts_get_shared_resource(void);
 
-/* This method signals the thread that invokes grpc_tsi_alts_shutdown() to
- * continue with destroying the cq as a part of shutdown process. */
-
-void grpc_tsi_alts_signal_for_cq_destroy(void);
-
 #endif /* GRPC_CORE_TSI_ALTS_TRANSPORT_SECURITY_H */
diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index 7d9ce0cf28..42a4a7aa04 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -281,7 +281,7 @@ CORE_SOURCE_FILES = [
     'src/core/tsi/alts/frame_protector/alts_unseal_privacy_integrity_crypter.cc',
     'src/core/tsi/alts/frame_protector/frame_handler.cc',
     'src/core/tsi/alts/handshaker/alts_handshaker_client.cc',
-    'src/core/tsi/alts/handshaker/alts_tsi_event.cc',
+    'src/core/tsi/alts/handshaker/alts_shared_resource.cc',
     'src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc',
     'src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_integrity_only_record_protocol.cc',
     'src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_privacy_integrity_record_protocol.cc',