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 check
static 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);
}