diff options
Diffstat (limited to 'test/core')
-rw-r--r-- | test/core/client_config/lb_policies_test.c | 120 | ||||
-rw-r--r-- | test/core/network_benchmarks/low_level_ping_pong.c | 70 | ||||
-rw-r--r-- | test/core/surface/byte_buffer_reader_test.c | 4 |
3 files changed, 98 insertions, 96 deletions
diff --git a/test/core/client_config/lb_policies_test.c b/test/core/client_config/lb_policies_test.c index 68e991716b..e44930ed53 100644 --- a/test/core/client_config/lb_policies_test.c +++ b/test/core/client_config/lb_policies_test.c @@ -119,14 +119,15 @@ static void test_spec_destroy(test_spec *spec) { static void *tag(gpr_intptr t) { return (void *)t; } -static gpr_timespec n_seconds_time(int n) { - return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n); +static gpr_timespec n_millis_time(int n) { + return gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_millis(n, GPR_TIMESPAN)); } static void drain_cq(grpc_completion_queue *cq) { grpc_event ev; do { - ev = grpc_completion_queue_next(cq, n_seconds_time(5), NULL); + ev = grpc_completion_queue_next(cq, n_millis_time(5000), NULL); } while (ev.type != GRPC_QUEUE_SHUTDOWN); } @@ -134,29 +135,47 @@ static void kill_server(const servers_fixture *f, size_t i) { gpr_log(GPR_INFO, "KILLING SERVER %d", i); GPR_ASSERT(f->servers[i] != NULL); grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000)); - GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000), - GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5), - NULL).type == GRPC_OP_COMPLETE); + GPR_ASSERT( + grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000), NULL) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->servers[i]); f->servers[i] = NULL; } -static void revive_server(const servers_fixture *f, size_t i) { +typedef struct request_data { + grpc_metadata_array initial_metadata_recv; + grpc_metadata_array trailing_metadata_recv; + char *details; + size_t details_capacity; + grpc_status_code status; + grpc_call_details *call_details; +} request_data; + +static void revive_server(const servers_fixture *f, request_data *rdata, + size_t i) { int got_port; gpr_log(GPR_INFO, "RAISE AGAIN SERVER %d", i); GPR_ASSERT(f->servers[i] == NULL); + + gpr_log(GPR_DEBUG, "revive: %s", f->servers_hostports[i]); + f->servers[i] = grpc_server_create(NULL, NULL); grpc_server_register_completion_queue(f->servers[i], f->cq, NULL); GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port( f->servers[i], f->servers_hostports[i])) > 0); grpc_server_start(f->servers[i]); + + GPR_ASSERT(GRPC_CALL_OK == + grpc_server_request_call(f->servers[i], &f->server_calls[i], + &rdata->call_details[i], + &f->request_metadata_recv[i], f->cq, + f->cq, tag(1000 + (int)i))); } static servers_fixture *setup_servers(const char *server_host, + request_data *rdata, const size_t num_servers) { servers_fixture *f = gpr_malloc(sizeof(servers_fixture)); - int *ports; - int got_port; size_t i; f->num_servers = num_servers; @@ -164,23 +183,16 @@ static servers_fixture *setup_servers(const char *server_host, f->request_metadata_recv = gpr_malloc(sizeof(grpc_metadata_array) * num_servers); /* Create servers. */ - ports = gpr_malloc(sizeof(int *) * num_servers); f->servers = gpr_malloc(sizeof(grpc_server *) * num_servers); f->servers_hostports = gpr_malloc(sizeof(char *) * num_servers); f->cq = grpc_completion_queue_create(NULL); for (i = 0; i < num_servers; i++) { - ports[i] = grpc_pick_unused_port_or_die(); - - gpr_join_host_port(&f->servers_hostports[i], server_host, ports[i]); - - f->servers[i] = grpc_server_create(NULL, NULL); - grpc_server_register_completion_queue(f->servers[i], f->cq, NULL); - GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port( - f->servers[i], f->servers_hostports[i])) > 0); - GPR_ASSERT(ports[i] == got_port); - grpc_server_start(f->servers[i]); + grpc_metadata_array_init(&f->request_metadata_recv[i]); + gpr_join_host_port(&f->servers_hostports[i], server_host, + grpc_pick_unused_port_or_die()); + f->servers[i] = 0; + revive_server(f, rdata, i); } - gpr_free(ports); return f; } @@ -191,8 +203,8 @@ static void teardown_servers(servers_fixture *f) { if (f->servers[i] == NULL) continue; grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000)); GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000), - GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5), - NULL).type == GRPC_OP_COMPLETE); + n_millis_time(5000), NULL) + .type == GRPC_OP_COMPLETE); grpc_server_destroy(f->servers[i]); } grpc_completion_queue_shutdown(f->cq); @@ -203,6 +215,7 @@ static void teardown_servers(servers_fixture *f) { for (i = 0; i < f->num_servers; i++) { gpr_free(f->servers_hostports[i]); + grpc_metadata_array_destroy(&f->request_metadata_recv[i]); } gpr_free(f->servers_hostports); @@ -211,22 +224,12 @@ static void teardown_servers(servers_fixture *f) { gpr_free(f); } -typedef struct request_data { - grpc_metadata_array initial_metadata_recv; - grpc_metadata_array trailing_metadata_recv; - char *details; - size_t details_capacity; - grpc_status_code status; - grpc_call_details *call_details; -} request_data; - /** Returns connection sequence (server indices), which must be freed */ 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; - gpr_timespec deadline; grpc_op ops[6]; grpc_op *op; int was_cancelled; @@ -237,7 +240,6 @@ int *perform_request(servers_fixture *f, grpc_channel *client, int completed_client; s_valid = gpr_malloc(sizeof(int) * f->num_servers); - rdata->call_details = gpr_malloc(sizeof(grpc_call_details) * f->num_servers); connection_sequence = gpr_malloc(sizeof(int) * spec->num_iters); /* Send a trivial request. */ @@ -253,7 +255,7 @@ int *perform_request(servers_fixture *f, grpc_channel *client, kill_server(f, i); } else if (spec->revive_at[iter_num][i] != 0) { /* killing takes precedence */ - revive_server(f, i); + revive_server(f, rdata, i); } } @@ -266,9 +268,9 @@ int *perform_request(servers_fixture *f, grpc_channel *client, } memset(s_valid, 0, f->num_servers * sizeof(int)); - deadline = n_seconds_time(1); c = grpc_channel_create_call(client, NULL, GRPC_PROPAGATE_DEFAULTS, f->cq, - "/foo", "foo.test.google.fr", deadline, NULL); + "/foo", "foo.test.google.fr", gpr_inf_future(GPR_CLOCK_REALTIME), + NULL); GPR_ASSERT(c); completed_client = 0; @@ -300,22 +302,9 @@ int *perform_request(servers_fixture *f, grpc_channel *client, GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL)); - /* "listen" on all servers */ - for (i = 0; i < f->num_servers; i++) { - grpc_metadata_array_init(&f->request_metadata_recv[i]); - if (f->servers[i] != NULL) { - GPR_ASSERT(GRPC_CALL_OK == - grpc_server_request_call(f->servers[i], &f->server_calls[i], - &rdata->call_details[i], - &f->request_metadata_recv[i], f->cq, - f->cq, tag(1000 + (int)i))); - } - } - s_idx = -1; - while ((ev = grpc_completion_queue_next( - f->cq, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(300), NULL)).type != - GRPC_QUEUE_TIMEOUT) { + while ((ev = grpc_completion_queue_next(f->cq, n_millis_time(s_idx == -1 ? 3000 : 200), 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", @@ -327,11 +316,14 @@ int *perform_request(servers_fixture *f, grpc_channel *client, s_valid[s_idx] = 1; connection_sequence[iter_num] = s_idx; } else if (read_tag == 1) { + gpr_log(GPR_DEBUG, "client timed out"); GPR_ASSERT(ev.success); completed_client = 1; } } + gpr_log(GPR_DEBUG, "s_idx=%d", s_idx); + if (s_idx >= 0) { op = ops; op->op = GRPC_OP_SEND_INITIAL_METADATA; @@ -361,25 +353,30 @@ int *perform_request(servers_fixture *f, grpc_channel *client, } cq_verify(cqv); + gpr_log(GPR_DEBUG, "status=%d; %s", rdata->status, rdata->details); GPR_ASSERT(rdata->status == GRPC_STATUS_UNIMPLEMENTED); GPR_ASSERT(0 == strcmp(rdata->details, "xyz")); GPR_ASSERT(0 == strcmp(rdata->call_details[s_idx].method, "/foo")); GPR_ASSERT(0 == strcmp(rdata->call_details[s_idx].host, "foo.test.google.fr")); GPR_ASSERT(was_cancelled == 1); + + grpc_call_destroy(f->server_calls[s_idx]); + + /* ask for the next request on this server */ + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( + f->servers[s_idx], &f->server_calls[s_idx], + &rdata->call_details[s_idx], + &f->request_metadata_recv[s_idx], f->cq, + f->cq, tag(1000 + (int)s_idx))); } else { + grpc_call_cancel(c, NULL); if (!completed_client) { cq_expect_completion(cqv, tag(1), 1); cq_verify(cqv); } } - for (i = 0; i < f->num_servers; i++) { - if (s_valid[i] != 0) { - grpc_call_destroy(f->server_calls[i]); - } - grpc_metadata_array_destroy(&f->request_metadata_recv[i]); - } grpc_metadata_array_destroy(&rdata->initial_metadata_recv); grpc_metadata_array_destroy(&rdata->trailing_metadata_recv); @@ -393,7 +390,6 @@ int *perform_request(servers_fixture *f, grpc_channel *client, gpr_free(rdata->details); } - gpr_free(rdata->call_details); gpr_free(s_valid); return connection_sequence; @@ -456,7 +452,10 @@ void run_spec(const test_spec *spec) { char *servers_hostports_str; int *actual_connection_sequence; request_data rdata; - servers_fixture *f = setup_servers("127.0.0.1", spec->num_servers); + servers_fixture *f; + rdata.call_details = + gpr_malloc(sizeof(grpc_call_details) * spec->num_servers); + f = setup_servers("127.0.0.1", &rdata, spec->num_servers); /* Create client. */ servers_hostports_str = gpr_strjoin_sep((const char **)f->servers_hostports, @@ -475,6 +474,7 @@ void run_spec(const test_spec *spec) { gpr_free(client_hostport); gpr_free(servers_hostports_str); gpr_free(actual_connection_sequence); + gpr_free(rdata.call_details); grpc_channel_destroy(client); teardown_servers(f); diff --git a/test/core/network_benchmarks/low_level_ping_pong.c b/test/core/network_benchmarks/low_level_ping_pong.c index be82a36306..7a2d894481 100644 --- a/test/core/network_benchmarks/low_level_ping_pong.c +++ b/test/core/network_benchmarks/low_level_ping_pong.c @@ -82,9 +82,9 @@ typedef struct thread_args { /* Basic call to read() */ static int read_bytes(int fd, char *buf, size_t read_size, int spin) { size_t bytes_read = 0; - int err; + ssize_t err; do { - err = (int)read(fd, buf + bytes_read, read_size - bytes_read); + err = read(fd, buf + bytes_read, read_size - bytes_read); if (err < 0) { if (errno == EINTR) { continue; @@ -115,6 +115,7 @@ static int poll_read_bytes(int fd, char *buf, size_t read_size, int spin) { struct pollfd pfd; size_t bytes_read = 0; int err; + ssize_t err2; pfd.fd = fd; pfd.events = POLLIN; @@ -132,13 +133,13 @@ static int poll_read_bytes(int fd, char *buf, size_t read_size, int spin) { GPR_ASSERT(err == 1); GPR_ASSERT(pfd.revents == POLLIN); do { - err = (int)read(fd, buf + bytes_read, read_size - bytes_read); - } while (err < 0 && errno == EINTR); - if (err < 0 && errno != EAGAIN) { + err2 = read(fd, buf + bytes_read, read_size - bytes_read); + } while (err2 < 0 && errno == EINTR); + if (err2 < 0 && errno != EAGAIN) { gpr_log(GPR_ERROR, "Read failed: %s", strerror(errno)); return -1; } - bytes_read += (size_t)err; + bytes_read += (size_t) err2; } while (bytes_read < read_size); return 0; } @@ -157,6 +158,7 @@ static int epoll_read_bytes(struct thread_args *args, char *buf, int spin) { struct epoll_event ev; size_t bytes_read = 0; int err; + ssize_t err2; size_t read_size = args->msg_size; do { @@ -172,11 +174,11 @@ static int epoll_read_bytes(struct thread_args *args, char *buf, int spin) { GPR_ASSERT(ev.data.fd == args->fds.read_fd); do { do { - err = (int)read(args->fds.read_fd, buf + bytes_read, - read_size - bytes_read); - } while (err < 0 && errno == EINTR); + err2 = read(args->fds.read_fd, buf + bytes_read, + read_size - bytes_read); + } while (err2 < 0 && errno == EINTR); if (errno == EAGAIN) break; - bytes_read += (size_t)err; + bytes_read += (size_t) err2; /* TODO(klempner): This should really be doing an extra call after we are done to ensure we see an EAGAIN */ } while (bytes_read < read_size); @@ -200,11 +202,11 @@ static int epoll_read_bytes_spin(struct thread_args *args, char *buf) { */ static int blocking_write_bytes(struct thread_args *args, char *buf) { size_t bytes_written = 0; - int err; + ssize_t err; size_t write_size = args->msg_size; do { - err = (int)write(args->fds.write_fd, buf + bytes_written, - write_size - bytes_written); + err = write(args->fds.write_fd, buf + bytes_written, + write_size - bytes_written); if (err < 0) { if (errno == EINTR) { continue; @@ -298,7 +300,7 @@ static void print_histogram(gpr_histogram *histogram) { static double now(void) { gpr_timespec tv = gpr_now(GPR_CLOCK_REALTIME); - return 1e9 * (double)tv.tv_sec + tv.tv_nsec; + return 1e9 * (double)tv.tv_sec + (double)tv.tv_nsec; } static void client_thread(thread_args *args) { @@ -374,7 +376,7 @@ error: return -1; } -static int connect_client(struct sockaddr *addr, int len) { +static int connect_client(struct sockaddr *addr, socklen_t len) { int fd = socket(addr->sa_family, SOCK_STREAM, 0); int err; if (fd < 0) { @@ -388,7 +390,7 @@ static int connect_client(struct sockaddr *addr, int len) { } do { - err = (int)connect(fd, addr, (socklen_t)len); + err = connect(fd, addr, len); } while (err < 0 && errno == EINTR); if (err < 0) { @@ -587,27 +589,27 @@ static int run_benchmark(char *socket_type, thread_args *client_args, return 0; } -static int run_all_benchmarks(int msg_size) { +static int run_all_benchmarks(size_t msg_size) { int error = 0; size_t i; for (i = 0; i < GPR_ARRAY_SIZE(test_strategies); ++i) { - test_strategy *ts = &test_strategies[i]; + test_strategy *strategy = &test_strategies[i]; size_t j; for (j = 0; j < GPR_ARRAY_SIZE(socket_types); ++j) { thread_args *client_args = malloc(sizeof(thread_args)); thread_args *server_args = malloc(sizeof(thread_args)); char *socket_type = socket_types[j]; - client_args->read_bytes = ts->read_strategy; + client_args->read_bytes = strategy->read_strategy; client_args->write_bytes = blocking_write_bytes; - client_args->setup = ts->setup; - client_args->msg_size = (size_t)msg_size; - client_args->strategy_name = ts->name; - server_args->read_bytes = ts->read_strategy; + client_args->setup = strategy->setup; + client_args->msg_size = msg_size; + client_args->strategy_name = strategy->name; + server_args->read_bytes = strategy->read_strategy; server_args->write_bytes = blocking_write_bytes; - server_args->setup = ts->setup; - server_args->msg_size = (size_t)msg_size; - server_args->strategy_name = ts->name; + server_args->setup = strategy->setup; + server_args->msg_size = msg_size; + server_args->strategy_name = strategy->name; error = run_benchmark(socket_type, client_args, server_args); if (error < 0) { return error; @@ -624,7 +626,7 @@ int main(int argc, char **argv) { char *read_strategy = NULL; char *socket_type = NULL; size_t i; - const test_strategy *ts = NULL; + const test_strategy *strategy = NULL; int error = 0; gpr_cmdline *cmdline = @@ -644,7 +646,7 @@ int main(int argc, char **argv) { if (read_strategy == NULL) { gpr_log(GPR_INFO, "No strategy specified, running all benchmarks"); - return run_all_benchmarks(msg_size); + return run_all_benchmarks((size_t)msg_size); } if (socket_type == NULL) { @@ -658,22 +660,22 @@ int main(int argc, char **argv) { for (i = 0; i < GPR_ARRAY_SIZE(test_strategies); ++i) { if (strcmp(test_strategies[i].name, read_strategy) == 0) { - ts = &test_strategies[i]; + strategy = &test_strategies[i]; } } - if (ts == NULL) { + if (strategy == NULL) { fprintf(stderr, "Invalid read strategy %s\n", read_strategy); return -1; } - client_args->read_bytes = ts->read_strategy; + client_args->read_bytes = strategy->read_strategy; client_args->write_bytes = blocking_write_bytes; - client_args->setup = ts->setup; + client_args->setup = strategy->setup; client_args->msg_size = (size_t)msg_size; client_args->strategy_name = read_strategy; - server_args->read_bytes = ts->read_strategy; + server_args->read_bytes = strategy->read_strategy; server_args->write_bytes = blocking_write_bytes; - server_args->setup = ts->setup; + server_args->setup = strategy->setup; server_args->msg_size = (size_t)msg_size; server_args->strategy_name = read_strategy; diff --git a/test/core/surface/byte_buffer_reader_test.c b/test/core/surface/byte_buffer_reader_test.c index c654f80f71..7f9cd6b62b 100644 --- a/test/core/surface/byte_buffer_reader_test.c +++ b/test/core/surface/byte_buffer_reader_test.c @@ -185,8 +185,8 @@ static void test_byte_buffer_from_reader(void) { } static void test_readall(void) { - const char* lotsa_as[512]; - const char* lotsa_bs[1024]; + char* lotsa_as[512]; + char* lotsa_bs[1024]; gpr_slice slices[2]; grpc_byte_buffer *buffer; grpc_byte_buffer_reader reader; |