gpr_slice grpc_chttp2_rst_stream_create(uint32_t id, uint32_t code,
                                        grpc_transport_one_way_stats *stats) {
  static const size_t frame_size = 13;
  gpr_slice slice = gpr_slice_malloc(frame_size);
  stats->framing_bytes += frame_size;
  uint8_t *p = GPR_SLICE_START_PTR(slice);

  *p++ = 0;
  *p++ = 0;
  *p++ = 4;
  *p++ = GRPC_CHTTP2_FRAME_RST_STREAM;
  *p++ = 0;
  *p++ = (uint8_t)(id >> 24);
  *p++ = (uint8_t)(id >> 16);
  *p++ = (uint8_t)(id >> 8);
  *p++ = (uint8_t)(id);
  *p++ = (uint8_t)(code >> 24);
  *p++ = (uint8_t)(code >> 16);
  *p++ = (uint8_t)(code >> 8);
  *p++ = (uint8_t)(code);

  return slice;
}

static gpr_slice *allocate_blocks(size_t num_bytes, size_t slice_size,
                                  size_t *num_blocks, uint8_t *current_data) {
  size_t nslices = num_bytes / slice_size + (num_bytes % slice_size ? 1 : 0);
  gpr_slice *slices = malloc(sizeof(gpr_slice) * nslices);
  size_t num_bytes_left = num_bytes;
  size_t i;
  size_t j;
  unsigned char *buf;
  *num_blocks = nslices;

  for (i = 0; i < nslices; ++i) {
    slices[i] = gpr_slice_malloc(slice_size > num_bytes_left ? num_bytes_left
                                                             : slice_size);
    num_bytes_left -= GPR_SLICE_LENGTH(slices[i]);
    buf = GPR_SLICE_START_PTR(slices[i]);
    for (j = 0; j < GPR_SLICE_LENGTH(slices[i]); ++j) {
      buf[j] = *current_data;
      (*current_data)++;
    }
  }
  GPR_ASSERT(num_bytes_left == 0);
  return slices;
}

/* Takes ownership of creds_path if not NULL. */
static grpc_call_credentials *create_default_creds_from_path(char *creds_path) {
  grpc_json *json = NULL;
  grpc_auth_json_key key;
  grpc_auth_refresh_token token;
  grpc_call_credentials *result = NULL;
  gpr_slice creds_data = gpr_empty_slice();
  int file_ok = 0;
  if (creds_path == NULL) goto end;
  creds_data = gpr_load_file(creds_path, 0, &file_ok);
  if (!file_ok) goto end;
  json = grpc_json_parse_string_with_len(
      (char *)GPR_SLICE_START_PTR(creds_data), GPR_SLICE_LENGTH(creds_data));
  if (json == NULL) goto end;

  /* First, try an auth json key. */
  key = grpc_auth_json_key_create_from_json(json);
  if (grpc_auth_json_key_is_valid(&key)) {
    result =
        grpc_service_account_jwt_access_credentials_create_from_auth_json_key(
            key, grpc_max_auth_token_lifetime());
    goto end;
  }

  /* Then try a refresh token if the auth json key was invalid. */
  token = grpc_auth_refresh_token_create_from_json(json);
  if (grpc_auth_refresh_token_is_valid(&token)) {
    result =
        grpc_refresh_token_credentials_create_from_auth_refresh_token(token);
    goto end;
  }

end:
  if (creds_path != NULL) gpr_free(creds_path);
  gpr_slice_unref(creds_data);
  if (json != NULL) grpc_json_destroy(json);
  return result;
}

static void check_jwt_signature(const char *b64_signature, RSA *rsa_key,
                                const char *signed_data,
                                size_t signed_data_size) {
  EVP_MD_CTX *md_ctx = EVP_MD_CTX_create();
  EVP_PKEY *key = EVP_PKEY_new();

  gpr_slice sig = grpc_base64_decode(b64_signature, 1);
  GPR_ASSERT(!GPR_SLICE_IS_EMPTY(sig));
  GPR_ASSERT(GPR_SLICE_LENGTH(sig) == 128);

  GPR_ASSERT(md_ctx != NULL);
  GPR_ASSERT(key != NULL);
  EVP_PKEY_set1_RSA(key, rsa_key);

  GPR_ASSERT(EVP_DigestVerifyInit(md_ctx, NULL, EVP_sha256(), NULL, key) == 1);
  GPR_ASSERT(EVP_DigestVerifyUpdate(md_ctx, signed_data, signed_data_size) ==
             1);
  GPR_ASSERT(EVP_DigestVerifyFinal(md_ctx, GPR_SLICE_START_PTR(sig),
                                   GPR_SLICE_LENGTH(sig)) == 1);

  gpr_slice_unref(sig);
  if (key != NULL) EVP_PKEY_free(key);
  if (md_ctx != NULL) EVP_MD_CTX_destroy(md_ctx);
}

static void test_expired_claims_failure(void) {
  grpc_jwt_claims *claims;
  gpr_slice s = gpr_slice_from_copied_string(expired_claims);
  grpc_json *json = grpc_json_parse_string_with_len(
      (char *)GPR_SLICE_START_PTR(s), GPR_SLICE_LENGTH(s));
  gpr_timespec exp_iat = {100, 0, GPR_CLOCK_REALTIME};
  gpr_timespec exp_exp = {120, 0, GPR_CLOCK_REALTIME};
  gpr_timespec exp_nbf = {60, 0, GPR_CLOCK_REALTIME};
  GPR_ASSERT(json != NULL);
  claims = grpc_jwt_claims_from_json(json, s);
  GPR_ASSERT(claims != NULL);
  GPR_ASSERT(grpc_jwt_claims_json(claims) == json);
  GPR_ASSERT(strcmp(grpc_jwt_claims_audience(claims), "https://foo.com") == 0);
  GPR_ASSERT(strcmp(grpc_jwt_claims_issuer(claims), "blah.foo.com") == 0);
  GPR_ASSERT(strcmp(grpc_jwt_claims_subject(claims), "[email protected]") == 0);
  GPR_ASSERT(strcmp(grpc_jwt_claims_id(claims), "jwtuniqueid") == 0);
  GPR_ASSERT(gpr_time_cmp(grpc_jwt_claims_issued_at(claims), exp_iat) == 0);
  GPR_ASSERT(gpr_time_cmp(grpc_jwt_claims_expires_at(claims), exp_exp) == 0);
  GPR_ASSERT(gpr_time_cmp(grpc_jwt_claims_not_before(claims), exp_nbf) == 0);

  GPR_ASSERT(grpc_jwt_claims_check(claims, "https://foo.com") ==
             GRPC_JWT_VERIFIER_TIME_CONSTRAINT_FAILURE);
  grpc_jwt_claims_destroy(claims);
}

void grpc_chttp2_goaway_append(uint32_t last_stream_id, uint32_t error_code,
                               gpr_slice debug_data,
                               gpr_slice_buffer *slice_buffer) {
  gpr_slice header = gpr_slice_malloc(9 + 4 + 4);
  uint8_t *p = GPR_SLICE_START_PTR(header);
  uint32_t frame_length;
  GPR_ASSERT(GPR_SLICE_LENGTH(debug_data) < UINT32_MAX - 4 - 4);
  frame_length = 4 + 4 + (uint32_t)GPR_SLICE_LENGTH(debug_data);

  /* frame header: length */
  *p++ = (uint8_t)(frame_length >> 16);
  *p++ = (uint8_t)(frame_length >> 8);
  *p++ = (uint8_t)(frame_length);
  /* frame header: type */
  *p++ = GRPC_CHTTP2_FRAME_GOAWAY;
  /* frame header: flags */
  *p++ = 0;
  /* frame header: stream id */
  *p++ = 0;
  *p++ = 0;
  *p++ = 0;
  *p++ = 0;
  /* payload: last stream id */
  *p++ = (uint8_t)(last_stream_id >> 24);
  *p++ = (uint8_t)(last_stream_id >> 16);
  *p++ = (uint8_t)(last_stream_id >> 8);
  *p++ = (uint8_t)(last_stream_id);
  /* payload: error code */
  *p++ = (uint8_t)(error_code >> 24);
  *p++ = (uint8_t)(error_code >> 16);
  *p++ = (uint8_t)(error_code >> 8);
  *p++ = (uint8_t)(error_code);
  GPR_ASSERT(p == GPR_SLICE_END_PTR(header));
  gpr_slice_buffer_add(slice_buffer, header);
  gpr_slice_buffer_add(slice_buffer, debug_data);
}

static void test_load_small_file(void) {
  FILE *tmp = NULL;
  gpr_slice slice;
  int success;
  char *tmp_name;
  const char *blah = "blah";

  LOG_TEST_NAME();

  tmp = gpr_tmpfile(prefix, &tmp_name);
  GPR_ASSERT(tmp_name != NULL);
  GPR_ASSERT(tmp != NULL);
  GPR_ASSERT(fwrite(blah, 1, strlen(blah), tmp) == strlen(blah));
  fclose(tmp);

  slice = gpr_load_file(tmp_name, &success);
  GPR_ASSERT(success == 1);
  GPR_ASSERT(GPR_SLICE_LENGTH(slice) == strlen(blah));
  GPR_ASSERT(!memcmp(GPR_SLICE_START_PTR(slice), blah, strlen(blah)));

  remove(tmp_name);
  gpr_free(tmp_name);
  gpr_slice_unref(slice);
}

static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep,
                           gpr_slice_buffer *slices, grpc_closure *cb) {
  unsigned i;
  tsi_result result = TSI_OK;
  secure_endpoint *ep = (secure_endpoint *)secure_ep;
  uint8_t *cur = GPR_SLICE_START_PTR(ep->write_staging_buffer);
  uint8_t *end = GPR_SLICE_END_PTR(ep->write_staging_buffer);

  gpr_slice_buffer_reset_and_unref(&ep->output_buffer);

  if (false && grpc_trace_secure_endpoint) {
    for (i = 0; i < slices->count; i++) {
      char *data =
          gpr_dump_slice(slices->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII);
      gpr_log(GPR_DEBUG, "WRITE %p: %s", ep, data);
      gpr_free(data);
    }
  }

  for (i = 0; i < slices->count; i++) {
    gpr_slice plain = slices->slices[i];
    uint8_t *message_bytes = GPR_SLICE_START_PTR(plain);
    size_t message_size = GPR_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);
      }
    } while (still_pending_size > 0);
    if (cur != GPR_SLICE_START_PTR(ep->write_staging_buffer)) {
      gpr_slice_buffer_add(
          &ep->output_buffer,
          gpr_slice_split_head(
              &ep->write_staging_buffer,
              (size_t)(cur - GPR_SLICE_START_PTR(ep->write_staging_buffer))));
    }
  }

  if (result != TSI_OK) {
    /* TODO(yangg) do different things according to the error type? */
    gpr_slice_buffer_reset_and_unref(&ep->output_buffer);
    grpc_exec_ctx_sched(
        exec_ctx, cb,
        grpc_set_tsi_error_result(GRPC_ERROR_CREATE("Wrap failed"), result),
        NULL);
    return;
  }

  grpc_endpoint_write(exec_ctx, ep->wrapped_ep, &ep->output_buffer, cb);
}

static void on_read(grpc_exec_ctx *exec_ctx, void *user_data,
                    grpc_error *error) {
  unsigned i;
  uint8_t keep_looping = 0;
  tsi_result result = TSI_OK;
  secure_endpoint *ep = (secure_endpoint *)user_data;
  uint8_t *cur = GPR_SLICE_START_PTR(ep->read_staging_buffer);
  uint8_t *end = GPR_SLICE_END_PTR(ep->read_staging_buffer);

  if (error != GRPC_ERROR_NONE) {
    gpr_slice_buffer_reset_and_unref(ep->read_buffer);
    call_read_cb(exec_ctx, ep, GRPC_ERROR_CREATE_REFERENCING(
                                   "Secure read failed", &error, 1));
    return;
  }

  /* TODO(yangg) check error, maybe bail out early */
  for (i = 0; i < ep->source_buffer.count; i++) {
    gpr_slice encrypted = ep->source_buffer.slices[i];
    uint8_t *message_bytes = GPR_SLICE_START_PTR(encrypted);
    size_t message_size = GPR_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 (cur != GPR_SLICE_START_PTR(ep->read_staging_buffer)) {
    gpr_slice_buffer_add(
        ep->read_buffer,
        gpr_slice_split_head(
            &ep->read_staging_buffer,
            (size_t)(cur - GPR_SLICE_START_PTR(ep->read_staging_buffer))));
  }

  /* TODO(yangg) experiment with moving this block after read_cb to see if it
     helps latency */
  gpr_slice_buffer_reset_and_unref(&ep->source_buffer);

  if (result != TSI_OK) {
    gpr_slice_buffer_reset_and_unref(ep->read_buffer);
    call_read_cb(exec_ctx, ep, grpc_set_tsi_error_result(
                                   GRPC_ERROR_CREATE("Unwrap failed"), result));
    return;
  }

  call_read_cb(exec_ctx, ep, GRPC_ERROR_NONE);
}

int grpc_chttp2_unlocking_check_writes(
    grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
    grpc_chttp2_transport_writing *transport_writing) {
  grpc_chttp2_stream_global *stream_global;
  grpc_chttp2_stream_writing *stream_writing;

  GPR_TIMER_BEGIN("grpc_chttp2_unlocking_check_writes", 0);

  /* simple writes are queued to qbuf, and flushed here */
  gpr_slice_buffer_swap(&transport_global->qbuf, &transport_writing->outbuf);
  GPR_ASSERT(transport_global->qbuf.count == 0);

  grpc_chttp2_hpack_compressor_set_max_table_size(
      &transport_writing->hpack_compressor,
      transport_global->settings[GRPC_PEER_SETTINGS]
                                [GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE]);

  if (transport_global->dirtied_local_settings &&
      !transport_global->sent_local_settings) {
    gpr_slice_buffer_add(
        &transport_writing->outbuf,
        grpc_chttp2_settings_create(
            transport_global->settings[GRPC_SENT_SETTINGS],
            transport_global->settings[GRPC_LOCAL_SETTINGS],
            transport_global->force_send_settings, GRPC_CHTTP2_NUM_SETTINGS));
    transport_global->force_send_settings = 0;
    transport_global->dirtied_local_settings = 0;
    transport_global->sent_local_settings = 1;
  }

  GRPC_CHTTP2_FLOW_MOVE_TRANSPORT("write", transport_writing, outgoing_window,
                                  transport_global, outgoing_window);
  if (transport_writing->outgoing_window > 0) {
    while (grpc_chttp2_list_pop_stalled_by_transport(transport_global,
                                                     &stream_global)) {
      grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global,
                                  false, "transport.read_flow_control");
    }
  }

  /* for each grpc_chttp2_stream that's become writable, frame it's data
     (according to available window sizes) and add to the output buffer */
  while (grpc_chttp2_list_pop_writable_stream(
      transport_global, transport_writing, &stream_global, &stream_writing)) {
    bool sent_initial_metadata = stream_writing->sent_initial_metadata;
    bool become_writable = false;

    stream_writing->id = stream_global->id;
    stream_writing->read_closed = stream_global->read_closed;

    GRPC_CHTTP2_FLOW_MOVE_STREAM("write", transport_writing, stream_writing,
                                 outgoing_window, stream_global,
                                 outgoing_window);

    if (!sent_initial_metadata && stream_global->send_initial_metadata) {
      stream_writing->send_initial_metadata =
          stream_global->send_initial_metadata;
      stream_global->send_initial_metadata = NULL;
      become_writable = true;
      sent_initial_metadata = true;
    }
    if (sent_initial_metadata) {
      if (stream_global->send_message != NULL) {
        gpr_slice hdr = gpr_slice_malloc(5);
        uint8_t *p = GPR_SLICE_START_PTR(hdr);
        uint32_t len = stream_global->send_message->length;
        GPR_ASSERT(stream_writing->send_message == NULL);
        p[0] = (stream_global->send_message->flags &
                GRPC_WRITE_INTERNAL_COMPRESS) != 0;
        p[1] = (uint8_t)(len >> 24);
        p[2] = (uint8_t)(len >> 16);
        p[3] = (uint8_t)(len >> 8);
        p[4] = (uint8_t)(len);
        gpr_slice_buffer_add(&stream_writing->flow_controlled_buffer, hdr);
        if (stream_global->send_message->length > 0) {
          stream_writing->send_message = stream_global->send_message;
        } else {
          stream_writing->send_message = NULL;
        }
        stream_writing->stream_fetched = 0;
        stream_global->send_message = NULL;
      }
      if ((stream_writing->send_message != NULL ||
           stream_writing->flow_controlled_buffer.length > 0) &&
          stream_writing->outgoing_window > 0) {
        if (transport_writing->outgoing_window > 0) {
          become_writable = true;
        } else {
          grpc_chttp2_list_add_stalled_by_transport(transport_writing,
                                                    stream_writing);
        }
      }
      if (stream_global->send_trailing_metadata) {
        stream_writing->send_trailing_metadata =
            stream_global->send_trailing_metadata;
        stream_global->send_trailing_metadata = NULL;
        become_writable = true;
      }
    }

//.........这里部分代码省略.........

/* encode an mdelem */
static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem,
                      framer_state *st) {
  uint32_t key_hash = elem->key->hash;
  uint32_t elem_hash = GRPC_MDSTR_KV_HASH(key_hash, elem->value->hash);
  size_t decoder_space_usage;
  uint32_t indices_key;
  int should_add_elem;

  GPR_ASSERT(GPR_SLICE_LENGTH(elem->key->slice) > 0);
  if (GPR_SLICE_START_PTR(elem->key->slice)[0] != ':') { /* regular header */
    st->seen_regular_header = 1;
  } else {
    GPR_ASSERT(
        st->seen_regular_header == 0 &&
        "Reserved header (colon-prefixed) happening after regular ones.");
  }

  inc_filter(HASH_FRAGMENT_1(elem_hash), &c->filter_elems_sum, c->filter_elems);

  /* is this elem currently in the decoders table? */

  if (c->entries_elems[HASH_FRAGMENT_2(elem_hash)] == elem &&
      c->indices_elems[HASH_FRAGMENT_2(elem_hash)] > c->tail_remote_index) {
    /* HIT: complete element (first cuckoo hash) */
    emit_indexed(c, dynidx(c, c->indices_elems[HASH_FRAGMENT_2(elem_hash)]),
                 st);
    return;
  }

  if (c->entries_elems[HASH_FRAGMENT_3(elem_hash)] == elem &&
      c->indices_elems[HASH_FRAGMENT_3(elem_hash)] > c->tail_remote_index) {
    /* HIT: complete element (second cuckoo hash) */
    emit_indexed(c, dynidx(c, c->indices_elems[HASH_FRAGMENT_3(elem_hash)]),
                 st);
    return;
  }

  /* should this elem be in the table? */
  decoder_space_usage = grpc_mdelem_get_size_in_hpack_table(elem);
  should_add_elem = decoder_space_usage < MAX_DECODER_SPACE_USAGE &&
                    c->filter_elems[HASH_FRAGMENT_1(elem_hash)] >=
                        c->filter_elems_sum / ONE_ON_ADD_PROBABILITY;

  /* no hits for the elem... maybe there's a key? */

  indices_key = c->indices_keys[HASH_FRAGMENT_2(key_hash)];
  if (c->entries_keys[HASH_FRAGMENT_2(key_hash)] == elem->key &&
      indices_key > c->tail_remote_index) {
    /* HIT: key (first cuckoo hash) */
    if (should_add_elem) {
      emit_lithdr_incidx(c, dynidx(c, indices_key), elem, st);
      add_elem(c, elem);
      return;
    } else {
      emit_lithdr_noidx(c, dynidx(c, indices_key), elem, st);
      return;
    }
    GPR_UNREACHABLE_CODE(return );
  }

  indices_key = c->indices_keys[HASH_FRAGMENT_3(key_hash)];
  if (c->entries_keys[HASH_FRAGMENT_3(key_hash)] == elem->key &&
      indices_key > c->tail_remote_index) {
    /* HIT: key (first cuckoo hash) */
    if (should_add_elem) {
      emit_lithdr_incidx(c, dynidx(c, indices_key), elem, st);
      add_elem(c, elem);
      return;
    } else {
      emit_lithdr_noidx(c, dynidx(c, indices_key), elem, st);
      return;
    }
    GPR_UNREACHABLE_CODE(return );
  }

  /* no elem, key in the table... fall back to literal emission */

  if (should_add_elem) {
    emit_lithdr_incidx_v(c, elem, st);
    add_elem(c, elem);
    return;
  } else {
    emit_lithdr_noidx_v(c, elem, st);
    return;
  }
  GPR_UNREACHABLE_CODE(return );
}

static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair(
    size_t slice_size, gpr_slice *leftover_slices, size_t leftover_nslices) {
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  tsi_frame_protector *fake_read_protector = tsi_create_fake_protector(NULL);
  tsi_frame_protector *fake_write_protector = tsi_create_fake_protector(NULL);
  grpc_endpoint_test_fixture f;
  grpc_endpoint_pair tcp;

  tcp = grpc_iomgr_create_endpoint_pair("fixture", slice_size);
  grpc_endpoint_add_to_pollset(&exec_ctx, tcp.client, g_pollset);
  grpc_endpoint_add_to_pollset(&exec_ctx, tcp.server, g_pollset);

  if (leftover_nslices == 0) {
    f.client_ep =
        grpc_secure_endpoint_create(fake_read_protector, tcp.client, NULL, 0);
  } else {
    unsigned i;
    tsi_result result;
    size_t still_pending_size;
    size_t total_buffer_size = 8192;
    size_t buffer_size = total_buffer_size;
    uint8_t *encrypted_buffer = gpr_malloc(buffer_size);
    uint8_t *cur = encrypted_buffer;
    gpr_slice encrypted_leftover;
    for (i = 0; i < leftover_nslices; i++) {
      gpr_slice plain = leftover_slices[i];
      uint8_t *message_bytes = GPR_SLICE_START_PTR(plain);
      size_t message_size = GPR_SLICE_LENGTH(plain);
      while (message_size > 0) {
        size_t protected_buffer_size_to_send = buffer_size;
        size_t processed_message_size = message_size;
        result = tsi_frame_protector_protect(
            fake_write_protector, message_bytes, &processed_message_size, cur,
            &protected_buffer_size_to_send);
        GPR_ASSERT(result == TSI_OK);
        message_bytes += processed_message_size;
        message_size -= processed_message_size;
        cur += protected_buffer_size_to_send;
        GPR_ASSERT(buffer_size >= protected_buffer_size_to_send);
        buffer_size -= protected_buffer_size_to_send;
      }
      gpr_slice_unref(plain);
    }
    do {
      size_t protected_buffer_size_to_send = buffer_size;
      result = tsi_frame_protector_protect_flush(fake_write_protector, cur,
                                                 &protected_buffer_size_to_send,
                                                 &still_pending_size);
      GPR_ASSERT(result == TSI_OK);
      cur += protected_buffer_size_to_send;
      GPR_ASSERT(buffer_size >= protected_buffer_size_to_send);
      buffer_size -= protected_buffer_size_to_send;
    } while (still_pending_size > 0);
    encrypted_leftover = gpr_slice_from_copied_buffer(
        (const char *)encrypted_buffer, total_buffer_size - buffer_size);
    f.client_ep = grpc_secure_endpoint_create(fake_read_protector, tcp.client,
                                              &encrypted_leftover, 1);
    gpr_slice_unref(encrypted_leftover);
    gpr_free(encrypted_buffer);
  }

  f.server_ep =
      grpc_secure_endpoint_create(fake_write_protector, tcp.server, NULL, 0);
  grpc_exec_ctx_finish(&exec_ctx);
  return f;
}

grpc_mdstr *grpc_mdstr_from_slice(grpc_mdctx *ctx, gpr_slice slice) {
  grpc_mdstr *result = grpc_mdstr_from_buffer(ctx, GPR_SLICE_START_PTR(slice),
                                              GPR_SLICE_LENGTH(slice));
  gpr_slice_unref(slice);
  return result;
}

static flush_result tcp_flush(grpc_tcp *tcp) {
  struct msghdr msg;
  struct iovec iov[MAX_WRITE_IOVEC];
  msg_iovlen_type iov_size;
  ssize_t sent_length;
  size_t sending_length;
  size_t trailing;
  size_t unwind_slice_idx;
  size_t unwind_byte_idx;

  for (;;) {
    sending_length = 0;
    unwind_slice_idx = tcp->outgoing_slice_idx;
    unwind_byte_idx = tcp->outgoing_byte_idx;
    for (iov_size = 0; tcp->outgoing_slice_idx != tcp->outgoing_buffer->count &&
                           iov_size != MAX_WRITE_IOVEC;
         iov_size++) {
      iov[iov_size].iov_base =
          GPR_SLICE_START_PTR(
              tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) +
          tcp->outgoing_byte_idx;
      iov[iov_size].iov_len =
          GPR_SLICE_LENGTH(
              tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) -
          tcp->outgoing_byte_idx;
      sending_length += iov[iov_size].iov_len;
      tcp->outgoing_slice_idx++;
      tcp->outgoing_byte_idx = 0;
    }
    GPR_ASSERT(iov_size > 0);

    msg.msg_name = NULL;
    msg.msg_namelen = 0;
    msg.msg_iov = iov;
    msg.msg_iovlen = iov_size;
    msg.msg_control = NULL;
    msg.msg_controllen = 0;
    msg.msg_flags = 0;

    GPR_TIMER_BEGIN("sendmsg", 1);
    do {
      /* TODO(klempner): Cork if this is a partial write */
      sent_length = sendmsg(tcp->fd, &msg, SENDMSG_FLAGS);
    } while (sent_length < 0 && errno == EINTR);
    GPR_TIMER_END("sendmsg", 0);

    if (sent_length < 0) {
      if (errno == EAGAIN) {
        tcp->outgoing_slice_idx = unwind_slice_idx;
        tcp->outgoing_byte_idx = unwind_byte_idx;
        return FLUSH_PENDING;
      } else {
        /* TODO(klempner): Log some of these */
        return FLUSH_ERROR;
      }
    }

    GPR_ASSERT(tcp->outgoing_byte_idx == 0);
    trailing = sending_length - (size_t)sent_length;
    while (trailing > 0) {
      size_t slice_length;

      tcp->outgoing_slice_idx--;
      slice_length = GPR_SLICE_LENGTH(
          tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]);
      if (slice_length > trailing) {
        tcp->outgoing_byte_idx = slice_length - trailing;
        break;
      } else {
        trailing -= slice_length;
      }
    }

    if (tcp->outgoing_slice_idx == tcp->outgoing_buffer->count) {
      return FLUSH_DONE;
    }
  };
}

grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
    grpc_exec_ctx *exec_ctx, void *parser,
    grpc_chttp2_transport_parsing *transport_parsing,
    grpc_chttp2_stream_parsing *stream_parsing, gpr_slice slice, int is_last) {
  uint8_t *const beg = GPR_SLICE_START_PTR(slice);
  uint8_t *const end = GPR_SLICE_END_PTR(slice);
  uint8_t *cur = beg;
  grpc_chttp2_data_parser *p = parser;
  uint32_t message_flags;
  grpc_chttp2_incoming_byte_stream *incoming_byte_stream;

  if (is_last && p->is_last_frame) {
    stream_parsing->received_close = 1;
  }

  if (cur == end) {
    return GRPC_CHTTP2_PARSE_OK;
  }

  switch (p->state) {
  fh_0:
    case GRPC_CHTTP2_DATA_FH_0:
      p->frame_type = *cur;
      switch (p->frame_type) {
        case 0:
          p->is_frame_compressed = 0; /* GPR_FALSE */
          break;
        case 1:
          p->is_frame_compressed = 1; /* GPR_TRUE */
          break;
        default:
          gpr_log(GPR_ERROR, "Bad GRPC frame type 0x%02x", p->frame_type);
          return GRPC_CHTTP2_STREAM_ERROR;
      }
      if (++cur == end) {
        p->state = GRPC_CHTTP2_DATA_FH_1;
        return GRPC_CHTTP2_PARSE_OK;
      }
    /* fallthrough */
    case GRPC_CHTTP2_DATA_FH_1:
      p->frame_size = ((uint32_t)*cur) << 24;
      if (++cur == end) {
        p->state = GRPC_CHTTP2_DATA_FH_2;
        return GRPC_CHTTP2_PARSE_OK;
      }
    /* fallthrough */
    case GRPC_CHTTP2_DATA_FH_2:
      p->frame_size |= ((uint32_t)*cur) << 16;
      if (++cur == end) {
        p->state = GRPC_CHTTP2_DATA_FH_3;
        return GRPC_CHTTP2_PARSE_OK;
      }
    /* fallthrough */
    case GRPC_CHTTP2_DATA_FH_3:
      p->frame_size |= ((uint32_t)*cur) << 8;
      if (++cur == end) {
        p->state = GRPC_CHTTP2_DATA_FH_4;
        return GRPC_CHTTP2_PARSE_OK;
      }
    /* fallthrough */
    case GRPC_CHTTP2_DATA_FH_4:
      p->frame_size |= ((uint32_t)*cur);
      p->state = GRPC_CHTTP2_DATA_FRAME;
      ++cur;
      message_flags = 0;
      if (p->is_frame_compressed) {
        message_flags |= GRPC_WRITE_INTERNAL_COMPRESS;
      }
      p->parsing_frame = incoming_byte_stream =
          grpc_chttp2_incoming_byte_stream_create(
              exec_ctx, transport_parsing, stream_parsing, p->frame_size,
              message_flags, &p->incoming_frames);
    /* fallthrough */
    case GRPC_CHTTP2_DATA_FRAME:
      if (cur == end) {
        grpc_chttp2_list_add_parsing_seen_stream(transport_parsing,
                                                 stream_parsing);
        return GRPC_CHTTP2_PARSE_OK;
      }
      grpc_chttp2_list_add_parsing_seen_stream(transport_parsing,
                                               stream_parsing);
      if ((uint32_t)(end - cur) == p->frame_size) {
        grpc_chttp2_incoming_byte_stream_push(
            exec_ctx, p->parsing_frame,
            gpr_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg)));
        grpc_chttp2_incoming_byte_stream_finished(exec_ctx, p->parsing_frame, 1,
                                                  1);
        p->parsing_frame = NULL;
        p->state = GRPC_CHTTP2_DATA_FH_0;
        return GRPC_CHTTP2_PARSE_OK;
      } else if ((uint32_t)(end - cur) > p->frame_size) {
        grpc_chttp2_incoming_byte_stream_push(
            exec_ctx, p->parsing_frame,
            gpr_slice_sub(slice, (size_t)(cur - beg),
                          (size_t)(cur + p->frame_size - beg)));
        grpc_chttp2_incoming_byte_stream_finished(exec_ctx, p->parsing_frame, 1,
                                                  1);
        p->parsing_frame = NULL;
        cur += p->frame_size;
        goto fh_0; /* loop */
//.........这里部分代码省略.........

/* Initiates a write. */
static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
                                            gpr_slice *slices, size_t nslices,
                                            grpc_endpoint_write_cb cb,
                                            void *arg) {
  grpc_tcp *tcp = (grpc_tcp *) ep;
  grpc_winsocket *socket = tcp->socket;
  grpc_winsocket_callback_info *info = &socket->write_info;
  unsigned i;
  DWORD bytes_sent;
  int status;
  WSABUF local_buffers[16];
  WSABUF *allocated = NULL;
  WSABUF *buffers = local_buffers;

  GPR_ASSERT(!tcp->outstanding_write);
  GPR_ASSERT(!tcp->shutting_down);
  tcp_ref(tcp);

  tcp->outstanding_write = 1;
  tcp->write_cb = cb;
  tcp->write_user_data = arg;

  gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices);

  if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) {
    buffers = (WSABUF *) gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count);
    allocated = buffers;
  }

  for (i = 0; i < tcp->write_slices.count; i++) {
    buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices.slices[i]);
    buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]);
  }

  /* First, let's try a synchronous, non-blocking write. */
  status = WSASend(socket->socket, buffers, tcp->write_slices.count,
                   &bytes_sent, 0, NULL, NULL);
  info->wsa_error = status == 0 ? 0 : WSAGetLastError();

  /* We would kind of expect to get a WSAEWOULDBLOCK here, especially on a busy
     connection that has its send queue filled up. But if we don't, then we can
     avoid doing an async write operation at all. */
  if (info->wsa_error != WSAEWOULDBLOCK) {
    grpc_endpoint_write_status ret = GRPC_ENDPOINT_WRITE_ERROR;
    if (status == 0) {
      ret = GRPC_ENDPOINT_WRITE_DONE;
      GPR_ASSERT(bytes_sent == tcp->write_slices.length);
    } else {
      char *utf8_message = gpr_format_message(info->wsa_error);
      gpr_log(GPR_ERROR, "WSASend error: %s", utf8_message);
      gpr_free(utf8_message);
    }
    if (allocated) gpr_free(allocated);
    gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
    tcp->outstanding_write = 0;
    tcp_unref(tcp);
    return ret;
  }

  /* If we got a WSAEWOULDBLOCK earlier, then we need to re-do the same
     operation, this time asynchronously. */
  memset(&socket->write_info.overlapped, 0, sizeof(OVERLAPPED));
  status = WSASend(socket->socket, buffers, tcp->write_slices.count,
                   &bytes_sent, 0, &socket->write_info.overlapped, NULL);
  if (allocated) gpr_free(allocated);

  /* It is possible the operation completed then. But we'd still get an IOCP
     notification. So let's ignore it and wait for the IOCP. */
  if (status != 0) {
    int error = WSAGetLastError();
    if (error != WSA_IO_PENDING) {
      char *utf8_message = gpr_format_message(WSAGetLastError());
      gpr_log(GPR_ERROR, "WSASend error: %s - this means we're going to leak.",
              utf8_message);
      gpr_free(utf8_message);
    /* I'm pretty sure this is a very bad situation there. Hence the log.
       What will happen now is that the socket will neither wait for read
       or write, unless the caller retry, which is unlikely, but I am not
       sure if that's guaranteed. And there might also be a read pending.
       This means that the future orphanage of that socket will be in limbo,
       and we're going to leak it. I have no idea what could cause this
       specific case however, aside from a parameter error from our call.
       Normal read errors would actually happen during the overlapped
       operation, which is the supported way to go for that. */
      tcp->outstanding_write = 0;
      tcp_unref(tcp);
      /* Per the comment above, I'm going to treat that case as a hard failure
         for now, and leave the option to catch that and debug. */
      __debugbreak();
      return GRPC_ENDPOINT_WRITE_ERROR;
    }
  }

  /* As all is now setup, we can now ask for the IOCP notification. It may
     trigger the callback immediately however, but no matter. */
  grpc_socket_notify_on_write(socket, on_write, tcp);
  return GRPC_ENDPOINT_WRITE_PENDING;
}

/* Initiates a write. */
static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
                                            gpr_slice *slices, size_t nslices,
                                            grpc_endpoint_write_cb cb,
                                            void *arg) {
  grpc_tcp *tcp = (grpc_tcp *) ep;
  grpc_winsocket *socket = tcp->socket;
  grpc_winsocket_callback_info *info = &socket->write_info;
  unsigned i;
  DWORD bytes_sent;
  int status;
  WSABUF local_buffers[16];
  WSABUF *allocated = NULL;
  WSABUF *buffers = local_buffers;

  GPR_ASSERT(!tcp->socket->write_info.outstanding);
  if (tcp->shutting_down) {
    return GRPC_ENDPOINT_WRITE_ERROR;
  }
  tcp_ref(tcp);

  tcp->socket->write_info.outstanding = 1;
  tcp->write_cb = cb;
  tcp->write_user_data = arg;

  gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices);

  if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) {
    buffers = (WSABUF *) gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count);
    allocated = buffers;
  }

  for (i = 0; i < tcp->write_slices.count; i++) {
    buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices.slices[i]);
    buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]);
  }

  /* First, let's try a synchronous, non-blocking write. */
  status = WSASend(socket->socket, buffers, tcp->write_slices.count,
                   &bytes_sent, 0, NULL, NULL);
  info->wsa_error = status == 0 ? 0 : WSAGetLastError();

  /* We would kind of expect to get a WSAEWOULDBLOCK here, especially on a busy
     connection that has its send queue filled up. But if we don't, then we can
     avoid doing an async write operation at all. */
  if (info->wsa_error != WSAEWOULDBLOCK) {
    grpc_endpoint_write_status ret = GRPC_ENDPOINT_WRITE_ERROR;
    if (status == 0) {
      ret = GRPC_ENDPOINT_WRITE_DONE;
      GPR_ASSERT(bytes_sent == tcp->write_slices.length);
    } else {
      char *utf8_message = gpr_format_message(info->wsa_error);
      gpr_log(GPR_ERROR, "WSASend error: %s", utf8_message);
      gpr_free(utf8_message);
    }
    if (allocated) gpr_free(allocated);
    gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
    tcp->socket->write_info.outstanding = 0;
    tcp_unref(tcp);
    return ret;
  }

  /* If we got a WSAEWOULDBLOCK earlier, then we need to re-do the same
     operation, this time asynchronously. */
  memset(&socket->write_info.overlapped, 0, sizeof(OVERLAPPED));
  status = WSASend(socket->socket, buffers, tcp->write_slices.count,
                   &bytes_sent, 0, &socket->write_info.overlapped, NULL);
  if (allocated) gpr_free(allocated);

  if (status != 0) {
    int wsa_error = WSAGetLastError();
    if (wsa_error != WSA_IO_PENDING) {
      gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
      tcp->socket->write_info.outstanding = 0;
      tcp_unref(tcp);
      return GRPC_ENDPOINT_WRITE_ERROR;
    }
  }

  /* As all is now setup, we can now ask for the IOCP notification. It may
     trigger the callback immediately however, but no matter. */
  grpc_socket_notify_on_write(socket, on_write, tcp);
  return GRPC_ENDPOINT_WRITE_PENDING;
}

const char *grpc_mdstr_as_c_string(grpc_mdstr *s) {
  return (const char *)GPR_SLICE_START_PTR(s->slice);
}