Merge pull request #17893 from yashykt/keepaliveopt

Reset keepalive timer on reading bytes
pull/17715/head
Yash Tibrewal 6 years ago committed by GitHub
commit f4079fe185
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 4
      src/core/ext/transport/chttp2/transport/chttp2_transport.cc
  2. 215
      test/core/end2end/tests/keepalive_timeout.cc

@ -2558,6 +2558,10 @@ static void read_action_locked(void* tp, grpc_error* error) {
} else if (t->closed_with_error == GRPC_ERROR_NONE) {
keep_reading = true;
GRPC_CHTTP2_REF_TRANSPORT(t, "keep_reading");
/* Since we have read a byte, reset the keepalive timer */
if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING) {
grpc_timer_cancel(&t->keepalive_ping_timer);
}
}
grpc_slice_buffer_reset_and_unref_internal(&t->read_buffer);

@ -220,8 +220,223 @@ static void test_keepalive_timeout(grpc_end2end_test_config config) {
config.tear_down_data(&f);
}
/* Verify that reads reset the keepalive ping timer. The client sends 30 pings
* with a sleep of 10ms in between. It has a configured keepalive timer of
* 200ms. In the success case, each ping ack should reset the keepalive timer so
* that the keepalive ping is never sent. */
static void test_read_delays_keepalive(grpc_end2end_test_config config) {
char* poller = gpr_getenv("GRPC_POLL_STRATEGY");
/* It is hard to get the timing right for the polling engines poll and poll-cv
*/
if (poller != nullptr &&
(0 == strcmp(poller, "poll-cv") || 0 == strcmp(poller, "poll"))) {
gpr_free(poller);
return;
}
gpr_free(poller);
const int kPingIntervalMS = 100;
grpc_arg keepalive_arg_elems[3];
keepalive_arg_elems[0].type = GRPC_ARG_INTEGER;
keepalive_arg_elems[0].key = const_cast<char*>(GRPC_ARG_KEEPALIVE_TIME_MS);
keepalive_arg_elems[0].value.integer = 20 * kPingIntervalMS;
keepalive_arg_elems[1].type = GRPC_ARG_INTEGER;
keepalive_arg_elems[1].key = const_cast<char*>(GRPC_ARG_KEEPALIVE_TIMEOUT_MS);
keepalive_arg_elems[1].value.integer = 0;
keepalive_arg_elems[2].type = GRPC_ARG_INTEGER;
keepalive_arg_elems[2].key = const_cast<char*>(GRPC_ARG_HTTP2_BDP_PROBE);
keepalive_arg_elems[2].value.integer = 0;
grpc_channel_args keepalive_args = {GPR_ARRAY_SIZE(keepalive_arg_elems),
keepalive_arg_elems};
grpc_end2end_test_fixture f = begin_test(config, "test_read_delays_keepalive",
&keepalive_args, nullptr);
/* Disable ping ack to trigger the keepalive timeout */
grpc_set_disable_ping_ack(true);
grpc_call* c;
grpc_call* s;
cq_verifier* cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op* op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_status_code status;
grpc_call_error error;
grpc_slice details;
int was_cancelled = 2;
grpc_byte_buffer* request_payload;
grpc_byte_buffer* request_payload_recv;
grpc_byte_buffer* response_payload;
grpc_byte_buffer* response_payload_recv;
int i;
grpc_slice request_payload_slice =
grpc_slice_from_copied_string("hello world");
grpc_slice response_payload_slice =
grpc_slice_from_copied_string("hello you");
gpr_timespec deadline = five_seconds_from_now();
c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
grpc_slice_from_static_string("/foo"), nullptr,
deadline, nullptr);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
op->flags = 0;
op->reserved = nullptr;
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->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
error =
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq, f.cq, tag(100));
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(100), 1);
cq_verify(cqv);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(101),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
for (i = 0; i < 30; i++) {
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = request_payload;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message.recv_message = &response_payload_recv;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(2),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message.recv_message = &request_payload_recv;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(102), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(102), 1);
cq_verify(cqv);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = response_payload;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(103), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
cq_verify(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
grpc_byte_buffer_destroy(request_payload_recv);
grpc_byte_buffer_destroy(response_payload_recv);
/* Sleep for a short interval to check if the client sends any pings */
gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(kPingIntervalMS));
}
grpc_slice_unref(request_payload_slice);
grpc_slice_unref(response_payload_slice);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(3),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
memset(ops, 0, sizeof(ops));
op = ops;
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
op->data.send_status_from_server.trailing_metadata_count = 0;
op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED;
grpc_slice status_details = grpc_slice_from_static_string("xyz");
op->data.send_status_from_server.status_details = &status_details;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(104),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
CQ_EXPECT_COMPLETION(cqv, tag(3), 1);
CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
CQ_EXPECT_COMPLETION(cqv, tag(104), 1);
cq_verify(cqv);
grpc_call_unref(c);
grpc_call_unref(s);
cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_slice_unref(details);
end_test(&f);
config.tear_down_data(&f);
}
void keepalive_timeout(grpc_end2end_test_config config) {
test_keepalive_timeout(config);
test_read_delays_keepalive(config);
}
void keepalive_timeout_pre_init(void) {}

Loading…
Cancel
Save