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

Remove closure list scheduling from combiners

Browse Source
reviewable/pr20542/r1
Yash Tibrewal 5 years ago
parent c596449430
commit 5643aa6cb3
4 changed files with 65 additions and 39 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. 91
      src/core/ext/transport/chttp2/transport/chttp2_transport.cc
  2. 2
      src/core/ext/transport/chttp2/transport/writing.cc
  3. 10
      src/core/lib/iomgr/combiner.cc
  4. 1
      src/core/lib/iomgr/combiner.h

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

@ -139,6 +139,8 @@ 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);
static void finish_bdp_ping_locked(void* tp, grpc_error* error);
static void next_bdp_ping_timer_expired(void* tp, grpc_error* error);
@ -153,6 +155,8 @@ static void retry_initiate_ping_locked(void* tp, grpc_error* error);
/** keepalive-relevant functions */
static void init_keepalive_ping(void* arg, grpc_error* error);
static void init_keepalive_ping_locked(void* arg, grpc_error* error);
static void start_keepalive_ping(void* arg, grpc_error* error);
static void finish_keepalive_ping(void* arg, grpc_error* error);
static void start_keepalive_ping_locked(void* arg, grpc_error* error);
static void finish_keepalive_ping_locked(void* arg, grpc_error* error);
static void keepalive_watchdog_fired(void* arg, grpc_error* error);
@ -385,18 +389,6 @@ static bool read_channel_args(grpc_chttp2_transport* t,
return enable_bdp;
}
static void init_transport_closures(grpc_chttp2_transport* t) {
GRPC_CLOSURE_INIT(&t->start_bdp_ping_locked, start_bdp_ping_locked, t,
nullptr);
GRPC_CLOSURE_INIT(&t->finish_bdp_ping_locked, finish_bdp_ping_locked, t,
nullptr);
GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked,
start_keepalive_ping_locked, t, nullptr);
GRPC_CLOSURE_INIT(&t->finish_keepalive_ping_locked,
finish_keepalive_ping_locked, t, nullptr);
}
static void init_transport_keepalive_settings(grpc_chttp2_transport* t) {
if (t->is_client) {
t->keepalive_time = g_default_client_keepalive_time_ms == INT_MAX
@ -486,8 +478,6 @@ grpc_chttp2_transport::grpc_chttp2_transport(
grpc_chttp2_hpack_parser_init(&hpack_parser);
grpc_chttp2_goaway_parser_init(&goaway_parser);
init_transport_closures(this);
/* configure http2 the way we like it */
if (is_client) {
queue_setting_update(this, GRPC_CHTTP2_SETTINGS_ENABLE_PUSH, 0);
@ -1680,7 +1670,7 @@ static void cancel_pings(grpc_chttp2_transport* t, grpc_error* error) {
GPR_ASSERT(error != GRPC_ERROR_NONE);
for (size_t j = 0; j < GRPC_CHTTP2_PCL_COUNT; j++) {
grpc_closure_list_fail_all(&pq->lists[j], GRPC_ERROR_REF(error));
t->combiner->Run(&pq->lists[j]);
GRPC_CLOSURE_LIST_SCHED(&pq->lists[j]);
}
GRPC_ERROR_UNREF(error);
}
@ -1706,24 +1696,38 @@ static void send_ping_locked(grpc_chttp2_transport* t,
*/
static void send_keepalive_ping_locked(grpc_chttp2_transport* t) {
if (t->closed_with_error != GRPC_ERROR_NONE) {
GRPC_CLOSURE_RUN(&t->start_keepalive_ping_locked,
GRPC_ERROR_REF(t->closed_with_error));
GRPC_CLOSURE_RUN(&t->finish_keepalive_ping_locked,
t->combiner->Run(GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked,
start_keepalive_ping_locked, t, nullptr),
GRPC_ERROR_REF(t->closed_with_error));
t->combiner->Run(
GRPC_CLOSURE_INIT(&t->finish_keepalive_ping_locked,
finish_keepalive_ping_locked, t, nullptr),
GRPC_ERROR_REF(t->closed_with_error));
return;
}
grpc_chttp2_ping_queue* pq = &t->ping_queue;
if (!grpc_closure_list_empty(pq->lists[GRPC_CHTTP2_PCL_INFLIGHT])) {
/* There is a ping in flight. Add yourself to the inflight closure list. */
GRPC_CLOSURE_RUN(&t->start_keepalive_ping_locked, GRPC_ERROR_NONE);
grpc_closure_list_append(&pq->lists[GRPC_CHTTP2_PCL_INFLIGHT],
&t->finish_keepalive_ping_locked, GRPC_ERROR_NONE);
t->combiner->Run(GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked,
start_keepalive_ping_locked, t, nullptr),
GRPC_ERROR_REF(t->closed_with_error));
grpc_closure_list_append(
&pq->lists[GRPC_CHTTP2_PCL_INFLIGHT],
GRPC_CLOSURE_INIT(&t->finish_keepalive_ping_locked,
finish_keepalive_ping, t, grpc_schedule_on_exec_ctx),
GRPC_ERROR_NONE);
return;
}
grpc_closure_list_append(&pq->lists[GRPC_CHTTP2_PCL_INITIATE],
&t->start_keepalive_ping_locked, GRPC_ERROR_NONE);
grpc_closure_list_append(&pq->lists[GRPC_CHTTP2_PCL_NEXT],
&t->finish_keepalive_ping_locked, GRPC_ERROR_NONE);
grpc_closure_list_append(
&pq->lists[GRPC_CHTTP2_PCL_INITIATE],
GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked, start_keepalive_ping,
t, grpc_schedule_on_exec_ctx),
GRPC_ERROR_NONE);
grpc_closure_list_append(
&pq->lists[GRPC_CHTTP2_PCL_NEXT],
GRPC_CLOSURE_INIT(&t->finish_keepalive_ping_locked, finish_keepalive_ping,
t, grpc_schedule_on_exec_ctx),
GRPC_ERROR_NONE);
}
void grpc_chttp2_retry_initiate_ping(void* tp, grpc_error* error) {
@ -1750,7 +1754,7 @@ void grpc_chttp2_ack_ping(grpc_chttp2_transport* t, uint64_t id) {
gpr_free(from);
return;
}
t->combiner->Run(&pq->lists[GRPC_CHTTP2_PCL_INFLIGHT]);
GRPC_CLOSURE_LIST_SCHED(&pq->lists[GRPC_CHTTP2_PCL_INFLIGHT]);
if (!grpc_closure_list_empty(pq->lists[GRPC_CHTTP2_PCL_NEXT])) {
grpc_chttp2_initiate_write(t, GRPC_CHTTP2_INITIATE_WRITE_CONTINUE_PINGS);
}
@ -2598,7 +2602,19 @@ static void continue_read_action_locked(grpc_chttp2_transport* t) {
// that kicks off finishes, it's unreffed
static void schedule_bdp_ping_locked(grpc_chttp2_transport* t) {
t->flow_control->bdp_estimator()->SchedulePing();
send_ping_locked(t, &t->start_bdp_ping_locked, &t->finish_bdp_ping_locked);
send_ping_locked(
t,
GRPC_CLOSURE_INIT(&t->start_bdp_ping_locked, start_bdp_ping, t,
grpc_schedule_on_exec_ctx),
GRPC_CLOSURE_INIT(&t->finish_bdp_ping_locked, finish_bdp_ping, t,
grpc_schedule_on_exec_ctx));
}
static void start_bdp_ping(void* tp, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(tp);
t->combiner->Run(GRPC_CLOSURE_INIT(&t->start_bdp_ping_locked,
start_bdp_ping_locked, t, nullptr),
GRPC_ERROR_REF(error));
}
static void start_bdp_ping_locked(void* tp, grpc_error* error) {
@ -2617,6 +2633,13 @@ static void start_bdp_ping_locked(void* tp, grpc_error* error) {
t->flow_control->bdp_estimator()->StartPing();
}
static void finish_bdp_ping(void* tp, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(tp);
t->combiner->Run(GRPC_CLOSURE_INIT(&t->finish_bdp_ping_locked,
finish_bdp_ping_locked, t, nullptr),
GRPC_ERROR_REF(error));
}
static void finish_bdp_ping_locked(void* tp, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(tp);
if (GRPC_TRACE_FLAG_ENABLED(grpc_http_trace)) {
@ -2764,6 +2787,13 @@ static void init_keepalive_ping_locked(void* arg, grpc_error* error) {
GRPC_CHTTP2_UNREF_TRANSPORT(t, "init keepalive ping");
}
static void start_keepalive_ping(void* arg, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(arg);
t->combiner->Run(GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked,
start_keepalive_ping_locked, t, nullptr),
GRPC_ERROR_REF(error));
}
static void start_keepalive_ping_locked(void* arg, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(arg);
if (error != GRPC_ERROR_NONE) {
@ -2783,6 +2813,13 @@ static void start_keepalive_ping_locked(void* arg, grpc_error* error) {
&t->keepalive_watchdog_fired_locked);
}
static void finish_keepalive_ping(void* arg, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(arg);
t->combiner->Run(GRPC_CLOSURE_INIT(&t->finish_keepalive_ping_locked,
finish_keepalive_ping_locked, t, nullptr),
GRPC_ERROR_REF(error));
}
static void finish_keepalive_ping_locked(void* arg, grpc_error* error) {
grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(arg);
if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) {

2
src/core/ext/transport/chttp2/transport/writing.cc
Unescape View File

@ -107,7 +107,7 @@ static void maybe_initiate_ping(grpc_chttp2_transport* t) {
pq->inflight_id = t->ping_ctr;
t->ping_ctr++;
t->combiner->Run(&pq->lists[GRPC_CHTTP2_PCL_INITIATE]);
GRPC_CLOSURE_LIST_SCHED(&pq->lists[GRPC_CHTTP2_PCL_INITIATE]);
grpc_closure_list_move(&pq->lists[GRPC_CHTTP2_PCL_NEXT],
&pq->lists[GRPC_CHTTP2_PCL_INFLIGHT]);
grpc_slice_buffer_add(&t->outbuf,

10
src/core/lib/iomgr/combiner.cc
Unescape View File

@ -337,16 +337,6 @@ void Combiner::Run(grpc_closure* closure, grpc_error* error) {
combiner_exec(this, closure, error);
}
void Combiner::Run(grpc_closure_list* list) {
grpc_closure* c = list->head;
while (c != nullptr) {
grpc_closure* next = c->next_data.next;
Run(c, c->error_data.error);
c = next;
}
list->head = list->tail = nullptr;
}
void Combiner::FinallyRun(grpc_closure* closure, grpc_error* error) {
GPR_ASSERT(closure->scheduler == nullptr ||
closure->scheduler ==

1
src/core/lib/iomgr/combiner.h
Unescape View File

@ -31,7 +31,6 @@ namespace grpc_core {
class Combiner {
public:
void Run(grpc_closure* closure, grpc_error* error);
void Run(grpc_closure_list* list);
void FinallyRun(grpc_closure* closure, grpc_error* error);
Combiner* next_combiner_on_this_exec_ctx = nullptr;
grpc_closure_scheduler scheduler;

Write Preview
Loading…
Cancel
Save