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

static void shutdown_server(grpc_end2end_test_fixture *f) {  if (!f->server) return;  grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000));  GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(1000),                                         GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5),                                         NULL).type == GRPC_OP_COMPLETE);  grpc_server_destroy(f->server);  f->server = NULL;}

static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,                                 grpc_error *error) {  size_t i;  int port = 0;  portreq *pr = arg;  int failed = 0;  grpc_httpcli_response *response = &pr->response;  if (error != GRPC_ERROR_NONE) {    failed = 1;    const char *msg = grpc_error_string(error);    gpr_log(GPR_DEBUG, "failed port pick from server: retrying [%s]", msg);    grpc_error_free_string(msg);  } else if (response->status != 200) {    failed = 1;    gpr_log(GPR_DEBUG, "failed port pick from server: status=%d",            response->status);  }  if (failed) {    grpc_httpcli_request req;    memset(&req, 0, sizeof(req));    GPR_ASSERT(pr->retries < 10);    gpr_sleep_until(gpr_time_add(        gpr_now(GPR_CLOCK_REALTIME),        gpr_time_from_millis(            (int64_t)(1000.0 * (1 + pow(1.3, pr->retries) * rand() / RAND_MAX)),            GPR_TIMESPAN)));    pr->retries++;    req.host = pr->server;    req.http.path = "/get";    grpc_http_response_destroy(&pr->response);    memset(&pr->response, 0, sizeof(pr->response));    grpc_resource_quota *resource_quota =        grpc_resource_quota_create("port_server_client/pick_retry");    grpc_httpcli_get(exec_ctx, pr->ctx, &pr->pops, resource_quota, &req,                     GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),                     grpc_closure_create(got_port_from_server, pr),                     &pr->response);    grpc_resource_quota_internal_unref(exec_ctx, resource_quota);    return;  }  GPR_ASSERT(response);  GPR_ASSERT(response->status == 200);  for (i = 0; i < response->body_length; i++) {    GPR_ASSERT(response->body[i] >= '0' && response->body[i] <= '9');    port = port * 10 + response->body[i] - '0';  }  GPR_ASSERT(port > 1024);  gpr_mu_lock(pr->mu);  pr->port = port;  GRPC_LOG_IF_ERROR(      "pollset_kick",      grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), NULL));  gpr_mu_unlock(pr->mu);}

int main(int argc, char **argv) {  grpc_channel *chan;  grpc_call *call;  grpc_completion_queue *cq;  cq_verifier *cqv;  grpc_op ops[6];  grpc_op *op;  grpc_metadata_array trailing_metadata_recv;  grpc_status_code status;  char *details = NULL;  size_t details_capacity = 0;  grpc_test_init(argc, argv);  grpc_init();  grpc_metadata_array_init(&trailing_metadata_recv);  chan = grpc_lame_client_channel_create("lampoon:national");  GPR_ASSERT(chan);  cq = grpc_completion_queue_create();  call = grpc_channel_create_call(chan, cq, "/Foo", "anywhere",                                  GRPC_TIMEOUT_SECONDS_TO_DEADLINE(100));  GPR_ASSERT(call);  cqv = cq_verifier_create(cq);  op = ops;  op->op = GRPC_OP_SEND_INITIAL_METADATA;  op->data.send_initial_metadata.count = 0;  op->flags = 0;  op++;  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;  op->data.recv_status_on_client.status = &status;  op->data.recv_status_on_client.status_details = &details;  op->data.recv_status_on_client.status_details_capacity = &details_capacity;  op->flags = 0;  op++;  GPR_ASSERT(GRPC_CALL_OK ==             grpc_call_start_batch(call, ops, op - ops, tag(1)));  /* the call should immediately fail */  cq_expect_completion(cqv, tag(1), 1);  cq_verify(cqv);  grpc_call_destroy(call);  grpc_channel_destroy(chan);  cq_verifier_destroy(cqv);  grpc_completion_queue_destroy(cq);  grpc_metadata_array_destroy(&trailing_metadata_recv);  gpr_free(details);  grpc_shutdown();  return 0;}

static int pick_port_using_server(char *server) {  grpc_httpcli_context context;  grpc_httpcli_request req;  portreq pr;  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  grpc_closure shutdown_closure;  grpc_init();  memset(&pr, 0, sizeof(pr));  memset(&req, 0, sizeof(req));  pr.pollset = gpr_malloc(grpc_pollset_size());  grpc_pollset_init(pr.pollset, &pr.mu);  grpc_closure_init(&shutdown_closure, destroy_pollset_and_shutdown,                    pr.pollset);  pr.port = -1;  pr.server = server;  pr.ctx = &context;  req.host = server;  req.path = "/get";  grpc_httpcli_context_init(&context);  grpc_httpcli_get(&exec_ctx, &context, pr.pollset, &req,                   GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), got_port_from_server,                   &pr);  grpc_exec_ctx_finish(&exec_ctx);  gpr_mu_lock(pr.mu);  while (pr.port == -1) {    grpc_pollset_worker *worker = NULL;    grpc_pollset_work(&exec_ctx, pr.pollset, &worker,                      gpr_now(GPR_CLOCK_MONOTONIC),                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1));  }  gpr_mu_unlock(pr.mu);  grpc_httpcli_context_destroy(&context);  grpc_pollset_shutdown(&exec_ctx, pr.pollset, &shutdown_closure);  grpc_exec_ctx_finish(&exec_ctx);  gpr_free(pr.pollset);  return pr.port;}

void test_succeeds(void) {  struct sockaddr_in addr;  socklen_t addr_len = sizeof(addr);  int svr_fd;  int r;  int connections_complete_before;  grpc_closure done;  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  gpr_log(GPR_DEBUG, "test_succeeds");  memset(&addr, 0, sizeof(addr));  addr.sin_family = AF_INET;  /* create a dummy server */  svr_fd = socket(AF_INET, SOCK_STREAM, 0);  GPR_ASSERT(svr_fd >= 0);  GPR_ASSERT(0 == bind(svr_fd, (struct sockaddr *)&addr, addr_len));  GPR_ASSERT(0 == listen(svr_fd, 1));  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  connections_complete_before = g_connections_complete;  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  /* connect to it */  GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0);  grpc_closure_init(&done, must_succeed, NULL);  grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, &g_pollset_set,                          (struct sockaddr *)&addr, addr_len,                          gpr_inf_future(GPR_CLOCK_REALTIME));  /* await the connection */  do {    addr_len = sizeof(addr);    r = accept(svr_fd, (struct sockaddr *)&addr, &addr_len);  } while (r == -1 && errno == EINTR);  GPR_ASSERT(r >= 0);  close(r);  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  while (g_connections_complete == connections_complete_before) {    grpc_pollset_worker *worker = NULL;    grpc_pollset_work(&exec_ctx, &g_pollset, &worker,                      gpr_now(GPR_CLOCK_MONOTONIC),                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5));    gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));    grpc_exec_ctx_flush(&exec_ctx);    gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  }  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  grpc_exec_ctx_finish(&exec_ctx);}

static void prepare_test(int is_client) {  int port;  char *server_hostport;  grpc_op *op;  g_state.is_client = is_client;  grpc_metadata_array_init(&g_state.initial_metadata_recv);  grpc_metadata_array_init(&g_state.trailing_metadata_recv);  g_state.deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(2);  g_state.cq = grpc_completion_queue_create(NULL);  g_state.cqv = cq_verifier_create(g_state.cq);  g_state.details = NULL;  g_state.details_capacity = 0;  if (is_client) {    /* create a call, channel to a non existant server */    g_state.chan =        grpc_insecure_channel_create("nonexistant:54321", NULL, NULL);    g_state.call = grpc_channel_create_call(        g_state.chan, NULL, GRPC_PROPAGATE_DEFAULTS, g_state.cq, "/Foo",        "nonexistant", g_state.deadline, NULL);  } else {    g_state.server = grpc_server_create(NULL, NULL);    grpc_server_register_completion_queue(g_state.server, g_state.cq, NULL);    port = grpc_pick_unused_port_or_die();    gpr_join_host_port(&server_hostport, "0.0.0.0", port);    grpc_server_add_insecure_http2_port(g_state.server, server_hostport);    grpc_server_start(g_state.server);    gpr_free(server_hostport);    gpr_join_host_port(&server_hostport, "localhost", port);    g_state.chan = grpc_insecure_channel_create(server_hostport, NULL, NULL);    gpr_free(server_hostport);    g_state.call = grpc_channel_create_call(        g_state.chan, NULL, GRPC_PROPAGATE_DEFAULTS, g_state.cq, "/Foo", "bar",        g_state.deadline, NULL);    grpc_metadata_array_init(&g_state.server_initial_metadata_recv);    grpc_call_details_init(&g_state.call_details);    op = g_state.ops;    op->op = GRPC_OP_SEND_INITIAL_METADATA;    op->data.send_initial_metadata.count = 0;    op->flags = 0;    op->reserved = NULL;    op++;    GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(g_state.call, g_state.ops,                                                     (size_t)(op - g_state.ops),                                                     tag(1), NULL));    GPR_ASSERT(GRPC_CALL_OK ==               grpc_server_request_call(g_state.server, &g_state.server_call,                                        &g_state.call_details,                                        &g_state.server_initial_metadata_recv,                                        g_state.cq, g_state.cq, tag(101)));    cq_expect_completion(g_state.cqv, tag(101), 1);    cq_expect_completion(g_state.cqv, tag(1), 1);    cq_verify(g_state.cqv);  }}

static void free_port_using_server(char *server, int port) {  grpc_httpcli_context context;  grpc_httpcli_request req;  freereq pr;  char *path;  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  grpc_closure shutdown_closure;  grpc_init();  memset(&pr, 0, sizeof(pr));  memset(&req, 0, sizeof(req));  pr.pollset = gpr_malloc(grpc_pollset_size());  grpc_pollset_init(pr.pollset, &pr.mu);  grpc_closure_init(&shutdown_closure, destroy_pollset_and_shutdown,                    pr.pollset);  req.host = server;  gpr_asprintf(&path, "/drop/%d", port);  req.path = path;  grpc_httpcli_context_init(&context);  grpc_httpcli_get(&exec_ctx, &context, pr.pollset, &req,                   GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), freed_port_from_server,                   &pr);  gpr_mu_lock(pr.mu);  while (!pr.done) {    grpc_pollset_worker *worker = NULL;    grpc_pollset_work(&exec_ctx, pr.pollset, &worker,                      gpr_now(GPR_CLOCK_MONOTONIC),                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1));  }  gpr_mu_unlock(pr.mu);  grpc_httpcli_context_destroy(&context);  grpc_exec_ctx_finish(&exec_ctx);  grpc_pollset_shutdown(&exec_ctx, pr.pollset, &shutdown_closure);  grpc_exec_ctx_finish(&exec_ctx);  gpr_free(pr.pollset);  gpr_free(path);}

/* Do both reading and writing using the grpc_endpoint API.   This also includes a test of the shutdown behavior. */static void read_and_write_test(grpc_endpoint_test_config config,                                size_t num_bytes, size_t write_size,                                size_t slice_size, int shutdown) {    struct read_and_write_test_state state;    gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);    grpc_endpoint_test_fixture f =        begin_test(config, "read_and_write_test", slice_size);    if (shutdown) {        gpr_log(GPR_INFO, "Start read and write shutdown test");    } else {        gpr_log(GPR_INFO, "Start read and write test with %d bytes, slice size %d",                num_bytes, slice_size);    }    state.read_ep = f.client_ep;    state.write_ep = f.server_ep;    state.target_bytes = num_bytes;    state.bytes_read = 0;    state.current_write_size = write_size;    state.bytes_written = 0;    state.read_done = 0;    state.write_done = 0;    state.current_read_data = 0;    state.current_write_data = 0;    /* Get started by pretending an initial write completed */    /* NOTE: Sets up initial conditions so we can have the same write handler       for the first iteration as for later iterations. It does the right thing       even when bytes_written is unsigned. */    state.bytes_written -= state.current_write_size;    read_and_write_test_write_handler(&state, GRPC_ENDPOINT_CB_OK);    grpc_endpoint_notify_on_read(state.read_ep, read_and_write_test_read_handler,                                 &state);    if (shutdown) {        gpr_log(GPR_DEBUG, "shutdown read");        grpc_endpoint_shutdown(state.read_ep);        gpr_log(GPR_DEBUG, "shutdown write");        grpc_endpoint_shutdown(state.write_ep);    }    gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));    while (!state.read_done || !state.write_done) {        GPR_ASSERT(gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), deadline) < 0);        grpc_pollset_work(g_pollset, deadline);    }    gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));    grpc_endpoint_destroy(state.read_ep);    grpc_endpoint_destroy(state.write_ep);    end_test(config);}

void test_times_out(void) {  struct sockaddr_in addr;  socklen_t addr_len = sizeof(addr);  int svr_fd;#define NUM_CLIENT_CONNECTS 10  int client_fd[NUM_CLIENT_CONNECTS];  int i;  int r;  gpr_event ev;  gpr_timespec connect_deadline;  gpr_event_init(&ev);  memset(&addr, 0, sizeof(addr));  addr.sin_family = AF_INET;  /* create a dummy server */  svr_fd = socket(AF_INET, SOCK_STREAM, 0);  GPR_ASSERT(svr_fd >= 0);  GPR_ASSERT(0 == bind(svr_fd, (struct sockaddr *)&addr, addr_len));  GPR_ASSERT(0 == listen(svr_fd, 1));  /* Get its address */  GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0);  /* tie up the listen buffer, which is somewhat arbitrarily sized. */  for (i = 0; i < NUM_CLIENT_CONNECTS; ++i) {    client_fd[i] = socket(AF_INET, SOCK_STREAM, 0);    grpc_set_socket_nonblocking(client_fd[i], 1);    do {      r = connect(client_fd[i], (struct sockaddr *)&addr, addr_len);    } while (r == -1 && errno == EINTR);    GPR_ASSERT(r < 0);    GPR_ASSERT(errno == EWOULDBLOCK || errno == EINPROGRESS);  }  /* connect to dummy server address */  connect_deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1);  grpc_tcp_client_connect(must_fail, &ev, (struct sockaddr *)&addr, addr_len,                          connect_deadline);  /* Make sure the event doesn't trigger early */  GPR_ASSERT(!gpr_event_wait(&ev, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(500)));  /* Now wait until it should have triggered */  sleep(1);  /* wait for the connection callback to finish */  GPR_ASSERT(gpr_event_wait(&ev, test_deadline()));  close(svr_fd);  for (i = 0; i < NUM_CLIENT_CONNECTS; ++i) {    close(client_fd[i]);  }}

// interleave waiting for an event with a timer checkstatic bool wait_loop(int deadline_seconds, gpr_event *ev) {  while (deadline_seconds) {    gpr_log(GPR_DEBUG, "Test: waiting for %d more seconds", deadline_seconds);    if (gpr_event_wait(ev, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1))) return true;    deadline_seconds--;    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;    grpc_timer_check(&exec_ctx, gpr_now(GPR_CLOCK_MONOTONIC), NULL);    grpc_exec_ctx_finish(&exec_ctx);  }  return false;}

void bad_ssl_run(grpc_server *server) {  int shutdown_started = 0;  int shutdown_finished = 0;  grpc_event ev;  grpc_call_error error;  grpc_call *s = NULL;  grpc_call_details call_details;  grpc_metadata_array request_metadata_recv;  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);  grpc_call_details_init(&call_details);  grpc_metadata_array_init(&request_metadata_recv);  grpc_server_register_completion_queue(server, cq, NULL);  grpc_server_start(server);  error = grpc_server_request_call(server, &s, &call_details,                                   &request_metadata_recv, cq, cq, (void *)1);  GPR_ASSERT(GRPC_CALL_OK == error);  signal(SIGINT, sigint_handler);  while (!shutdown_finished) {    if (got_sigint && !shutdown_started) {      gpr_log(GPR_INFO, "Shutting down due to SIGINT");      grpc_server_shutdown_and_notify(server, cq, NULL);      GPR_ASSERT(grpc_completion_queue_pluck(                     cq, NULL, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5), NULL)                     .type == GRPC_OP_COMPLETE);      grpc_completion_queue_shutdown(cq);      shutdown_started = 1;    }    ev = grpc_completion_queue_next(        cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),                         gpr_time_from_micros(1000000, GPR_TIMESPAN)),        NULL);    switch (ev.type) {      case GRPC_OP_COMPLETE:        GPR_ASSERT(ev.tag == (void *)1);        GPR_ASSERT(ev.success == 0);        break;      case GRPC_QUEUE_SHUTDOWN:        GPR_ASSERT(shutdown_started);        shutdown_finished = 1;        break;      case GRPC_QUEUE_TIMEOUT:        break;    }  }  GPR_ASSERT(s == NULL);  grpc_call_details_destroy(&call_details);  grpc_metadata_array_destroy(&request_metadata_recv);}

/* Write to a socket using the grpc_tcp API, then drain it directly.   Note that if the write does not complete immediately we need to drain the   socket in parallel with the read. */static void write_test(size_t num_bytes, size_t slice_size) {  int sv[2];  grpc_endpoint *ep;  struct write_socket_state state;  size_t num_blocks;  gpr_slice *slices;  gpr_uint8 current_data = 0;  gpr_slice_buffer outgoing;  grpc_closure write_done_closure;  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  gpr_log(GPR_INFO, "Start write test with %d bytes, slice size %d", num_bytes,          slice_size);  create_sockets(sv);  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test"),                       GRPC_TCP_DEFAULT_READ_SLICE_SIZE, "test");  grpc_endpoint_add_to_pollset(&exec_ctx, ep, &g_pollset);  state.ep = ep;  state.write_done = 0;  slices = allocate_blocks(num_bytes, slice_size, &num_blocks, &current_data);  gpr_slice_buffer_init(&outgoing);  gpr_slice_buffer_addn(&outgoing, slices, num_blocks);  grpc_closure_init(&write_done_closure, write_done, &state);  grpc_endpoint_write(&exec_ctx, ep, &outgoing, &write_done_closure);  drain_socket_blocking(sv[0], num_bytes, num_bytes);  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  for (;;) {    grpc_pollset_worker worker;    if (state.write_done) {      break;    }    grpc_pollset_work(&exec_ctx, &g_pollset, &worker,                      gpr_now(GPR_CLOCK_MONOTONIC), deadline);    gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));    grpc_exec_ctx_finish(&exec_ctx);    gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  }  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  gpr_slice_buffer_destroy(&outgoing);  grpc_endpoint_destroy(&exec_ctx, ep);  gpr_free(slices);  grpc_exec_ctx_finish(&exec_ctx);}

static void test_connect(int n) {  struct sockaddr_storage addr;  socklen_t addr_len = sizeof(addr);  int svrfd, clifd;  grpc_tcp_server *s = grpc_tcp_server_create();  int nconnects_before;  gpr_timespec deadline;  grpc_pollset *pollsets[1];  int i;  LOG_TEST("test_connect");  gpr_log(GPR_INFO, "clients=%d", n);  memset(&addr, 0, sizeof(addr));  addr.ss_family = AF_INET;  GPR_ASSERT(grpc_tcp_server_add_port(s, (struct sockaddr *)&addr, addr_len));  svrfd = grpc_tcp_server_get_fd(s, 0);  GPR_ASSERT(svrfd >= 0);  GPR_ASSERT(getsockname(svrfd, (struct sockaddr *)&addr, &addr_len) == 0);  GPR_ASSERT(addr_len <= sizeof(addr));  pollsets[0] = &g_pollset;  grpc_tcp_server_start(s, pollsets, 1, on_connect, NULL);  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  for (i = 0; i < n; i++) {    deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);    nconnects_before = g_nconnects;    clifd = socket(addr.ss_family, SOCK_STREAM, 0);    GPR_ASSERT(clifd >= 0);    gpr_log(GPR_DEBUG, "start connect");    GPR_ASSERT(connect(clifd, (struct sockaddr *)&addr, addr_len) == 0);    gpr_log(GPR_DEBUG, "wait");    while (g_nconnects == nconnects_before &&           gpr_time_cmp(deadline, gpr_now(GPR_CLOCK_REALTIME)) > 0) {      grpc_pollset_work(&g_pollset, deadline);    }    gpr_log(GPR_DEBUG, "wait done");    GPR_ASSERT(g_nconnects == nconnects_before + 1);    close(clifd);  }  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  grpc_tcp_server_destroy(s, NULL, NULL);}

static void test_connect(int n) {  struct sockaddr_storage addr;  socklen_t addr_len = sizeof(addr);  int svrfd, clifd;  grpc_tcp_server *s = grpc_tcp_server_create();  int nconnects_before;  gpr_timespec deadline;  int i;  LOG_TEST();  gpr_log(GPR_INFO, "clients=%d", n);  gpr_mu_lock(&mu);  memset(&addr, 0, sizeof(addr));  addr.ss_family = AF_INET;  GPR_ASSERT(grpc_tcp_server_add_port(s, (struct sockaddr *)&addr, addr_len));  svrfd = grpc_tcp_server_get_fd(s, 0);  GPR_ASSERT(svrfd >= 0);  GPR_ASSERT(getsockname(svrfd, (struct sockaddr *)&addr, &addr_len) == 0);  GPR_ASSERT(addr_len <= sizeof(addr));  grpc_tcp_server_start(s, NULL, 0, on_connect, NULL);  for (i = 0; i < n; i++) {    deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1);    nconnects_before = nconnects;    clifd = socket(addr.ss_family, SOCK_STREAM, 0);    GPR_ASSERT(clifd >= 0);    GPR_ASSERT(connect(clifd, (struct sockaddr *)&addr, addr_len) == 0);    while (nconnects == nconnects_before) {      GPR_ASSERT(gpr_cv_wait(&cv, &mu, deadline) == 0);    }    GPR_ASSERT(nconnects == nconnects_before + 1);    close(clifd);    if (i != n - 1) {      sleep(1);    }  }  gpr_mu_unlock(&mu);  grpc_tcp_server_destroy(s);}

static void test_ping(grpc_end2end_test_config config) {  grpc_end2end_test_fixture f = config.create_fixture(NULL, NULL);  cq_verifier *cqv = cq_verifier_create(f.cq);  grpc_connectivity_state state = GRPC_CHANNEL_IDLE;  int i;  config.init_client(&f, NULL, NULL);  config.init_server(&f, NULL);  grpc_channel_ping(f.client, f.cq, tag(0), NULL);  CQ_EXPECT_COMPLETION(cqv, tag(0), 0);  /* check that we're still in idle, and start connecting */  GPR_ASSERT(grpc_channel_check_connectivity_state(f.client, 1) ==             GRPC_CHANNEL_IDLE);  /* we'll go through some set of transitions (some might be missed), until     READY is reached */  while (state != GRPC_CHANNEL_READY) {    grpc_channel_watch_connectivity_state(        f.client, state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(99));    CQ_EXPECT_COMPLETION(cqv, tag(99), 1);    cq_verify(cqv);    state = grpc_channel_check_connectivity_state(f.client, 0);    GPR_ASSERT(state == GRPC_CHANNEL_READY ||               state == GRPC_CHANNEL_CONNECTING ||               state == GRPC_CHANNEL_TRANSIENT_FAILURE);  }  for (i = 1; i <= 5; i++) {    grpc_channel_ping(f.client, f.cq, tag(i), NULL);    CQ_EXPECT_COMPLETION(cqv, tag(i), 1);    cq_verify(cqv);  }  grpc_server_shutdown_and_notify(f.server, f.cq, tag(0xdead));  CQ_EXPECT_COMPLETION(cqv, tag(0xdead), 1);  cq_verify(cqv);  /* cleanup server */  grpc_server_destroy(f.server);  grpc_channel_destroy(f.client);  grpc_completion_queue_shutdown(f.cq);  grpc_completion_queue_destroy(f.cq);  config.tear_down_data(&f);  cq_verifier_destroy(cqv);}

/* Write to a socket until it fills up, then read from it using the grpc_tcp   API. */static void large_read_test(ssize_t slice_size) {  int sv[2];  grpc_endpoint *ep;  struct read_socket_state state;  ssize_t written_bytes;  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);  gpr_log(GPR_INFO, "Start large read test, slice size %d", slice_size);  create_sockets(sv);  ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), slice_size,                       "test");  grpc_endpoint_add_to_pollset(ep, &g_pollset);  written_bytes = fill_socket(sv[0]);  gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes);  state.ep = ep;  state.read_bytes = 0;  state.target_read_bytes = written_bytes;  gpr_slice_buffer_init(&state.incoming);  grpc_iomgr_closure_init(&state.read_cb, read_cb, &state);  switch (grpc_endpoint_read(ep, &state.incoming, &state.read_cb)) {    case GRPC_ENDPOINT_DONE:      read_cb(&state, 1);      break;    case GRPC_ENDPOINT_ERROR:      read_cb(&state, 0);      break;    case GRPC_ENDPOINT_PENDING:      break;  }  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  while (state.read_bytes < state.target_read_bytes) {    grpc_pollset_worker worker;    grpc_pollset_work(&g_pollset, &worker, gpr_now(GPR_CLOCK_MONOTONIC),                      deadline);  }  GPR_ASSERT(state.read_bytes == state.target_read_bytes);  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  gpr_slice_buffer_destroy(&state.incoming);  grpc_endpoint_destroy(ep);}

static void shutdown_during_write_test(grpc_endpoint_test_config config,                                       size_t slice_size) {  /* test that shutdown with a pending write creates no leaks */  gpr_timespec deadline;  size_t size;  size_t nblocks;  int current_data = 1;  shutdown_during_write_test_state read_st;  shutdown_during_write_test_state write_st;  gpr_slice *slices;  grpc_endpoint_test_fixture f = begin_test(config, __FUNCTION__, slice_size);  gpr_log(GPR_INFO, "testing shutdown during a write");  read_st.ep = f.client_ep;  write_st.ep = f.server_ep;  gpr_event_init(&read_st.ev);  gpr_event_init(&write_st.ev);  grpc_endpoint_notify_on_read(      read_st.ep, shutdown_during_write_test_read_handler, &read_st);  for (size = 1;; size *= 2) {    slices = allocate_blocks(size, 1, &nblocks, &current_data);    switch (grpc_endpoint_write(write_st.ep, slices, nblocks,                                shutdown_during_write_test_write_handler,                                &write_st)) {      case GRPC_ENDPOINT_WRITE_DONE:        break;      case GRPC_ENDPOINT_WRITE_ERROR:        gpr_log(GPR_ERROR, "error writing");        abort();      case GRPC_ENDPOINT_WRITE_PENDING:        grpc_endpoint_shutdown(write_st.ep);        deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);        GPR_ASSERT(gpr_event_wait(&write_st.ev, deadline));        grpc_endpoint_destroy(write_st.ep);        GPR_ASSERT(gpr_event_wait(&read_st.ev, deadline));        gpr_free(slices);        end_test(config);        return;    }    gpr_free(slices);  }  gpr_log(GPR_ERROR, "should never reach here");  abort();}

static void test_ping() {  grpc_channel *client;  request_data rdata;  servers_fixture *f;  cq_verifier *cqv;  grpc_connectivity_state state = GRPC_CHANNEL_IDLE;  const size_t num_servers = 1;  int i;  rdata.call_details = gpr_malloc(sizeof(grpc_call_details) * num_servers);  f = setup_servers("127.0.0.1", &rdata, num_servers);  cqv = cq_verifier_create(f->cq);  client = create_client(f);  grpc_channel_ping(client, f->cq, tag(0), NULL);  cq_expect_completion(cqv, tag(0), 0);  /* check that we're still in idle, and start connecting */  GPR_ASSERT(grpc_channel_check_connectivity_state(client, 1) ==             GRPC_CHANNEL_IDLE);  /* we'll go through some set of transitions (some might be missed), until     READY is reached */  while (state != GRPC_CHANNEL_READY) {    grpc_channel_watch_connectivity_state(        client, state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f->cq, tag(99));    cq_expect_completion(cqv, tag(99), 1);    cq_verify(cqv);    state = grpc_channel_check_connectivity_state(client, 0);    GPR_ASSERT(state == GRPC_CHANNEL_READY ||               state == GRPC_CHANNEL_CONNECTING ||               state == GRPC_CHANNEL_TRANSIENT_FAILURE);  }  for (i = 1; i <= 5; i++) {    grpc_channel_ping(client, f->cq, tag(i), NULL);    cq_expect_completion(cqv, tag(i), 1);    cq_verify(cqv);  }  gpr_free(rdata.call_details);  grpc_channel_destroy(client);  teardown_servers(f);  cq_verifier_destroy(cqv);}

static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,                                 const grpc_httpcli_response *response) {  size_t i;  int port = 0;  portreq *pr = arg;  int failed = 0;  if (!response) {    failed = 1;    gpr_log(GPR_DEBUG,            "failed port pick from server: retrying [response=NULL]");  } else if (response->status != 200) {    failed = 1;    gpr_log(GPR_DEBUG, "failed port pick from server: status=%d",            response->status);  }  if (failed) {    grpc_httpcli_request req;    memset(&req, 0, sizeof(req));    GPR_ASSERT(pr->retries < 10);    gpr_sleep_until(gpr_time_add(        gpr_now(GPR_CLOCK_REALTIME),        gpr_time_from_millis(            (int64_t)(1000.0 * (1 + pow(1.3, pr->retries) * rand() / RAND_MAX)),            GPR_TIMESPAN)));    pr->retries++;    req.host = pr->server;    req.http.path = "/get";    grpc_httpcli_get(exec_ctx, pr->ctx, &pr->pops, &req,                     GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), got_port_from_server,                     pr);    return;  }  GPR_ASSERT(response);  GPR_ASSERT(response->status == 200);  for (i = 0; i < response->body_length; i++) {    GPR_ASSERT(response->body[i] >= '0' && response->body[i] <= '9');    port = port * 10 + response->body[i] - '0';  }  GPR_ASSERT(port > 1024);  gpr_mu_lock(pr->mu);  pr->port = port;  grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), NULL);  gpr_mu_unlock(pr->mu);}

/* Write to a socket until it fills up, then read from it using the grpc_tcp   API. */static void large_read_test(size_t slice_size) {  int sv[2];  grpc_endpoint *ep;  struct read_socket_state state;  ssize_t written_bytes;  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  gpr_log(GPR_INFO, "Start large read test, slice size %d", slice_size);  create_sockets(sv);  ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), slice_size,                       "test");  grpc_endpoint_add_to_pollset(&exec_ctx, ep, &g_pollset);  written_bytes = fill_socket(sv[0]);  gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes);  state.ep = ep;  state.read_bytes = 0;  state.target_read_bytes = (size_t)written_bytes;  gpr_slice_buffer_init(&state.incoming);  grpc_closure_init(&state.read_cb, read_cb, &state);  grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  while (state.read_bytes < state.target_read_bytes) {    grpc_pollset_worker worker;    grpc_pollset_work(&exec_ctx, &g_pollset, &worker,                      gpr_now(GPR_CLOCK_MONOTONIC), deadline);    gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));    grpc_exec_ctx_finish(&exec_ctx);    gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  }  GPR_ASSERT(state.read_bytes == state.target_read_bytes);  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  gpr_slice_buffer_destroy(&state.incoming);  grpc_endpoint_destroy(&exec_ctx, ep);  grpc_exec_ctx_finish(&exec_ctx);}

void cq_verify(cq_verifier *v) {  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);  grpc_event ev;  expectation *e;  char *s;  gpr_strvec have_tags;  gpr_strvec_init(&have_tags);  while (v->expect.next != &v->expect) {    ev = grpc_completion_queue_next(v->cq, deadline, NULL);    if (ev.type == GRPC_QUEUE_TIMEOUT) {      fail_no_event_received(v);      break;    }    for (e = v->expect.next; e != &v->expect; e = e->next) {      gpr_asprintf(&s, " %p", e->tag);      gpr_strvec_add(&have_tags, s);      if (e->tag == ev.tag) {        verify_matches(e, &ev);        e->next->prev = e->prev;        e->prev->next = e->next;        gpr_free(e);        break;      }    }    if (e == &v->expect) {      s = grpc_event_string(&ev);      gpr_log(GPR_ERROR, "event not found: %s", s);      gpr_free(s);      s = gpr_strvec_flatten(&have_tags, NULL);      gpr_log(GPR_ERROR, "have tags:%s", s);      gpr_free(s);      gpr_strvec_destroy(&have_tags);      abort();    }  }  gpr_strvec_destroy(&have_tags);}

int main(int argc, char **argv) {  grpc_channel *chan;  grpc_call *call;  grpc_metadata md = {"a", "b", 1, {{NULL, NULL, NULL}}};  grpc_completion_queue *cq;  cq_verifier *cqv;  grpc_test_init(argc, argv);  grpc_init();  chan = grpc_lame_client_channel_create();  GPR_ASSERT(chan);  call = grpc_channel_create_call_old(chan, "/Foo", "anywhere",                                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(100));  GPR_ASSERT(call);  cq = grpc_completion_queue_create();  cqv = cq_verifier_create(cq);  /* we should be able to add metadata */  GPR_ASSERT(GRPC_CALL_OK == grpc_call_add_metadata_old(call, &md, 0));  /* and invoke the call */  GPR_ASSERT(GRPC_CALL_OK == grpc_call_invoke_old(call, cq, tag(2), tag(3), 0));  /* the call should immediately fail */  cq_expect_client_metadata_read(cqv, tag(2), NULL);  cq_expect_finished(cqv, tag(3), NULL);  cq_verify(cqv);  grpc_call_destroy(call);  grpc_channel_destroy(chan);  cq_verifier_destroy(cqv);  grpc_completion_queue_destroy(cq);  grpc_shutdown();  return 0;}

static void test_add_closure(void) {  grpc_closure c;  int done = 0;  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  grpc_workqueue *wq = grpc_workqueue_create(&exec_ctx);  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5);  grpc_pollset_worker worker;  grpc_closure_init(&c, must_succeed, &done);  grpc_workqueue_push(wq, &c, 1);  grpc_workqueue_add_to_pollset(&exec_ctx, wq, &g_pollset);  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));  GPR_ASSERT(!done);  grpc_pollset_work(&exec_ctx, &g_pollset, &worker,                    gpr_now(deadline.clock_type), deadline);  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));  grpc_exec_ctx_finish(&exec_ctx);  GPR_ASSERT(done);  GRPC_WORKQUEUE_UNREF(&exec_ctx, wq, "destroy");  grpc_exec_ctx_finish(&exec_ctx);}

static void test_flush(void) {  grpc_closure c;  int done = 0;  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  grpc_workqueue *wq = grpc_workqueue_create(&exec_ctx);  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5);  grpc_pollset_worker *worker = NULL;  grpc_closure_init(&c, must_succeed, &done);  grpc_exec_ctx_enqueue(&exec_ctx, &c, true, NULL);  grpc_workqueue_flush(&exec_ctx, wq);  grpc_workqueue_add_to_pollset(&exec_ctx, wq, g_pollset);  gpr_mu_lock(g_mu);  GPR_ASSERT(!done);  grpc_pollset_work(&exec_ctx, g_pollset, &worker, gpr_now(deadline.clock_type),                    deadline);  gpr_mu_unlock(g_mu);  grpc_exec_ctx_finish(&exec_ctx);  GPR_ASSERT(done);  GRPC_WORKQUEUE_UNREF(&exec_ctx, wq, "destroy");  grpc_exec_ctx_finish(&exec_ctx);}

static gpr_timespec n_seconds_time(int n) {  return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n);}

static void run_test(const char *target, size_t nops) {    grpc_channel_credentials *ssl_creds =        grpc_ssl_credentials_create(NULL, NULL, NULL);    grpc_channel *channel;    grpc_call *c;    grpc_metadata_array initial_metadata_recv;    grpc_metadata_array trailing_metadata_recv;    char *details = NULL;    size_t details_capacity = 0;    grpc_status_code status;    grpc_call_error error;    gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5);    grpc_completion_queue *cq = grpc_completion_queue_create(NULL);    cq_verifier *cqv = cq_verifier_create(cq);    grpc_op ops[6];    grpc_op *op;    grpc_arg ssl_name_override = {GRPC_ARG_STRING,                                  GRPC_SSL_TARGET_NAME_OVERRIDE_ARG,    {"foo.test.google.fr"}                                 };    grpc_channel_args args;    args.num_args = 1;    args.args = &ssl_name_override;    grpc_metadata_array_init(&initial_metadata_recv);    grpc_metadata_array_init(&trailing_metadata_recv);    channel = grpc_secure_channel_create(ssl_creds, target, &args, NULL);    c = grpc_channel_create_call(channel, NULL, GRPC_PROPAGATE_DEFAULTS, cq,                                 "/foo", "foo.test.google.fr:1234", deadline,                                 NULL);    op = ops;    op->op = GRPC_OP_SEND_INITIAL_METADATA;    op->data.send_initial_metadata.count = 0;    op->flags = 0;    op->reserved = NULL;    op++;    op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;    op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;    op->data.recv_status_on_client.status = &status;    op->data.recv_status_on_client.status_details = &details;    op->data.recv_status_on_client.status_details_capacity = &details_capacity;    op->flags = 0;    op->reserved = NULL;    op++;    op->op = GRPC_OP_RECV_INITIAL_METADATA;    op->data.recv_initial_metadata = &initial_metadata_recv;    op->flags = 0;    op->reserved = NULL;    op++;    op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;    op->flags = 0;    op->reserved = NULL;    op++;    error = grpc_call_start_batch(c, ops, nops, tag(1), NULL);    GPR_ASSERT(GRPC_CALL_OK == error);    cq_expect_completion(cqv, tag(1), 1);    cq_verify(cqv);    GPR_ASSERT(status != GRPC_STATUS_OK);    grpc_call_destroy(c);    gpr_free(details);    grpc_metadata_array_destroy(&initial_metadata_recv);    grpc_metadata_array_destroy(&trailing_metadata_recv);    grpc_channel_destroy(channel);    grpc_completion_queue_destroy(cq);    cq_verifier_destroy(cqv);    grpc_channel_credentials_release(ssl_creds);}

static void test_receive(int number_of_clients) {  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;  struct sockaddr_storage addr;  socklen_t addr_len = sizeof(addr);  int clifd, svrfd;  grpc_udp_server *s = grpc_udp_server_create();  int i;  int number_of_reads_before;  gpr_timespec deadline;  grpc_pollset *pollsets[1];  LOG_TEST("test_receive");  gpr_log(GPR_INFO, "clients=%d", number_of_clients);  g_number_of_bytes_read = 0;  g_number_of_orphan_calls = 0;  memset(&addr, 0, sizeof(addr));  addr.ss_family = AF_INET;  GPR_ASSERT(grpc_udp_server_add_port(s, (struct sockaddr *)&addr, addr_len,                                      on_read, on_fd_orphaned));  svrfd = grpc_udp_server_get_fd(s, 0);  GPR_ASSERT(svrfd >= 0);  GPR_ASSERT(getsockname(svrfd, (struct sockaddr *)&addr, &addr_len) == 0);  GPR_ASSERT(addr_len <= sizeof(addr));  pollsets[0] = g_pollset;  grpc_udp_server_start(&exec_ctx, s, pollsets, 1, NULL);  gpr_mu_lock(g_mu);  for (i = 0; i < number_of_clients; i++) {    deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);    number_of_reads_before = g_number_of_reads;    /* Create a socket, send a packet to the UDP server. */    clifd = socket(addr.ss_family, SOCK_DGRAM, 0);    GPR_ASSERT(clifd >= 0);    GPR_ASSERT(connect(clifd, (struct sockaddr *)&addr, addr_len) == 0);    GPR_ASSERT(5 == write(clifd, "hello", 5));    while (g_number_of_reads == number_of_reads_before &&           gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) > 0) {      grpc_pollset_worker *worker = NULL;      GPR_ASSERT(GRPC_LOG_IF_ERROR(          "pollset_work",          grpc_pollset_work(&exec_ctx, g_pollset, &worker,                            gpr_now(GPR_CLOCK_MONOTONIC), deadline)));      gpr_mu_unlock(g_mu);      grpc_exec_ctx_finish(&exec_ctx);      gpr_mu_lock(g_mu);    }    GPR_ASSERT(g_number_of_reads == number_of_reads_before + 1);    close(clifd);  }  GPR_ASSERT(g_number_of_bytes_read == 5 * number_of_clients);  gpr_mu_unlock(g_mu);  grpc_udp_server_destroy(&exec_ctx, s, NULL);  grpc_exec_ctx_finish(&exec_ctx);  /* The server had a single FD, which should have been orphaned. */  GPR_ASSERT(g_number_of_orphan_calls == 1);}

/** Returns connection sequence (server indices), which must be freed */static int *perform_request(servers_fixture *f, grpc_channel *client,                            request_data *rdata, const test_spec *spec) {  grpc_call *c;  int s_idx;  int *s_valid;  grpc_op ops[6];  grpc_op *op;  int was_cancelled;  size_t i, iter_num;  grpc_event ev;  int read_tag;  int *connection_sequence;  int completed_client;  s_valid = gpr_malloc(sizeof(int) * f->num_servers);  connection_sequence = gpr_malloc(sizeof(int) * spec->num_iters);  for (iter_num = 0; iter_num < spec->num_iters; iter_num++) {    cq_verifier *cqv = cq_verifier_create(f->cq);    rdata->details = NULL;    rdata->details_capacity = 0;    was_cancelled = 2;    for (i = 0; i < f->num_servers; i++) {      if (spec->kill_at[iter_num][i] != 0) {        kill_server(f, i);      } else if (spec->revive_at[iter_num][i] != 0) {        /* killing takes precedence */        revive_server(f, rdata, i);      }    }    connection_sequence[iter_num] = -1;    grpc_metadata_array_init(&rdata->initial_metadata_recv);    grpc_metadata_array_init(&rdata->trailing_metadata_recv);    for (i = 0; i < f->num_servers; i++) {      grpc_call_details_init(&rdata->call_details[i]);    }    memset(s_valid, 0, f->num_servers * sizeof(int));    c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, f->cq,                                 "/foo", "foo.test.google.fr",                                 gpr_inf_future(GPR_CLOCK_REALTIME), NULL);    GPR_ASSERT(c);    completed_client = 0;    op = ops;    op->op = GRPC_OP_SEND_INITIAL_METADATA;    op->data.send_initial_metadata.count = 0;    op->flags = 0;    op->reserved = NULL;    op++;    op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;    op->flags = 0;    op->reserved = NULL;    op++;    op->op = GRPC_OP_RECV_INITIAL_METADATA;    op->data.recv_initial_metadata = &rdata->initial_metadata_recv;    op->flags = 0;    op->reserved = NULL;    op++;    op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;    op->data.recv_status_on_client.trailing_metadata =        &rdata->trailing_metadata_recv;    op->data.recv_status_on_client.status = &rdata->status;    op->data.recv_status_on_client.status_details = &rdata->details;    op->data.recv_status_on_client.status_details_capacity =        &rdata->details_capacity;    op->flags = 0;    op->reserved = NULL;    op++;    GPR_ASSERT(GRPC_CALL_OK ==               grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL));    s_idx = -1;    while ((ev = grpc_completion_queue_next(                f->cq, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1), NULL))               .type != GRPC_QUEUE_TIMEOUT) {      GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);      read_tag = ((int)(gpr_intptr)ev.tag);      gpr_log(GPR_DEBUG, "EVENT: success:%d, type:%d, tag:%d iter:%d",              ev.success, ev.type, read_tag, iter_num);      if (ev.success && read_tag >= 1000) {        GPR_ASSERT(s_idx == -1); /* only one server must reply */        /* only server notifications for non-shutdown events */        s_idx = read_tag - 1000;        s_valid[s_idx] = 1;        connection_sequence[iter_num] = s_idx;        break;      } else if (read_tag == 1) {        gpr_log(GPR_DEBUG, "client timed out");        GPR_ASSERT(ev.success);        completed_client = 1;      }    }    if (s_idx >= 0) {      op = ops;//.........这里部分代码省略.........

void grpc_run_bad_client_test(const char *name, const char *client_payload,                              size_t client_payload_length,                              grpc_bad_client_server_side_validator validator) {  grpc_endpoint_pair sfd;  thd_args a;  gpr_thd_id id;  gpr_slice slice =      gpr_slice_from_copied_buffer(client_payload, client_payload_length);  /* Add a debug log */  gpr_log(GPR_INFO, "TEST: %s", name);  /* Init grpc */  grpc_init();  /* Create endpoints */  sfd = grpc_iomgr_create_endpoint_pair(65536);  /* Create server, completion events */  a.server = grpc_server_create_from_filters(NULL, 0, NULL);  a.cq = grpc_completion_queue_create();  gpr_event_init(&a.done_thd);  gpr_event_init(&a.done_write);  a.validator = validator;  grpc_server_register_completion_queue(a.server, a.cq);  grpc_server_start(a.server);  grpc_create_chttp2_transport(server_setup_transport, &a, NULL, sfd.server,                               NULL, 0, grpc_mdctx_create(), 0);  /* Bind everything into the same pollset */  grpc_endpoint_add_to_pollset(sfd.client, grpc_cq_pollset(a.cq));  grpc_endpoint_add_to_pollset(sfd.server, grpc_cq_pollset(a.cq));  /* Check a ground truth */  GPR_ASSERT(grpc_server_has_open_connections(a.server));  /* Start validator */  gpr_thd_new(&id, thd_func, &a, NULL);  /* Write data */  switch (grpc_endpoint_write(sfd.client, &slice, 1, done_write, &a)) {    case GRPC_ENDPOINT_WRITE_DONE:      done_write(&a, 1);      break;    case GRPC_ENDPOINT_WRITE_PENDING:      break;    case GRPC_ENDPOINT_WRITE_ERROR:      done_write(&a, 0);      break;  }  /* Await completion */  GPR_ASSERT(      gpr_event_wait(&a.done_write, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)));  GPR_ASSERT(gpr_event_wait(&a.done_thd, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)));  /* Shutdown */  grpc_endpoint_destroy(sfd.client);  grpc_server_destroy(a.server);  grpc_completion_queue_destroy(a.cq);  grpc_shutdown();}

int main(int argc, char **argv) {  gpr_slice slice = gpr_slice_from_copied_string("x");  double start, stop;  unsigned i;  char *fake_argv[1];  int payload_size = 1;  int secure = 0;  char *target = "localhost:443";  gpr_cmdline *cl;  grpc_event event;  char *scenario_name = "ping-pong-request";  scenario sc = {NULL, NULL, NULL};  gpr_timers_set_log_filename("latency_trace.fling_client.txt");  grpc_init();  GPR_ASSERT(argc >= 1);  fake_argv[0] = argv[0];  grpc_test_init(1, fake_argv);  cl = gpr_cmdline_create("fling client");  gpr_cmdline_add_int(cl, "payload_size", "Size of the payload to send",                      &payload_size);  gpr_cmdline_add_string(cl, "target", "Target host:port", &target);  gpr_cmdline_add_flag(cl, "secure", "Run with security?", &secure);  gpr_cmdline_add_string(cl, "scenario", "Scenario", &scenario_name);  gpr_cmdline_parse(cl, argc, argv);  gpr_cmdline_destroy(cl);  for (i = 0; i < GPR_ARRAY_SIZE(scenarios); i++) {    if (0 == strcmp(scenarios[i].name, scenario_name)) {      sc = scenarios[i];    }  }  if (!sc.name) {    fprintf(stderr, "unsupported scenario '%s'. Valid are:", scenario_name);    for (i = 0; i < GPR_ARRAY_SIZE(scenarios); i++) {      fprintf(stderr, " %s", scenarios[i].name);    }    return 1;  }  channel = grpc_insecure_channel_create(target, NULL, NULL);  cq = grpc_completion_queue_create(NULL);  the_buffer = grpc_raw_byte_buffer_create(&slice, (size_t)payload_size);  histogram = gpr_histogram_create(0.01, 60e9);  sc.init();  gpr_timespec end_warmup = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3);  gpr_timespec end_profiling = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(30);  while (gpr_time_cmp(gpr_now(end_warmup.clock_type), end_warmup) < 0) {    sc.do_one_step();  }  gpr_log(GPR_INFO, "start profiling");  grpc_profiler_start("client.prof");  while (gpr_time_cmp(gpr_now(end_profiling.clock_type), end_profiling) < 0) {    start = now();    sc.do_one_step();    stop = now();    gpr_histogram_add(histogram, stop - start);  }  grpc_profiler_stop();  if (call) {    grpc_call_destroy(call);  }  grpc_channel_destroy(channel);  grpc_completion_queue_shutdown(cq);  do {    event = grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),                                       NULL);  } while (event.type != GRPC_QUEUE_SHUTDOWN);  grpc_completion_queue_destroy(cq);  grpc_byte_buffer_destroy(the_buffer);  gpr_slice_unref(slice);  gpr_log(GPR_INFO, "latency (50/95/99/99.9): %f/%f/%f/%f",          gpr_histogram_percentile(histogram, 50),          gpr_histogram_percentile(histogram, 95),          gpr_histogram_percentile(histogram, 99),          gpr_histogram_percentile(histogram, 99.9));  gpr_histogram_destroy(histogram);  grpc_shutdown();  return 0;}