自学教程：C++ GRPC_ERROR_CREATE函数代码示例

static void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {  round_robin_lb_policy *p = (round_robin_lb_policy *)pol;  pending_pick *pp;  size_t i;  gpr_mu_lock(&p->mu);  if (grpc_lb_round_robin_trace) {    gpr_log(GPR_DEBUG, "Shutting down Round Robin policy at %p", (void *)pol);  }  p->shutdown = 1;  while ((pp = p->pending_picks)) {    p->pending_picks = pp->next;    *pp->target = NULL;    grpc_exec_ctx_sched(exec_ctx, pp->on_complete,                        GRPC_ERROR_CREATE("Channel Shutdown"), NULL);    gpr_free(pp);  }  grpc_connectivity_state_set(      exec_ctx, &p->state_tracker, GRPC_CHANNEL_SHUTDOWN,      GRPC_ERROR_CREATE("Channel Shutdown"), "rr_shutdown");  for (i = 0; i < p->num_subchannels; i++) {    subchannel_data *sd = p->subchannels[i];    grpc_subchannel_notify_on_state_change(exec_ctx, sd->subchannel, NULL, NULL,                                           &sd->connectivity_changed_closure);  }  gpr_mu_unlock(&p->mu);}

static grpc_error *add_socket_to_server(grpc_tcp_server *s, uv_tcp_t *handle,                                        const grpc_resolved_address *addr,                                        unsigned port_index,                                        grpc_tcp_listener **listener) {  grpc_tcp_listener *sp = NULL;  int port = -1;  int status;  grpc_error *error;  grpc_resolved_address sockname_temp;  // The last argument to uv_tcp_bind is flags  status = uv_tcp_bind(handle, (struct sockaddr *)addr->addr, 0);  if (status != 0) {    error = GRPC_ERROR_CREATE("Failed to bind to port");    error =        grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, uv_strerror(status));    return error;  }  status = uv_listen((uv_stream_t *)handle, SOMAXCONN, on_connect);  if (status != 0) {    error = GRPC_ERROR_CREATE("Failed to listen to port");    error =        grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, uv_strerror(status));    return error;  }  sockname_temp.len = (int)sizeof(struct sockaddr_storage);  status = uv_tcp_getsockname(handle, (struct sockaddr *)&sockname_temp.addr,                              (int *)&sockname_temp.len);  if (status != 0) {    error = GRPC_ERROR_CREATE("getsockname failed");    error =        grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, uv_strerror(status));    return error;  }  port = grpc_sockaddr_get_port(&sockname_temp);  GPR_ASSERT(port >= 0);  GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server");  sp = gpr_malloc(sizeof(grpc_tcp_listener));  sp->next = NULL;  if (s->head == NULL) {    s->head = sp;  } else {    s->tail->next = sp;  }  s->tail = sp;  sp->server = s;  sp->handle = handle;  sp->port = port;  sp->port_index = port_index;  handle->data = sp;  s->open_ports++;  GPR_ASSERT(sp->handle);  *listener = sp;  return GRPC_ERROR_NONE;}

static grpc_error *update_incoming_window(    grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing,    grpc_chttp2_stream_parsing *stream_parsing) {  uint32_t incoming_frame_size = transport_parsing->incoming_frame_size;  if (incoming_frame_size > transport_parsing->incoming_window) {    char *msg;    gpr_asprintf(&msg, "frame of size %d overflows incoming window of %" PRId64,                 transport_parsing->incoming_frame_size,                 transport_parsing->incoming_window);    grpc_error *err = GRPC_ERROR_CREATE(msg);    gpr_free(msg);    return err;  }  if (incoming_frame_size > stream_parsing->incoming_window) {    char *msg;    gpr_asprintf(&msg, "frame of size %d overflows incoming window of %" PRId64,                 transport_parsing->incoming_frame_size,                 stream_parsing->incoming_window);    grpc_error *err = GRPC_ERROR_CREATE(msg);    gpr_free(msg);    return err;  }  GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", transport_parsing, incoming_window,                                   incoming_frame_size);  GRPC_CHTTP2_FLOW_DEBIT_STREAM("parse", transport_parsing, stream_parsing,                                incoming_window, incoming_frame_size);  stream_parsing->received_bytes += incoming_frame_size;  grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing);  return GRPC_ERROR_NONE;}

/* Asynchronous callback from the IOCP, or the background thread. */static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, grpc_error *error) {  grpc_tcp *tcp = tcpp;  grpc_closure *cb = tcp->read_cb;  grpc_winsocket *socket = tcp->socket;  gpr_slice sub;  grpc_winsocket_callback_info *info = &socket->read_info;  GRPC_ERROR_REF(error);  if (error == GRPC_ERROR_NONE) {    if (info->wsa_error != 0 && !tcp->shutting_down) {      char *utf8_message = gpr_format_message(info->wsa_error);      error = GRPC_ERROR_CREATE(utf8_message);      gpr_free(utf8_message);      gpr_slice_unref(tcp->read_slice);    } else {      if (info->bytes_transfered != 0 && !tcp->shutting_down) {        sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);        gpr_slice_buffer_add(tcp->read_slices, sub);      } else {        gpr_slice_unref(tcp->read_slice);        error = GRPC_ERROR_CREATE("End of TCP stream");      }    }  }  tcp->read_cb = NULL;  TCP_UNREF(tcp, "read");  grpc_exec_ctx_sched(exec_ctx, cb, error, NULL);}

static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,                                          grpc_server *server, size_t cq_idx,                                          requested_call *rc) {  call_data *calld = NULL;  request_matcher *rm = NULL;  int request_id;  if (gpr_atm_acq_load(&server->shutdown_flag)) {    fail_call(exec_ctx, server, cq_idx, rc,              GRPC_ERROR_CREATE("Server Shutdown"));    return GRPC_CALL_OK;  }  request_id = gpr_stack_lockfree_pop(server->request_freelist_per_cq[cq_idx]);  if (request_id == -1) {    /* out of request ids: just fail this one */    fail_call(exec_ctx, server, cq_idx, rc,              grpc_error_set_int(GRPC_ERROR_CREATE("Out of request ids"),                                 GRPC_ERROR_INT_LIMIT,                                 server->max_requested_calls_per_cq));    return GRPC_CALL_OK;  }  switch (rc->type) {    case BATCH_CALL:      rm = &server->unregistered_request_matcher;      break;    case REGISTERED_CALL:      rm = &rc->data.registered.registered_method->request_matcher;      break;  }  server->requested_calls_per_cq[cq_idx][request_id] = *rc;  gpr_free(rc);  if (gpr_stack_lockfree_push(rm->requests_per_cq[cq_idx], request_id)) {    /* this was the first queued request: we need to lock and start       matching calls */    gpr_mu_lock(&server->mu_call);    while ((calld = rm->pending_head) != NULL) {      request_id = gpr_stack_lockfree_pop(rm->requests_per_cq[cq_idx]);      if (request_id == -1) break;      rm->pending_head = calld->pending_next;      gpr_mu_unlock(&server->mu_call);      gpr_mu_lock(&calld->mu_state);      if (calld->state == ZOMBIED) {        gpr_mu_unlock(&calld->mu_state);        grpc_closure_init(            &calld->kill_zombie_closure, kill_zombie,            grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));        grpc_exec_ctx_sched(exec_ctx, &calld->kill_zombie_closure,                            GRPC_ERROR_NONE, NULL);      } else {        GPR_ASSERT(calld->state == PENDING);        calld->state = ACTIVATED;        gpr_mu_unlock(&calld->mu_state);        publish_call(exec_ctx, server, calld, cq_idx,                     &server->requested_calls_per_cq[cq_idx][request_id]);      }      gpr_mu_lock(&server->mu_call);    }    gpr_mu_unlock(&server->mu_call);  }  return GRPC_CALL_OK;}

static grpc_error *try_split_host_port(const char *name,                                       const char *default_port, char **host,                                       char **port) {  /* parse name, splitting it into host and port parts */  grpc_error *error;  gpr_split_host_port(name, host, port);  if (host == NULL) {    char *msg;    gpr_asprintf(&msg, "unparseable host:port: '%s'", name);    error = GRPC_ERROR_CREATE(msg);    gpr_free(msg);    return error;  }  if (port == NULL) {    if (default_port == NULL) {      char *msg;      gpr_asprintf(&msg, "no port in name '%s'", name);      error = GRPC_ERROR_CREATE(msg);      gpr_free(msg);      return error;    }    *port = gpr_strdup(default_port);  }  return GRPC_ERROR_NONE;}

static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *user_data,                            grpc_security_status status,                            grpc_auth_context *auth_context) {    grpc_security_handshake *h = user_data;    tsi_frame_protector *protector;    tsi_result result;    if (status != GRPC_SECURITY_OK) {        security_handshake_done(            exec_ctx, h,            grpc_error_set_int(GRPC_ERROR_CREATE("Error checking peer."),                               GRPC_ERROR_INT_SECURITY_STATUS, status));        return;    }    h->auth_context = GRPC_AUTH_CONTEXT_REF(auth_context, "handshake");    result =        tsi_handshaker_create_frame_protector(h->handshaker, NULL, &protector);    if (result != TSI_OK) {        security_handshake_done(            exec_ctx, h,            grpc_set_tsi_error_result(                GRPC_ERROR_CREATE("Frame protector creation failed"), result));        return;    }    h->secure_endpoint =        grpc_secure_endpoint_create(protector, h->wrapped_endpoint,                                    h->left_overs.slices, h->left_overs.count);    h->left_overs.count = 0;    h->left_overs.length = 0;    security_handshake_done(exec_ctx, h, GRPC_ERROR_NONE);    return;}

static void test_leftover(grpc_endpoint_test_config config, size_t slice_size) {  grpc_endpoint_test_fixture f = config.create_fixture(slice_size);  grpc_slice_buffer incoming;  grpc_slice s =      grpc_slice_from_copied_string("hello world 12345678900987654321");  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  int n = 0;  grpc_closure done_closure;  gpr_log(GPR_INFO, "Start test left over");  grpc_slice_buffer_init(&incoming);  grpc_closure_init(&done_closure, inc_call_ctr, &n, grpc_schedule_on_exec_ctx);  grpc_endpoint_read(&exec_ctx, f.client_ep, &incoming, &done_closure);  grpc_exec_ctx_finish(&exec_ctx);  GPR_ASSERT(n == 1);  GPR_ASSERT(incoming.count == 1);  GPR_ASSERT(grpc_slice_eq(s, incoming.slices[0]));  grpc_endpoint_shutdown(&exec_ctx, f.client_ep,                         GRPC_ERROR_CREATE("test_leftover end"));  grpc_endpoint_shutdown(&exec_ctx, f.server_ep,                         GRPC_ERROR_CREATE("test_leftover end"));  grpc_endpoint_destroy(&exec_ctx, f.client_ep);  grpc_endpoint_destroy(&exec_ctx, f.server_ep);  grpc_exec_ctx_finish(&exec_ctx);  grpc_slice_unref_internal(&exec_ctx, s);  grpc_slice_buffer_destroy_internal(&exec_ctx, &incoming);  clean_up();}

static grpc_error *init_frame_parser(grpc_exec_ctx *exec_ctx,                                     grpc_chttp2_transport *t) {  if (t->is_first_frame &&      t->incoming_frame_type != GRPC_CHTTP2_FRAME_SETTINGS) {    char *msg;    gpr_asprintf(        &msg, "Expected SETTINGS frame as the first frame, got frame type %d",        t->incoming_frame_type);    grpc_error *err = GRPC_ERROR_CREATE(msg);    gpr_free(msg);    return err;  }  t->is_first_frame = false;  if (t->expect_continuation_stream_id != 0) {    if (t->incoming_frame_type != GRPC_CHTTP2_FRAME_CONTINUATION) {      char *msg;      gpr_asprintf(&msg, "Expected CONTINUATION frame, got frame type %02x",                   t->incoming_frame_type);      grpc_error *err = GRPC_ERROR_CREATE(msg);      gpr_free(msg);      return err;    }    if (t->expect_continuation_stream_id != t->incoming_stream_id) {      char *msg;      gpr_asprintf(          &msg,          "Expected CONTINUATION frame for grpc_chttp2_stream %08x, got "          "grpc_chttp2_stream %08x",          t->expect_continuation_stream_id, t->incoming_stream_id);      grpc_error *err = GRPC_ERROR_CREATE(msg);      gpr_free(msg);      return err;    }    return init_header_frame_parser(exec_ctx, t, 1);  }  switch (t->incoming_frame_type) {    case GRPC_CHTTP2_FRAME_DATA:      return init_data_frame_parser(exec_ctx, t);    case GRPC_CHTTP2_FRAME_HEADER:      return init_header_frame_parser(exec_ctx, t, 0);    case GRPC_CHTTP2_FRAME_CONTINUATION:      return GRPC_ERROR_CREATE("Unexpected CONTINUATION frame");    case GRPC_CHTTP2_FRAME_RST_STREAM:      return init_rst_stream_parser(exec_ctx, t);    case GRPC_CHTTP2_FRAME_SETTINGS:      return init_settings_frame_parser(exec_ctx, t);    case GRPC_CHTTP2_FRAME_WINDOW_UPDATE:      return init_window_update_frame_parser(exec_ctx, t);    case GRPC_CHTTP2_FRAME_PING:      return init_ping_parser(exec_ctx, t);    case GRPC_CHTTP2_FRAME_GOAWAY:      return init_goaway_parser(exec_ctx, t);    default:      if (grpc_http_trace) {        gpr_log(GPR_ERROR, "Unknown frame type %02x", t->incoming_frame_type);      }      return init_skip_frame_parser(exec_ctx, t, 0);  }}

int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,                                      grpc_server_credentials *creds) {  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  grpc_error *err = GRPC_ERROR_NONE;  grpc_server_security_connector *sc = NULL;  int port_num = 0;  GRPC_API_TRACE(      "grpc_server_add_secure_http2_port("      "server=%p, addr=%s, creds=%p)",      3, (server, addr, creds));  // Create security context.  if (creds == NULL) {    err = GRPC_ERROR_CREATE(        "No credentials specified for secure server port (creds==NULL)");    goto done;  }  grpc_security_status status =      grpc_server_credentials_create_security_connector(&exec_ctx, creds, &sc);  if (status != GRPC_SECURITY_OK) {    char *msg;    gpr_asprintf(&msg,                 "Unable to create secure server with credentials of type %s.",                 creds->type);    err = grpc_error_set_int(GRPC_ERROR_CREATE(msg),                             GRPC_ERROR_INT_SECURITY_STATUS, status);    gpr_free(msg);    goto done;  }  // Create channel args.  grpc_arg args_to_add[2];  args_to_add[0] = grpc_server_credentials_to_arg(creds);  args_to_add[1] = grpc_security_connector_to_arg(&sc->base);  grpc_channel_args *args =      grpc_channel_args_copy_and_add(grpc_server_get_channel_args(server),                                     args_to_add, GPR_ARRAY_SIZE(args_to_add));  // Add server port.  err = grpc_chttp2_server_add_port(&exec_ctx, server, addr, args, &port_num);done:  if (sc != NULL) {    GRPC_SECURITY_CONNECTOR_UNREF(&exec_ctx, &sc->base, "server");  }  grpc_exec_ctx_finish(&exec_ctx);  if (err != GRPC_ERROR_NONE) {    const char *msg = grpc_error_string(err);    gpr_log(GPR_ERROR, "%s", msg);    grpc_error_free_string(msg);    GRPC_ERROR_UNREF(err);  }  return port_num;}

grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx,                                   grpc_closure *shutdown_complete,                                   const grpc_channel_args *args,                                   grpc_tcp_server **server) {  gpr_once_init(&check_init, init);  grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server));  s->so_reuseport = has_so_reuseport;  s->resource_quota = grpc_resource_quota_create(NULL);  for (size_t i = 0; i < (args == NULL ? 0 : args->num_args); i++) {    if (0 == strcmp(GRPC_ARG_ALLOW_REUSEPORT, args->args[i].key)) {      if (args->args[i].type == GRPC_ARG_INTEGER) {        s->so_reuseport =            has_so_reuseport && (args->args[i].value.integer != 0);      } else {        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);        gpr_free(s);        return GRPC_ERROR_CREATE(GRPC_ARG_ALLOW_REUSEPORT                                 " must be an integer");      }    } else if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) {      if (args->args[i].type == GRPC_ARG_POINTER) {        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);        s->resource_quota =            grpc_resource_quota_internal_ref(args->args[i].value.pointer.p);      } else {        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);        gpr_free(s);        return GRPC_ERROR_CREATE(GRPC_ARG_RESOURCE_QUOTA                                 " must be a pointer to a buffer pool");      }    }  }  gpr_ref_init(&s->refs, 1);  gpr_mu_init(&s->mu);  s->active_ports = 0;  s->destroyed_ports = 0;  s->shutdown = false;  s->shutdown_starting.head = NULL;  s->shutdown_starting.tail = NULL;  s->shutdown_complete = shutdown_complete;  s->on_accept_cb = NULL;  s->on_accept_cb_arg = NULL;  s->head = NULL;  s->tail = NULL;  s->nports = 0;  gpr_atm_no_barrier_store(&s->next_pollset_to_assign, 0);  *server = s;  return GRPC_ERROR_NONE;}

grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx,                                   grpc_closure *shutdown_complete,                                   const grpc_channel_args *args,                                   grpc_tcp_server **server) {  grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server));  s->resource_quota = grpc_resource_quota_create(NULL);  for (size_t i = 0; i < (args == NULL ? 0 : args->num_args); i++) {    if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) {      if (args->args[i].type == GRPC_ARG_POINTER) {        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);        s->resource_quota =            grpc_resource_quota_internal_ref(args->args[i].value.pointer.p);      } else {        grpc_resource_quota_internal_unref(exec_ctx, s->resource_quota);        gpr_free(s);        return GRPC_ERROR_CREATE(GRPC_ARG_RESOURCE_QUOTA                                 " must be a pointer to a buffer pool");      }    }  }  gpr_ref_init(&s->refs, 1);  s->on_accept_cb = NULL;  s->on_accept_cb_arg = NULL;  s->open_ports = 0;  s->head = NULL;  s->tail = NULL;  s->shutdown_starting.head = NULL;  s->shutdown_starting.tail = NULL;  s->shutdown_complete = shutdown_complete;  *server = s;  return GRPC_ERROR_NONE;}

static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,                                  grpc_call_element *elem,                                  grpc_call_element_args *args) {    return grpc_error_set_int(GRPC_ERROR_CREATE("access denied"),                              GRPC_ERROR_INT_GRPC_STATUS,                              GRPC_STATUS_PERMISSION_DENIED);}

static grpc_error *handle_addrinfo_result(int status, struct addrinfo *result,                                          grpc_resolved_addresses **addresses) {  struct addrinfo *resp;  size_t i;  if (status != 0) {    grpc_error *error;    *addresses = NULL;    error = GRPC_ERROR_CREATE("getaddrinfo failed");    error =        grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, uv_strerror(status));    return error;  }  (*addresses) = gpr_malloc(sizeof(grpc_resolved_addresses));  (*addresses)->naddrs = 0;  for (resp = result; resp != NULL; resp = resp->ai_next) {    (*addresses)->naddrs++;  }  (*addresses)->addrs =      gpr_malloc(sizeof(grpc_resolved_address) * (*addresses)->naddrs);  i = 0;  for (resp = result; resp != NULL; resp = resp->ai_next) {    memcpy(&(*addresses)->addrs[i].addr, resp->ai_addr, resp->ai_addrlen);    (*addresses)->addrs[i].len = resp->ai_addrlen;    i++;  }  {    for (i = 0; i < (*addresses)->naddrs; i++) {      char *buf;      grpc_sockaddr_to_string(&buf, &(*addresses)->addrs[i], 0);      gpr_free(buf);    }  }  return GRPC_ERROR_NONE;}

static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,                                     grpc_security_status status,                                     grpc_endpoint *secure_endpoint,                                     grpc_auth_context *auth_context) {    connector *c = arg;    gpr_mu_lock(&c->mu);    grpc_error *error = GRPC_ERROR_NONE;    if (c->connecting_endpoint == NULL) {        memset(c->result, 0, sizeof(*c->result));        gpr_mu_unlock(&c->mu);    } else if (status != GRPC_SECURITY_OK) {        error = grpc_error_set_int(GRPC_ERROR_CREATE("Secure handshake failed"),                                   GRPC_ERROR_INT_SECURITY_STATUS, status);        memset(c->result, 0, sizeof(*c->result));        c->connecting_endpoint = NULL;        gpr_mu_unlock(&c->mu);    } else {        grpc_arg auth_context_arg;        c->connecting_endpoint = NULL;        gpr_mu_unlock(&c->mu);        c->result->transport = grpc_create_chttp2_transport(                                   exec_ctx, c->args.channel_args, secure_endpoint, 1);        grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, NULL);        auth_context_arg = grpc_auth_context_to_arg(auth_context);        c->result->channel_args =            grpc_channel_args_copy_and_add(c->tmp_args, &auth_context_arg, 1);    }    grpc_closure *notify = c->notify;    c->notify = NULL;    grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);}

static grpc_error *send_handshake_bytes_to_peer_locked(grpc_exec_ctx *exec_ctx,                                                       security_handshaker *h) {  // Get data to send.  tsi_result result = TSI_OK;  size_t offset = 0;  do {    size_t to_send_size = h->handshake_buffer_size - offset;    result = tsi_handshaker_get_bytes_to_send_to_peer(        h->handshaker, h->handshake_buffer + offset, &to_send_size);    offset += to_send_size;    if (result == TSI_INCOMPLETE_DATA) {      h->handshake_buffer_size *= 2;      h->handshake_buffer =          gpr_realloc(h->handshake_buffer, h->handshake_buffer_size);    }  } while (result == TSI_INCOMPLETE_DATA);  if (result != TSI_OK) {    return grpc_set_tsi_error_result(GRPC_ERROR_CREATE("Handshake failed"),                                     result);  }  // Send data.  grpc_slice to_send =      grpc_slice_from_copied_buffer((const char *)h->handshake_buffer, offset);  grpc_slice_buffer_reset_and_unref(&h->outgoing);  grpc_slice_buffer_add(&h->outgoing, to_send);  grpc_endpoint_write(exec_ctx, h->args->endpoint, &h->outgoing,                      &h->on_handshake_data_sent_to_peer);  return GRPC_ERROR_NONE;}

static void uv_tc_on_connect(uv_connect_t *req, int status) {  grpc_uv_tcp_connect *connect = req->data;  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  grpc_error *error = GRPC_ERROR_NONE;  int done;  grpc_closure *closure = connect->closure;  grpc_timer_cancel(&exec_ctx, &connect->alarm);  if (status == 0) {    *connect->endpoint = grpc_tcp_create(        connect->tcp_handle, connect->resource_quota, connect->addr_name);  } else {    error = GRPC_ERROR_CREATE("Failed to connect to remote host");    error = grpc_error_set_int(error, GRPC_ERROR_INT_ERRNO, -status);    error =        grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, uv_strerror(status));    if (status == UV_ECANCELED) {      error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR,                                 "Timeout occurred");      // This should only happen if the handle is already closed    } else {      error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR,                                 uv_strerror(status));      uv_close((uv_handle_t *)connect->tcp_handle, tcp_close_callback);    }  }  done = (--connect->refs == 0);  if (done) {    uv_tcp_connect_cleanup(&exec_ctx, connect);  }  grpc_closure_sched(&exec_ctx, closure, error);  grpc_exec_ctx_finish(&exec_ctx);}

static void on_initial_header(grpc_exec_ctx *exec_ctx, void *tp,                              grpc_mdelem *md) {  grpc_chttp2_transport *t = tp;  grpc_chttp2_stream *s = t->incoming_stream;  GPR_TIMER_BEGIN("on_initial_header", 0);  GPR_ASSERT(s != NULL);  GRPC_CHTTP2_IF_TRACING(gpr_log(      GPR_INFO, "HTTP:%d:HDR:%s: %s: %s", s->id, t->is_client ? "CLI" : "SVR",      grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value)));  if (md->key == GRPC_MDSTR_GRPC_STATUS && md != GRPC_MDELEM_GRPC_STATUS_0) {    /* TODO(ctiller): check for a status like " 0" */    s->seen_error = true;  }  if (md->key == GRPC_MDSTR_GRPC_TIMEOUT) {    gpr_timespec *cached_timeout = grpc_mdelem_get_user_data(md, free_timeout);    if (!cached_timeout) {      /* not already parsed: parse it now, and store the result away */      cached_timeout = gpr_malloc(sizeof(gpr_timespec));      if (!grpc_http2_decode_timeout(grpc_mdstr_as_c_string(md->value),                                     cached_timeout)) {        gpr_log(GPR_ERROR, "Ignoring bad timeout value '%s'",                grpc_mdstr_as_c_string(md->value));        *cached_timeout = gpr_inf_future(GPR_TIMESPAN);      }      grpc_mdelem_set_user_data(md, free_timeout, cached_timeout);    }    grpc_chttp2_incoming_metadata_buffer_set_deadline(        &s->metadata_buffer[0],        gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), *cached_timeout));    GRPC_MDELEM_UNREF(md);  } else {    const size_t new_size = s->metadata_buffer[0].size + GRPC_MDELEM_LENGTH(md);    const size_t metadata_size_limit =        t->settings[GRPC_ACKED_SETTINGS]                   [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];    if (new_size > metadata_size_limit) {      gpr_log(GPR_DEBUG,              "received initial metadata size exceeds limit (%" PRIuPTR              " vs. %" PRIuPTR ")",              new_size, metadata_size_limit);      grpc_chttp2_cancel_stream(          exec_ctx, t, s,          grpc_error_set_int(              GRPC_ERROR_CREATE("received initial metadata size exceeds limit"),              GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED));      grpc_chttp2_parsing_become_skip_parser(exec_ctx, t);      s->seen_error = true;      GRPC_MDELEM_UNREF(md);    } else {      grpc_chttp2_incoming_metadata_buffer_add(&s->metadata_buffer[0], md);    }  }  GPR_TIMER_END("on_initial_header", 0);}

static grpc_error *init_settings_frame_parser(grpc_exec_ctx *exec_ctx,                                              grpc_chttp2_transport *t) {  if (t->incoming_stream_id != 0) {    return GRPC_ERROR_CREATE("Settings frame received for grpc_chttp2_stream");  }  grpc_error *err = grpc_chttp2_settings_parser_begin_frame(      &t->simple.settings, t->incoming_frame_size, t->incoming_frame_flags,      t->settings[GRPC_PEER_SETTINGS]);  if (err != GRPC_ERROR_NONE) {    return err;  }  if (t->incoming_frame_flags & GRPC_CHTTP2_FLAG_ACK) {    memcpy(t->settings[GRPC_ACKED_SETTINGS], t->settings[GRPC_SENT_SETTINGS],           GRPC_CHTTP2_NUM_SETTINGS * sizeof(uint32_t));    grpc_chttp2_hptbl_set_max_bytes(        &t->hpack_parser.table,        t->settings[GRPC_ACKED_SETTINGS]                   [GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE]);    t->sent_local_settings = 0;  }  t->parser = grpc_chttp2_settings_parser_parse;  t->parser_data = &t->simple.settings;  return GRPC_ERROR_NONE;}

static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,                                  grpc_server *server) {  size_t i;  if (!gpr_atm_acq_load(&server->shutdown_flag) || server->shutdown_published) {    return;  }  kill_pending_work_locked(exec_ctx, server,                           GRPC_ERROR_CREATE("Server Shutdown"));  if (server->root_channel_data.next != &server->root_channel_data ||      server->listeners_destroyed < num_listeners(server)) {    if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME),                                  server->last_shutdown_message_time),                     gpr_time_from_seconds(1, GPR_TIMESPAN)) >= 0) {      server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME);      gpr_log(GPR_DEBUG,              "Waiting for %d channels and %d/%d listeners to be destroyed"              " before shutting down server",              num_channels(server),              num_listeners(server) - server->listeners_destroyed,              num_listeners(server));    }    return;  }  server->shutdown_published = 1;  for (i = 0; i < server->num_shutdown_tags; i++) {    server_ref(server);    grpc_cq_end_op(exec_ctx, server->shutdown_tags[i].cq,                   server->shutdown_tags[i].tag, GRPC_ERROR_NONE,                   done_shutdown_event, server,                   &server->shutdown_tags[i].completion);  }}

// If the handshake failed or we're shutting down, clean up and invoke the// callback with the error.static void handshake_failed_locked(grpc_exec_ctx* exec_ctx,                                    http_connect_handshaker* handshaker,                                    grpc_error* error) {  if (error == GRPC_ERROR_NONE) {    // If we were shut down after an endpoint operation succeeded but    // before the endpoint callback was invoked, we need to generate our    // own error.    error = GRPC_ERROR_CREATE("Handshaker shutdown");  }  if (!handshaker->shutdown) {    // TODO(ctiller): It is currently necessary to shutdown endpoints    // before destroying them, even if we know that there are no    // pending read/write callbacks.  This should be fixed, at which    // point this can be removed.    grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint);    // Not shutting down, so the handshake failed.  Clean up before    // invoking the callback.    cleanup_args_for_failure_locked(handshaker);    // Set shutdown to true so that subsequent calls to    // http_connect_handshaker_shutdown() do nothing.    handshaker->shutdown = true;  }  // Invoke callback.  grpc_exec_ctx_sched(exec_ctx, handshaker->on_handshake_done, error, NULL);}

static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {  chttp2_connector *c = arg;  gpr_mu_lock(&c->mu);  GPR_ASSERT(c->connecting);  c->connecting = false;  if (error != GRPC_ERROR_NONE || c->shutdown) {    if (error == GRPC_ERROR_NONE) {      error = GRPC_ERROR_CREATE("connector shutdown");    } else {      error = GRPC_ERROR_REF(error);    }    memset(c->result, 0, sizeof(*c->result));    grpc_closure *notify = c->notify;    c->notify = NULL;    grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);    if (c->endpoint != NULL) grpc_endpoint_shutdown(exec_ctx, c->endpoint);    gpr_mu_unlock(&c->mu);    chttp2_connector_unref(exec_ctx, arg);  } else {    GPR_ASSERT(c->endpoint != NULL);    if (!GRPC_SLICE_IS_EMPTY(c->args.initial_connect_string)) {      grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,                        c);      grpc_slice_buffer_init(&c->initial_string_buffer);      grpc_slice_buffer_add(&c->initial_string_buffer,                            c->args.initial_connect_string);      grpc_endpoint_write(exec_ctx, c->endpoint, &c->initial_string_buffer,                          &c->initial_string_sent);    } else {      start_handshake_locked(exec_ctx, c);    }    gpr_mu_unlock(&c->mu);  }}

static void add_error(grpc_error **combined, grpc_error *error) {  if (error == GRPC_ERROR_NONE) return;  if (*combined == GRPC_ERROR_NONE) {    *combined = GRPC_ERROR_CREATE("Client auth metadata plugin error");  }  *combined = grpc_error_add_child(*combined, error);}

static void send_handshake_bytes_to_peer(grpc_exec_ctx *exec_ctx,        grpc_security_handshake *h) {    size_t offset = 0;    tsi_result result = TSI_OK;    gpr_slice to_send;    do {        size_t to_send_size = h->handshake_buffer_size - offset;        result = tsi_handshaker_get_bytes_to_send_to_peer(                     h->handshaker, h->handshake_buffer + offset, &to_send_size);        offset += to_send_size;        if (result == TSI_INCOMPLETE_DATA) {            h->handshake_buffer_size *= 2;            h->handshake_buffer =                gpr_realloc(h->handshake_buffer, h->handshake_buffer_size);        }    } while (result == TSI_INCOMPLETE_DATA);    if (result != TSI_OK) {        security_handshake_done(exec_ctx, h,                                grpc_set_tsi_error_result(                                    GRPC_ERROR_CREATE("Handshake failed"), result));        return;    }    to_send =        gpr_slice_from_copied_buffer((const char *)h->handshake_buffer, offset);    gpr_slice_buffer_reset_and_unref(&h->outgoing);    gpr_slice_buffer_add(&h->outgoing, to_send);    /* TODO(klempner,jboeuf): This should probably use the client setup       deadline */    grpc_endpoint_write(exec_ctx, h->wrapped_endpoint, &h->outgoing,                        &h->on_handshake_data_sent_to_peer);}

// If the handshake failed or we're shutting down, clean up and invoke the// callback with the error.static void security_handshake_failed_locked(grpc_exec_ctx *exec_ctx,                                             security_handshaker *h,                                             grpc_error *error) {  if (error == GRPC_ERROR_NONE) {    // If we were shut down after the handshake succeeded but before an    // endpoint callback was invoked, we need to generate our own error.    error = GRPC_ERROR_CREATE("Handshaker shutdown");  }  const char *msg = grpc_error_string(error);  gpr_log(GPR_DEBUG, "Security handshake failed: %s", msg);  grpc_error_free_string(msg);  if (!h->shutdown) {    // TODO(ctiller): It is currently necessary to shutdown endpoints    // before destroying them, even if we know that there are no    // pending read/write callbacks.  This should be fixed, at which    // point this can be removed.    grpc_endpoint_shutdown(exec_ctx, h->args->endpoint);    // Not shutting down, so the write failed.  Clean up before    // invoking the callback.    cleanup_args_for_failure_locked(h);    // Set shutdown to true so that subsequent calls to    // security_handshaker_shutdown() do nothing.    h->shutdown = true;  }  // Invoke callback.  grpc_exec_ctx_sched(exec_ctx, h->on_handshake_done, error, NULL);}

static grpc_error *init_settings_frame_parser(    grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) {  if (transport_parsing->incoming_stream_id != 0) {    return GRPC_ERROR_CREATE("Settings frame received for grpc_chttp2_stream");  }  grpc_error *err = grpc_chttp2_settings_parser_begin_frame(      &transport_parsing->simple.settings,      transport_parsing->incoming_frame_size,      transport_parsing->incoming_frame_flags, transport_parsing->settings);  if (err != GRPC_ERROR_NONE) {    return err;  }  if (transport_parsing->incoming_frame_flags & GRPC_CHTTP2_FLAG_ACK) {    transport_parsing->settings_ack_received = 1;    grpc_chttp2_hptbl_set_max_bytes(        &transport_parsing->hpack_parser.table,        transport_parsing            ->last_sent_settings[GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE]);    transport_parsing->max_frame_size =        transport_parsing            ->last_sent_settings[GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE];  }  transport_parsing->parser = grpc_chttp2_settings_parser_parse;  transport_parsing->parser_data = &transport_parsing->simple.settings;  return GRPC_ERROR_NONE;}

// Callback to read the HTTP CONNECT request.// TODO(roth): Technically, for any of the failure modes handled by this// function, we should handle the error by returning an HTTP response to// the client indicating that the request failed.  However, for the purposes// of this test code, it's fine to pretend this is a client-side error,// which will cause the client connection to be dropped.static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,                                 grpc_error* error) {  proxy_connection* conn = arg;  if (error != GRPC_ERROR_NONE) {    proxy_connection_failed(exec_ctx, conn, true /* is_client */,                            "HTTP proxy read request", error);    return;  }  // Read request and feed it to the parser.  for (size_t i = 0; i < conn->client_read_buffer.count; ++i) {    if (GRPC_SLICE_LENGTH(conn->client_read_buffer.slices[i]) > 0) {      error = grpc_http_parser_parse(&conn->http_parser,                                     conn->client_read_buffer.slices[i], NULL);      if (error != GRPC_ERROR_NONE) {        proxy_connection_failed(exec_ctx, conn, true /* is_client */,                                "HTTP proxy request parse", error);        GRPC_ERROR_UNREF(error);        return;      }    }  }  grpc_slice_buffer_reset_and_unref(&conn->client_read_buffer);  // If we're not done reading the request, read more data.  if (conn->http_parser.state != GRPC_HTTP_BODY) {    grpc_endpoint_read(exec_ctx, conn->client_endpoint,                       &conn->client_read_buffer, &conn->on_read_request_done);    return;  }  // Make sure we got a CONNECT request.  if (strcmp(conn->http_request.method, "CONNECT") != 0) {    char* msg;    gpr_asprintf(&msg, "HTTP proxy got request method %s",                 conn->http_request.method);    error = GRPC_ERROR_CREATE(msg);    gpr_free(msg);    proxy_connection_failed(exec_ctx, conn, true /* is_client */,                            "HTTP proxy read request", error);    GRPC_ERROR_UNREF(error);    return;  }  // Resolve address.  grpc_resolved_addresses* resolved_addresses = NULL;  error = grpc_blocking_resolve_address(conn->http_request.path, "80",                                        &resolved_addresses);  if (error != GRPC_ERROR_NONE) {    proxy_connection_failed(exec_ctx, conn, true /* is_client */,                            "HTTP proxy DNS lookup", error);    GRPC_ERROR_UNREF(error);    return;  }  GPR_ASSERT(resolved_addresses->naddrs >= 1);  // Connect to requested address.  // The connection callback inherits our reference to conn.  const gpr_timespec deadline = gpr_time_add(      gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(10, GPR_TIMESPAN));  grpc_tcp_client_connect(exec_ctx, &conn->on_server_connect_done,                          &conn->server_endpoint, conn->pollset_set, NULL,                          &resolved_addresses->addrs[0], deadline);  grpc_resolved_addresses_destroy(resolved_addresses);}

static void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {  pick_first_lb_policy *p = (pick_first_lb_policy *)pol;  pending_pick *pp;  grpc_connected_subchannel *selected;  gpr_mu_lock(&p->mu);  selected = GET_SELECTED(p);  p->shutdown = 1;  pp = p->pending_picks;  p->pending_picks = NULL;  grpc_connectivity_state_set(      exec_ctx, &p->state_tracker, GRPC_CHANNEL_SHUTDOWN,      GRPC_ERROR_CREATE("Channel shutdown"), "shutdown");  /* cancel subscription */  if (selected != NULL) {    grpc_connected_subchannel_notify_on_state_change(        exec_ctx, selected, NULL, NULL, &p->connectivity_changed);  } else if (p->num_subchannels > 0) {    grpc_subchannel_notify_on_state_change(        exec_ctx, p->subchannels[p->checking_subchannel], NULL, NULL,        &p->connectivity_changed);  }  gpr_mu_unlock(&p->mu);  while (pp != NULL) {    pending_pick *next = pp->next;    *pp->target = NULL;    grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_NONE, NULL);    gpr_free(pp);    pp = next;  }}

static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,                                  grpc_channel_element *elem,                                  grpc_transport_op *op) {  channel_data *chand = elem->channel_data;  grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL);  GPR_ASSERT(op->set_accept_stream == false);  if (op->bind_pollset != NULL) {    grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties,                                 op->bind_pollset);  }  gpr_mu_lock(&chand->mu);  if (op->on_connectivity_state_change != NULL) {    grpc_connectivity_state_notify_on_state_change(        exec_ctx, &chand->state_tracker, op->connectivity_state,        op->on_connectivity_state_change);    op->on_connectivity_state_change = NULL;    op->connectivity_state = NULL;  }  if (op->send_ping != NULL) {    if (chand->lb_policy == NULL) {      grpc_exec_ctx_sched(exec_ctx, op->send_ping,                          GRPC_ERROR_CREATE("Ping with no load balancing"),                          NULL);    } else {      grpc_lb_policy_ping_one(exec_ctx, chand->lb_policy, op->send_ping);      op->bind_pollset = NULL;    }    op->send_ping = NULL;  }  if (op->disconnect_with_error != GRPC_ERROR_NONE) {    if (chand->resolver != NULL) {      set_channel_connectivity_state_locked(          exec_ctx, chand, GRPC_CHANNEL_SHUTDOWN,          GRPC_ERROR_REF(op->disconnect_with_error), "disconnect");      grpc_resolver_shutdown(exec_ctx, chand->resolver);      GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel");      chand->resolver = NULL;      if (!chand->started_resolving) {        grpc_closure_list_fail_all(&chand->waiting_for_config_closures,                                   GRPC_ERROR_REF(op->disconnect_with_error));        grpc_exec_ctx_enqueue_list(exec_ctx,                                   &chand->waiting_for_config_closures, NULL);      }      if (chand->lb_policy != NULL) {        grpc_pollset_set_del_pollset_set(exec_ctx,                                         chand->lb_policy->interested_parties,                                         chand->interested_parties);        GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel");        chand->lb_policy = NULL;      }    }    GRPC_ERROR_UNREF(op->disconnect_with_error);  }  gpr_mu_unlock(&chand->mu);}

void grpc_chttp2_data_parser_destroy(grpc_exec_ctx *exec_ctx,                                     grpc_chttp2_data_parser *parser) {  if (parser->parsing_frame != NULL) {    grpc_chttp2_incoming_byte_stream_finished(        exec_ctx, parser->parsing_frame, GRPC_ERROR_CREATE("Parser destroyed"));  }  GRPC_ERROR_UNREF(parser->error);}