Merge pull request #12418 from jiangtaoli2016/fake_zero_copy_protector

Fake zero copy protector

3 files changed, 143 insertions, 103 deletions

diff --git a/src/core/lib/security/transport/secure_endpoint.c b/src/core/lib/security/transport/secure_endpoint.c
index 5e41b94ff8..ae5633b82c 100644
--- a/src/core/lib/security/transport/secure_endpoint.c
+++ b/src/core/lib/security/transport/secure_endpoint.c
@@ -34,7 +34,7 @@
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/slice/slice_string_helpers.h"
 #include "src/core/lib/support/string.h"
-#include "src/core/tsi/transport_security_interface.h"
+#include "src/core/tsi/transport_security_grpc.h"
 
 #define STAGING_BUFFER_SIZE 8192
 
@@ -42,6 +42,7 @@ typedef struct {
   grpc_endpoint base;
   grpc_endpoint *wrapped_ep;
   struct tsi_frame_protector *protector;
+  struct tsi_zero_copy_grpc_protector *zero_copy_protector;
   gpr_mu protector_mu;
   /* saved upper level callbacks and user_data. */
   grpc_closure *read_cb;
@@ -67,6 +68,7 @@ static void destroy(grpc_exec_ctx *exec_ctx, secure_endpoint *secure_ep) {
   secure_endpoint *ep = secure_ep;
   grpc_endpoint_destroy(exec_ctx, ep->wrapped_ep);
   tsi_frame_protector_destroy(ep->protector);
+  tsi_zero_copy_grpc_protector_destroy(exec_ctx, ep->zero_copy_protector);
   grpc_slice_buffer_destroy_internal(exec_ctx, &ep->leftover_bytes);
   grpc_slice_unref_internal(exec_ctx, ep->read_staging_buffer);
   grpc_slice_unref_internal(exec_ctx, ep->write_staging_buffer);
@@ -159,51 +161,58 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *user_data,
     return;
   }
 
-  /* TODO(yangg) check error, maybe bail out early */
-  for (i = 0; i < ep->source_buffer.count; i++) {
-    grpc_slice encrypted = ep->source_buffer.slices[i];
-    uint8_t *message_bytes = GRPC_SLICE_START_PTR(encrypted);
-    size_t message_size = GRPC_SLICE_LENGTH(encrypted);
-
-    while (message_size > 0 || keep_looping) {
-      size_t unprotected_buffer_size_written = (size_t)(end - cur);
-      size_t processed_message_size = message_size;
-      gpr_mu_lock(&ep->protector_mu);
-      result = tsi_frame_protector_unprotect(ep->protector, message_bytes,
-                                             &processed_message_size, cur,
-                                             &unprotected_buffer_size_written);
-      gpr_mu_unlock(&ep->protector_mu);
-      if (result != TSI_OK) {
-        gpr_log(GPR_ERROR, "Decryption error: %s",
-                tsi_result_to_string(result));
-        break;
-      }
-      message_bytes += processed_message_size;
-      message_size -= processed_message_size;
-      cur += unprotected_buffer_size_written;
-
-      if (cur == end) {
-        flush_read_staging_buffer(ep, &cur, &end);
-        /* Force to enter the loop again to extract buffered bytes in protector.
-           The bytes could be buffered because of running out of staging_buffer.
-           If this happens at the end of all slices, doing another unprotect
-           avoids leaving data in the protector. */
-        keep_looping = 1;
-      } else if (unprotected_buffer_size_written > 0) {
-        keep_looping = 1;
-      } else {
-        keep_looping = 0;
+  if (ep->zero_copy_protector != NULL) {
+    // Use zero-copy grpc protector to unprotect.
+    result = tsi_zero_copy_grpc_protector_unprotect(
+        exec_ctx, ep->zero_copy_protector, &ep->source_buffer, ep->read_buffer);
+  } else {
+    // Use frame protector to unprotect.
+    /* TODO(yangg) check error, maybe bail out early */
+    for (i = 0; i < ep->source_buffer.count; i++) {
+      grpc_slice encrypted = ep->source_buffer.slices[i];
+      uint8_t *message_bytes = GRPC_SLICE_START_PTR(encrypted);
+      size_t message_size = GRPC_SLICE_LENGTH(encrypted);
+
+      while (message_size > 0 || keep_looping) {
+        size_t unprotected_buffer_size_written = (size_t)(end - cur);
+        size_t processed_message_size = message_size;
+        gpr_mu_lock(&ep->protector_mu);
+        result = tsi_frame_protector_unprotect(
+            ep->protector, message_bytes, &processed_message_size, cur,
+            &unprotected_buffer_size_written);
+        gpr_mu_unlock(&ep->protector_mu);
+        if (result != TSI_OK) {
+          gpr_log(GPR_ERROR, "Decryption error: %s",
+                  tsi_result_to_string(result));
+          break;
+        }
+        message_bytes += processed_message_size;
+        message_size -= processed_message_size;
+        cur += unprotected_buffer_size_written;
+
+        if (cur == end) {
+          flush_read_staging_buffer(ep, &cur, &end);
+          /* Force to enter the loop again to extract buffered bytes in
+             protector. The bytes could be buffered because of running out of
+             staging_buffer. If this happens at the end of all slices, doing
+             another unprotect avoids leaving data in the protector. */
+          keep_looping = 1;
+        } else if (unprotected_buffer_size_written > 0) {
+          keep_looping = 1;
+        } else {
+          keep_looping = 0;
+        }
       }
+      if (result != TSI_OK) break;
     }
-    if (result != TSI_OK) break;
-  }
 
-  if (cur != GRPC_SLICE_START_PTR(ep->read_staging_buffer)) {
-    grpc_slice_buffer_add(
-        ep->read_buffer,
-        grpc_slice_split_head(
-            &ep->read_staging_buffer,
-            (size_t)(cur - GRPC_SLICE_START_PTR(ep->read_staging_buffer))));
+    if (cur != GRPC_SLICE_START_PTR(ep->read_staging_buffer)) {
+      grpc_slice_buffer_add(
+          ep->read_buffer,
+          grpc_slice_split_head(
+              &ep->read_staging_buffer,
+              (size_t)(cur - GRPC_SLICE_START_PTR(ep->read_staging_buffer))));
+    }
   }
 
   /* TODO(yangg) experiment with moving this block after read_cb to see if it
@@ -270,54 +279,62 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep,
     }
   }
 
-  for (i = 0; i < slices->count; i++) {
-    grpc_slice plain = slices->slices[i];
-    uint8_t *message_bytes = GRPC_SLICE_START_PTR(plain);
-    size_t message_size = GRPC_SLICE_LENGTH(plain);
-    while (message_size > 0) {
-      size_t protected_buffer_size_to_send = (size_t)(end - cur);
-      size_t processed_message_size = message_size;
-      gpr_mu_lock(&ep->protector_mu);
-      result = tsi_frame_protector_protect(ep->protector, message_bytes,
-                                           &processed_message_size, cur,
-                                           &protected_buffer_size_to_send);
-      gpr_mu_unlock(&ep->protector_mu);
-      if (result != TSI_OK) {
-        gpr_log(GPR_ERROR, "Encryption error: %s",
-                tsi_result_to_string(result));
-        break;
-      }
-      message_bytes += processed_message_size;
-      message_size -= processed_message_size;
-      cur += protected_buffer_size_to_send;
-
-      if (cur == end) {
-        flush_write_staging_buffer(ep, &cur, &end);
+  if (ep->zero_copy_protector != NULL) {
+    // Use zero-copy grpc protector to protect.
+    result = tsi_zero_copy_grpc_protector_protect(
+        exec_ctx, ep->zero_copy_protector, slices, &ep->output_buffer);
+  } else {
+    // Use frame protector to protect.
+    for (i = 0; i < slices->count; i++) {
+      grpc_slice plain = slices->slices[i];
+      uint8_t *message_bytes = GRPC_SLICE_START_PTR(plain);
+      size_t message_size = GRPC_SLICE_LENGTH(plain);
+      while (message_size > 0) {
+        size_t protected_buffer_size_to_send = (size_t)(end - cur);
+        size_t processed_message_size = message_size;
+        gpr_mu_lock(&ep->protector_mu);
+        result = tsi_frame_protector_protect(ep->protector, message_bytes,
+                                             &processed_message_size, cur,
+                                             &protected_buffer_size_to_send);
+        gpr_mu_unlock(&ep->protector_mu);
+        if (result != TSI_OK) {
+          gpr_log(GPR_ERROR, "Encryption error: %s",
+                  tsi_result_to_string(result));
+          break;
+        }
+        message_bytes += processed_message_size;
+        message_size -= processed_message_size;
+        cur += protected_buffer_size_to_send;
+
+        if (cur == end) {
+          flush_write_staging_buffer(ep, &cur, &end);
+        }
       }
-    }
-    if (result != TSI_OK) break;
-  }
-  if (result == TSI_OK) {
-    size_t still_pending_size;
-    do {
-      size_t protected_buffer_size_to_send = (size_t)(end - cur);
-      gpr_mu_lock(&ep->protector_mu);
-      result = tsi_frame_protector_protect_flush(ep->protector, cur,
-                                                 &protected_buffer_size_to_send,
-                                                 &still_pending_size);
-      gpr_mu_unlock(&ep->protector_mu);
       if (result != TSI_OK) break;
-      cur += protected_buffer_size_to_send;
-      if (cur == end) {
-        flush_write_staging_buffer(ep, &cur, &end);
+    }
+    if (result == TSI_OK) {
+      size_t still_pending_size;
+      do {
+        size_t protected_buffer_size_to_send = (size_t)(end - cur);
+        gpr_mu_lock(&ep->protector_mu);
+        result = tsi_frame_protector_protect_flush(
+            ep->protector, cur, &protected_buffer_size_to_send,
+            &still_pending_size);
+        gpr_mu_unlock(&ep->protector_mu);
+        if (result != TSI_OK) break;
+        cur += protected_buffer_size_to_send;
+        if (cur == end) {
+          flush_write_staging_buffer(ep, &cur, &end);
+        }
+      } while (still_pending_size > 0);
+      if (cur != GRPC_SLICE_START_PTR(ep->write_staging_buffer)) {
+        grpc_slice_buffer_add(
+            &ep->output_buffer,
+            grpc_slice_split_head(
+                &ep->write_staging_buffer,
+                (size_t)(cur -
+                         GRPC_SLICE_START_PTR(ep->write_staging_buffer))));
       }
-    } while (still_pending_size > 0);
-    if (cur != GRPC_SLICE_START_PTR(ep->write_staging_buffer)) {
-      grpc_slice_buffer_add(
-          &ep->output_buffer,
-          grpc_slice_split_head(
-              &ep->write_staging_buffer,
-              (size_t)(cur - GRPC_SLICE_START_PTR(ep->write_staging_buffer))));
     }
   }
 
@@ -389,13 +406,16 @@ static const grpc_endpoint_vtable vtable = {endpoint_read,
                                             endpoint_get_fd};
 
 grpc_endpoint *grpc_secure_endpoint_create(
-    struct tsi_frame_protector *protector, grpc_endpoint *transport,
-    grpc_slice *leftover_slices, size_t leftover_nslices) {
+    struct tsi_frame_protector *protector,
+    struct tsi_zero_copy_grpc_protector *zero_copy_protector,
+    grpc_endpoint *transport, grpc_slice *leftover_slices,
+    size_t leftover_nslices) {
   size_t i;
   secure_endpoint *ep = (secure_endpoint *)gpr_malloc(sizeof(secure_endpoint));
   ep->base.vtable = &vtable;
   ep->wrapped_ep = transport;
   ep->protector = protector;
+  ep->zero_copy_protector = zero_copy_protector;
   grpc_slice_buffer_init(&ep->leftover_bytes);
   for (i = 0; i < leftover_nslices; i++) {
     grpc_slice_buffer_add(&ep->leftover_bytes,
diff --git a/src/core/lib/security/transport/secure_endpoint.h b/src/core/lib/security/transport/secure_endpoint.h
index 1c5555f3df..3323a6ff42 100644
--- a/src/core/lib/security/transport/secure_endpoint.h
+++ b/src/core/lib/security/transport/secure_endpoint.h
@@ -23,12 +23,17 @@
 #include "src/core/lib/iomgr/endpoint.h"
 
 struct tsi_frame_protector;
+struct tsi_zero_copy_grpc_protector;
 
 extern grpc_tracer_flag grpc_trace_secure_endpoint;
 
-/* Takes ownership of protector and to_wrap, and refs leftover_slices. */
+/* Takes ownership of protector, zero_copy_protector, and to_wrap, and refs
+ * leftover_slices. If zero_copy_protector is not NULL, protector will never be
+ * used. */
 grpc_endpoint *grpc_secure_endpoint_create(
-    struct tsi_frame_protector *protector, grpc_endpoint *to_wrap,
-    grpc_slice *leftover_slices, size_t leftover_nslices);
+    struct tsi_frame_protector *protector,
+    struct tsi_zero_copy_grpc_protector *zero_copy_protector,
+    grpc_endpoint *to_wrap, grpc_slice *leftover_slices,
+    size_t leftover_nslices);
 
 #endif /* GRPC_CORE_LIB_SECURITY_TRANSPORT_SECURE_ENDPOINT_H */
diff --git a/src/core/lib/security/transport/security_handshaker.c b/src/core/lib/security/transport/security_handshaker.c
index fc9c9f980f..ea9608f444 100644
--- a/src/core/lib/security/transport/security_handshaker.c
+++ b/src/core/lib/security/transport/security_handshaker.c
@@ -32,6 +32,7 @@
 #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"
+#include "src/core/tsi/transport_security_grpc.h"
 
 #define GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE 256
 
@@ -135,17 +136,31 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
     security_handshake_failed_locked(exec_ctx, h, GRPC_ERROR_REF(error));
     goto done;
   }
-  // Create frame protector.
-  tsi_frame_protector *protector;
-  tsi_result result = tsi_handshaker_result_create_frame_protector(
-      h->handshaker_result, NULL, &protector);
-  if (result != TSI_OK) {
+  // Create zero-copy frame protector, if implemented.
+  tsi_zero_copy_grpc_protector *zero_copy_protector = NULL;
+  tsi_result result = tsi_handshaker_result_create_zero_copy_grpc_protector(
+      h->handshaker_result, NULL, &zero_copy_protector);
+  if (result != TSI_OK && result != TSI_UNIMPLEMENTED) {
     error = grpc_set_tsi_error_result(
-        GRPC_ERROR_CREATE_FROM_STATIC_STRING("Frame protector creation failed"),
+        GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+            "Zero-copy frame protector creation failed"),
         result);
     security_handshake_failed_locked(exec_ctx, h, error);
     goto done;
   }
+  // Create frame protector if zero-copy frame protector is NULL.
+  tsi_frame_protector *protector = NULL;
+  if (zero_copy_protector == NULL) {
+    result = tsi_handshaker_result_create_frame_protector(h->handshaker_result,
+                                                          NULL, &protector);
+    if (result != TSI_OK) {
+      error = grpc_set_tsi_error_result(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+                                            "Frame protector creation failed"),
+                                        result);
+      security_handshake_failed_locked(exec_ctx, h, error);
+      goto done;
+    }
+  }
   // Get unused bytes.
   const unsigned char *unused_bytes = NULL;
   size_t unused_bytes_size = 0;
@@ -155,12 +170,12 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
   if (unused_bytes_size > 0) {
     grpc_slice slice =
         grpc_slice_from_copied_buffer((char *)unused_bytes, unused_bytes_size);
-    h->args->endpoint =
-        grpc_secure_endpoint_create(protector, h->args->endpoint, &slice, 1);
+    h->args->endpoint = grpc_secure_endpoint_create(
+        protector, zero_copy_protector, h->args->endpoint, &slice, 1);
     grpc_slice_unref_internal(exec_ctx, slice);
   } else {
-    h->args->endpoint =
-        grpc_secure_endpoint_create(protector, h->args->endpoint, NULL, 0);
+    h->args->endpoint = grpc_secure_endpoint_create(
+        protector, zero_copy_protector, h->args->endpoint, NULL, 0);
   }
   tsi_handshaker_result_destroy(h->handshaker_result);
   h->handshaker_result = NULL;