Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Do not send BDP pings if there is no receive side activity.

Browse Source
pull/22997/head
Yash Tibrewal 5 years ago
parent c79bef55ee
commit 09904691e0
5 changed files with 270 additions and 50 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 19
      src/core/ext/transport/chttp2/transport/chttp2_transport.cc
  2. 4
      src/core/ext/transport/chttp2/transport/internal.h
  3. 10
      src/core/ext/transport/chttp2/transport/parsing.cc
  4. 3
      src/core/lib/transport/bdp_estimator.h
  5. 284
      test/core/transport/chttp2/too_many_pings_test.cc

19
src/core/ext/transport/chttp2/transport/chttp2_transport.cc
Unescape View File

@ -140,7 +140,6 @@ static void post_destructive_reclaimer(grpc_chttp2_transport* t);
static void close_transport_locked(grpc_chttp2_transport* t, grpc_error* error);
static void end_all_the_calls(grpc_chttp2_transport* t, grpc_error* error);
static void schedule_bdp_ping_locked(grpc_chttp2_transport* t);
static void start_bdp_ping(void* tp, grpc_error* error);
static void finish_bdp_ping(void* tp, grpc_error* error);
static void start_bdp_ping_locked(void* tp, grpc_error* error);
@ -511,8 +510,7 @@ grpc_chttp2_transport::grpc_chttp2_transport(
init_keepalive_pings_if_enabled(this);
if (enable_bdp) {
GRPC_CHTTP2_REF_TRANSPORT(this, "bdp_ping");
schedule_bdp_ping_locked(this);
bdp_ping_blocked = true;
grpc_chttp2_act_on_flowctl_action(flow_control->PeriodicUpdate(), this,
nullptr);
}
@ -2496,11 +2494,6 @@ static void read_action_locked(void* tp, grpc_error* error) {
grpc_error* errors[3] = {GRPC_ERROR_REF(error), GRPC_ERROR_NONE,
GRPC_ERROR_NONE};
for (; i < t->read_buffer.count && errors[1] == GRPC_ERROR_NONE; i++) {
grpc_core::BdpEstimator* bdp_est = t->flow_control->bdp_estimator();
if (bdp_est) {
bdp_est->AddIncomingBytes(
static_cast<int64_t> GRPC_SLICE_LENGTH(t->read_buffer.slices[i]));
}
errors[1] = grpc_chttp2_perform_read(t, t->read_buffer.slices[i]);
}
if (errors[1] != GRPC_ERROR_NONE) {
@ -2579,7 +2572,7 @@ static void continue_read_action_locked(grpc_chttp2_transport* t) {
// t is reffed prior to calling the first time, and once the callback chain
// that kicks off finishes, it's unreffed
static void schedule_bdp_ping_locked(grpc_chttp2_transport* t) {
void schedule_bdp_ping_locked(grpc_chttp2_transport* t) {
t->flow_control->bdp_estimator()->SchedulePing();
send_ping_locked(
t,
@ -2666,7 +2659,13 @@ static void next_bdp_ping_timer_expired_locked(void* tp, grpc_error* error) {
GRPC_CHTTP2_UNREF_TRANSPORT(t, "bdp_ping");
return;
}
schedule_bdp_ping_locked(t);
if (t->flow_control->bdp_estimator()->accumulator() == 0) {
// Block the bdp ping till we receive more data.
GRPC_CHTTP2_UNREF_TRANSPORT(t, "bdp_ping");
t->bdp_ping_blocked = true;
} else {
schedule_bdp_ping_locked(t);
}
}
void grpc_chttp2_config_default_keepalive_args(grpc_channel_args* args,

4
src/core/ext/transport/chttp2/transport/internal.h
Unescape View File

@ -430,6 +430,8 @@ struct grpc_chttp2_transport {
grpc_chttp2_write_cb* write_cb_pool = nullptr;
/* bdp estimator */
bool bdp_ping_blocked =
false; /* Is the BDP blocked due to not receiving any data? */
grpc_closure next_bdp_ping_timer_expired_locked;
grpc_closure start_bdp_ping_locked;
grpc_closure finish_bdp_ping_locked;
@ -874,4 +876,6 @@ void grpc_chttp2_config_default_keepalive_args(grpc_channel_args* args,
void grpc_chttp2_retry_initiate_ping(void* tp, grpc_error* error);
void schedule_bdp_ping_locked(grpc_chttp2_transport* t);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */

10
src/core/ext/transport/chttp2/transport/parsing.cc
Unescape View File

@ -334,6 +334,16 @@ void grpc_chttp2_parsing_become_skip_parser(grpc_chttp2_transport* t) {
}
static grpc_error* init_data_frame_parser(grpc_chttp2_transport* t) {
// Update BDP accounting since we have received a data frame.
grpc_core::BdpEstimator* bdp_est = t->flow_control->bdp_estimator();
if (bdp_est) {
if (t->bdp_ping_blocked) {
t->bdp_ping_blocked = false;
GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
schedule_bdp_ping_locked(t);
}
bdp_est->AddIncomingBytes(t->incoming_frame_size);
}
grpc_chttp2_stream* s =
grpc_chttp2_parsing_lookup_stream(t, t->incoming_stream_id);
grpc_error* err = GRPC_ERROR_NONE;

3
src/core/lib/transport/bdp_estimator.h
Unescape View File

@ -68,13 +68,14 @@ class BdpEstimator {
}
GPR_ASSERT(ping_state_ == PingState::SCHEDULED);
ping_state_ = PingState::STARTED;
accumulator_ = 0;
ping_start_time_ = gpr_now(GPR_CLOCK_MONOTONIC);
}
// Completes a previously started ping, returns when to schedule the next one
grpc_millis CompletePing();
int64_t accumulator() { return accumulator_; }
private:
enum class PingState { UNSCHEDULED, SCHEDULED, STARTED };

284
test/core/transport/chttp2/too_many_pings_test.cc
Unescape View File

@ -245,6 +245,43 @@ grpc_status_code PerformWaitingCall(grpc_channel* channel, grpc_server* server,
return status;
}
// Shuts down and destroys the server.
void ServerShutdownAndDestroy(grpc_server* server, grpc_completion_queue* cq) {
// Shutdown and destroy server
grpc_server_shutdown_and_notify(server, cq, (void*)(1000));
while (grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),
nullptr)
.tag != (void*)(1000))
;
grpc_server_destroy(server);
}
void VerifyChannelReady(grpc_channel* channel, grpc_completion_queue* cq) {
grpc_connectivity_state state =
grpc_channel_check_connectivity_state(channel, 1 /* try_to_connect */);
while (state != GRPC_CHANNEL_READY) {
grpc_channel_watch_connectivity_state(
channel, state, grpc_timeout_seconds_to_deadline(5), cq, nullptr);
grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5),
nullptr);
state = grpc_channel_check_connectivity_state(channel, 0);
}
}
void VerifyChannelDisconnected(grpc_channel* channel,
grpc_completion_queue* cq) {
// Verify channel gets disconnected. Use a ping to make sure that clients
// tries sending/receiving bytes if the channel is connected.
grpc_channel_ping(channel, cq, (void*)(2000), nullptr);
grpc_event ev = grpc_completion_queue_next(
cq, grpc_timeout_seconds_to_deadline(5), nullptr);
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
GPR_ASSERT(ev.tag == (void*)(2000));
GPR_ASSERT(ev.success == 0);
GPR_ASSERT(grpc_channel_check_connectivity_state(channel, 0) !=
GRPC_CHANNEL_READY);
}
class KeepaliveThrottlingTest : public ::testing::Test {
protected:
// Starts the server and makes sure that the channel is able to get connected.
@ -267,45 +304,6 @@ class KeepaliveThrottlingTest : public ::testing::Test {
grpc_server_start(server);
return server;
}
// Shuts down and destroys the server. Also, makes sure that the channel
// receives the disconnection event.
void ServerShutdownAndDestroy(grpc_server* server,
grpc_completion_queue* cq) {
// Shutdown and destroy server
grpc_server_shutdown_and_notify(server, cq, (void*)(1000));
while (grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),
nullptr)
.tag != (void*)(1000))
;
grpc_server_destroy(server);
}
void VerifyChannelReady(grpc_channel* channel, grpc_completion_queue* cq) {
grpc_connectivity_state state =
grpc_channel_check_connectivity_state(channel, 1 /* try_to_connect */);
while (state != GRPC_CHANNEL_READY) {
grpc_channel_watch_connectivity_state(
channel, state, grpc_timeout_seconds_to_deadline(5), cq, nullptr);
grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5),
nullptr);
state = grpc_channel_check_connectivity_state(channel, 0);
}
}
void VerifyChannelDisconnected(grpc_channel* channel,
grpc_completion_queue* cq) {
// Verify channel gets disconnected. Use a ping to make sure that clients
// tries sending/receiving bytes if the channel is connected.
grpc_channel_ping(channel, cq, (void*)(2000), nullptr);
grpc_event ev = grpc_completion_queue_next(
cq, grpc_timeout_seconds_to_deadline(5), nullptr);
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
GPR_ASSERT(ev.tag == (void*)(2000));
GPR_ASSERT(ev.success == 0);
GPR_ASSERT(grpc_channel_check_connectivity_state(channel, 0) !=
GRPC_CHANNEL_READY);
}
};
TEST_F(KeepaliveThrottlingTest, KeepaliveThrottlingMultipleChannels) {
@ -528,6 +526,214 @@ TEST_F(KeepaliveThrottlingTest,
grpc_completion_queue_destroy(cq);
}
// Perform a simple RPC where the client makes a request expecting a response
// with payload.
void PerformCallWithResponsePayload(grpc_channel* channel, grpc_server* server,
grpc_completion_queue* cq) {
grpc_slice response_payload_slice = grpc_slice_from_static_string("hello");
grpc_call* c;
grpc_call* s;
grpc_byte_buffer* response_payload =
grpc_raw_byte_buffer_create(&response_payload_slice, 1);
cq_verifier* cqv = cq_verifier_create(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_byte_buffer* response_payload_recv = nullptr;
grpc_call_details call_details;
grpc_status_code status;
grpc_call_error error;
grpc_slice details;
int was_cancelled = 2;
gpr_timespec deadline = grpc_timeout_seconds_to_deadline(60);
c = grpc_channel_create_call(channel, nullptr, GRPC_PROPAGATE_DEFAULTS, 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_SEND_CLOSE_FROM_CLIENT;
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_MESSAGE;
op->data.recv_message.recv_message = &response_payload_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(server, &s, &call_details,
&request_metadata_recv, cq, cq, tag(101));
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(101), 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++;
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_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = response_payload;
op->flags = 0;
op->reserved = nullptr;
op++;
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_OK;
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(103),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
GPR_ASSERT(was_cancelled == 0);
GPR_ASSERT(
byte_buffer_eq_slice(response_payload_recv, response_payload_slice));
grpc_slice_unref(details);
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_call_unref(c);
grpc_call_unref(s);
cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(response_payload);
grpc_byte_buffer_destroy(response_payload_recv);
}
TEST(TooManyPings, BdpPingNotSentWithoutReceiveSideActivity) {
grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
// create the server
std::string server_address =
grpc_core::JoinHostPort("localhost", grpc_pick_unused_port_or_die());
grpc_arg server_args[] = {
grpc_channel_arg_integer_create(
const_cast<char*>(
GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS),
60 * 1000),
grpc_channel_arg_integer_create(
const_cast<char*>(GRPC_ARG_HTTP2_MAX_PING_STRIKES), 1)};
grpc_channel_args server_channel_args = {GPR_ARRAY_SIZE(server_args),
server_args};
grpc_server* server = grpc_server_create(&server_channel_args, nullptr);
grpc_server_register_completion_queue(server, cq, nullptr);
GPR_ASSERT(
grpc_server_add_insecure_http2_port(server, server_address.c_str()));
grpc_server_start(server);
// create the channel (bdp pings are enabled by default)
grpc_arg client_args[] = {
grpc_channel_arg_integer_create(
const_cast<char*>(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA), 0),
grpc_channel_arg_integer_create(
const_cast<char*>(
GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS),
0),
grpc_channel_arg_integer_create(
const_cast<char*>(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS), 1)};
grpc_channel_args client_channel_args = {GPR_ARRAY_SIZE(client_args),
client_args};
grpc_channel* channel = grpc_insecure_channel_create(
server_address.c_str(), &client_channel_args, nullptr);
VerifyChannelReady(channel, cq);
cq_verifier* cqv = cq_verifier_create(cq);
cq_verify_empty_timeout(cqv, 1);
// Channel should be able to send two pings without disconnect if there was no
// BDP sent.
grpc_channel_ping(channel, cq, tag(1), nullptr);
CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
cq_verify(cqv, 5);
// Second ping
grpc_channel_ping(channel, cq, tag(2), nullptr);
CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
cq_verify(cqv, 5);
ASSERT_EQ(grpc_channel_check_connectivity_state(channel, 0),
GRPC_CHANNEL_READY);
PerformCallWithResponsePayload(channel, server, cq);
// The call with a response payload should have triggered a BDP ping.
// Send two more pings to verify. The second ping should cause a disconnect.
// If BDP was not sent, the second ping would not cause a disconnect.
grpc_channel_ping(channel, cq, tag(3), nullptr);
CQ_EXPECT_COMPLETION(cqv, tag(3), 1);
cq_verify(cqv, 5);
// Second ping
grpc_channel_ping(channel, cq, tag(4), nullptr);
CQ_EXPECT_COMPLETION(cqv, tag(4), 1);
cq_verify(cqv, 5);
ASSERT_NE(grpc_channel_check_connectivity_state(channel, 0),
GRPC_CHANNEL_READY);
cq_verifier_destroy(cqv);
// shutdown and destroy the client and server
ServerShutdownAndDestroy(server, cq);
grpc_channel_destroy(channel);
grpc_completion_queue_shutdown(cq);
while (grpc_completion_queue_next(cq, gpr_inf_future(GPR_CLOCK_REALTIME),
nullptr)
.type != GRPC_QUEUE_SHUTDOWN)
;
grpc_completion_queue_destroy(cq);
}
} // namespace
int main(int argc, char** argv) {

Write Preview
Loading…
Cancel
Save