diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 3ee5e976f8492ec59c930f43c728fcce37b689f6..ddbc656b96dedf1801ca7cce09d646ddd7e8640f 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -265,7 +265,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
.gain_d = 0,
.initial_control_value = log2(DEFAULT_WINDOW),
.min_control_value = -1,
- .max_control_value = 22,
+ .max_control_value = 25,
.integral_range = 10});
grpc_chttp2_goaway_parser_init(&t->goaway_parser);
@@ -569,6 +569,14 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
GRPC_ERROR_UNREF(s->read_closed_error);
GRPC_ERROR_UNREF(s->write_closed_error);
+ if (s->incoming_window_delta > 0) {
+ GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA(
+ "destroy", t, s, s->incoming_window_delta);
+ } else if (s->incoming_window_delta < 0) {
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA(
+ "destroy", t, s, -s->incoming_window_delta);
+ }
+
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "stream");
GPR_TIMER_END("destroy_stream", 0);
@@ -1801,13 +1809,13 @@ static void update_bdp(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
if (delta == 0 || (bdp != 0 && delta > -1024 && delta < 1024)) {
return;
}
+ if (grpc_bdp_estimator_trace) {
+ gpr_log(GPR_DEBUG, "%s: update initial window size to %d", t->peer_string,
+ (int)bdp);
+ }
push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, bdp);
}
-/*******************************************************************************
- * INPUT PROCESSING - PARSING
- */
-
static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t) {
grpc_http_parser parser;
@@ -2054,8 +2062,8 @@ static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx,
(int64_t)have_already) {
write_type = GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED;
}
- GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, incoming_window_delta,
- add_max_recv_bytes);
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA("op", t, s,
+ add_max_recv_bytes);
GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, announce_window,
add_max_recv_bytes);
if ((int64_t)s->incoming_window_delta + (int64_t)initial_window_size -
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index 075d421dd46ea32aaeb1d15ac6332c77bcbe8585..5d41f4bfda652dc000b957e090e3577e6d265ebd 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -271,10 +271,6 @@ struct grpc_chttp2_transport {
/** data to write next write */
grpc_slice_buffer qbuf;
- /** window available to announce to peer */
- int64_t announce_incoming_window;
- /** how much window would we like to have for incoming_window */
- uint32_t connection_window_target;
/** how much data are we willing to buffer when the WRITE_BUFFER_HINT is set?
*/
uint32_t write_buffer_size;
@@ -328,6 +324,16 @@ struct grpc_chttp2_transport {
/** window available for peer to send to us */
int64_t incoming_window;
+ /** calculating what we should give for incoming window:
+ we track the total amount of flow control over initial window size
+ across all streams: this is data that we want to receive right now (it
+ has an outstanding read)
+ and the total amount of flow control under initial window size across all
+ streams: this is data we've read early
+ we want to adjust incoming_window such that:
+ incoming_window = total_over - max(bdp - total_under, 0) */
+ int64_t stream_total_over_incoming_window;
+ int64_t stream_total_under_incoming_window;
/* deframing */
grpc_chttp2_deframe_transport_state deframe_state;
@@ -634,6 +640,44 @@ typedef enum {
GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, dst_context, 0, dst_context, dst_var, \
amount)
+#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE( \
+ phase, transport, dst_context) \
+ if (dst_context->incoming_window_delta < 0) { \
+ transport->stream_total_under_incoming_window += \
+ dst_context->incoming_window_delta; \
+ } else if (dst_context->incoming_window_delta > 0) { \
+ transport->stream_total_over_incoming_window -= \
+ dst_context->incoming_window_delta; \
+ }
+
+#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE( \
+ phase, transport, dst_context) \
+ if (dst_context->incoming_window_delta < 0) { \
+ transport->stream_total_under_incoming_window -= \
+ dst_context->incoming_window_delta; \
+ } else if (dst_context->incoming_window_delta > 0) { \
+ transport->stream_total_over_incoming_window += \
+ dst_context->incoming_window_delta; \
+ }
+
+#define GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA( \
+ phase, transport, dst_context, amount) \
+ GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport, \
+ dst_context); \
+ GRPC_CHTTP2_FLOW_DEBIT_STREAM(phase, transport, dst_context, \
+ incoming_window_delta, amount); \
+ GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \
+ dst_context);
+
+#define GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA( \
+ phase, transport, dst_context, amount) \
+ GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport, \
+ dst_context); \
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM(phase, transport, dst_context, \
+ incoming_window_delta, amount); \
+ GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \
+ dst_context);
+
#define GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, id, dst_context, \
dst_var, amount) \
do { \
@@ -752,4 +796,6 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t,
grpc_chttp2_stream *s, grpc_error *error);
+uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t);
+
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */
diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c
index 24bd93067b3bea3bfa0ee56166ab9a40a4fdf7d9..7ed00522c3ea3ee88afa1dee8479aaa1b7dc9881 100644
--- a/src/core/ext/transport/chttp2/transport/parsing.c
+++ b/src/core/ext/transport/chttp2/transport/parsing.c
@@ -376,15 +376,6 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
return err;
}
- uint32_t target_incoming_window = GPR_MAX(
- t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
- 1024);
- GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window,
- incoming_frame_size);
- if (t->incoming_window <= target_incoming_window / 2) {
- grpc_chttp2_initiate_write(exec_ctx, t, false, "flow_control");
- }
-
if (s != NULL) {
if (incoming_frame_size >
s->incoming_window_delta +
@@ -402,8 +393,8 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
return err;
}
- GRPC_CHTTP2_FLOW_DEBIT_STREAM("parse", t, s, incoming_window_delta,
- incoming_frame_size);
+ GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA("parse", t, s,
+ incoming_frame_size);
if ((int64_t)t->settings[GRPC_SENT_SETTINGS]
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] +
(int64_t)s->incoming_window_delta - (int64_t)s->announce_window <=
@@ -417,6 +408,13 @@ static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx,
s->received_bytes += incoming_frame_size;
}
+ uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t);
+ GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window,
+ incoming_frame_size);
+ if (t->incoming_window <= target_incoming_window / 2) {
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "flow_control");
+ }
+
return GRPC_ERROR_NONE;
}
diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c
index 05e6f59947a426ff3dca952cff1744df3e2e8d69..2b9d93cae7cc0b3db995d081c3684a57e4509891 100644
--- a/src/core/ext/transport/chttp2/transport/writing.c
+++ b/src/core/ext/transport/chttp2/transport/writing.c
@@ -152,6 +152,17 @@ static bool stream_ref_if_not_destroyed(gpr_refcount *r) {
return true;
}
+uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t) {
+ return (uint32_t)GPR_MAX(
+ (int64_t)((1u << 31) - 1),
+ t->stream_total_over_incoming_window +
+ (int64_t)GPR_MAX(
+ t->settings[GRPC_SENT_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] -
+ t->stream_total_under_incoming_window,
+ 0));
+}
+
bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t) {
grpc_chttp2_stream *s;
@@ -310,13 +321,12 @@ bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx,
/* if the grpc_chttp2_transport is ready to send a window update, do so here
also; 3/4 is a magic number that will likely get tuned soon */
- uint32_t target_incoming_window = GPR_MAX(
- t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
- 1024);
+ uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t);
uint32_t threshold_to_send_transport_window_update =
t->outbuf.count > 0 ? 3 * target_incoming_window / 4
: target_incoming_window / 2;
- if (t->incoming_window <= threshold_to_send_transport_window_update) {
+ if (t->incoming_window <= threshold_to_send_transport_window_update &&
+ t->incoming_window != target_incoming_window) {
maybe_initiate_ping(exec_ctx, t,
GRPC_CHTTP2_PING_BEFORE_TRANSPORT_WINDOW_UPDATE);
uint32_t announced = (uint32_t)GPR_CLAMP(
diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c
index 3f992977c086d0955be79605218da52e22cb27d4..bb185351a8e2b61238ca649853e4db4cff9209b9 100644
--- a/src/core/lib/channel/http_server_filter.c
+++ b/src/core/lib/channel/http_server_filter.c
@@ -198,14 +198,17 @@ static grpc_error *server_filter_incoming_metadata(grpc_exec_ctx *exec_ctx,
GRPC_ERROR_STR_KEY, ":path"));
}
- if (b->idx.named.host != NULL) {
+ if (b->idx.named.host != NULL && b->idx.named.authority == NULL) {
+ grpc_linked_mdelem *el = b->idx.named.host;
+ grpc_mdelem md = GRPC_MDELEM_REF(el->md);
+ grpc_metadata_batch_remove(exec_ctx, b, el);
add_error(
error_name, &error,
- grpc_metadata_batch_substitute(
- exec_ctx, b, b->idx.named.host,
- grpc_mdelem_from_slices(
- exec_ctx, GRPC_MDSTR_AUTHORITY,
- grpc_slice_ref_internal(GRPC_MDVALUE(b->idx.named.host->md)))));
+ grpc_metadata_batch_add_head(
+ exec_ctx, b, el, grpc_mdelem_from_slices(
+ exec_ctx, GRPC_MDSTR_AUTHORITY,
+ grpc_slice_ref_internal(GRPC_MDVALUE(md)))));
+ GRPC_MDELEM_UNREF(exec_ctx, md);
}
if (b->idx.named.authority == NULL) {
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index 51842fc208144478f32cec6cc9bde3f5164d7fbc..fc56843128ebe378265e925399429e83d30b91a2 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -1405,16 +1405,6 @@ static void pollset_destroy(grpc_pollset *pollset) {
gpr_mu_destroy(&pollset->po.mu);
}
-static void pollset_reset(grpc_pollset *pollset) {
- GPR_ASSERT(pollset->shutting_down);
- GPR_ASSERT(!pollset_has_workers(pollset));
- pollset->shutting_down = false;
- pollset->finish_shutdown_called = false;
- pollset->kicked_without_pollers = false;
- pollset->shutdown_done = NULL;
- GPR_ASSERT(pollset->po.pi == NULL);
-}
-
static bool maybe_do_workqueue_work(grpc_exec_ctx *exec_ctx,
polling_island *pi) {
if (gpr_mu_trylock(&pi->workqueue_read_mu)) {
@@ -1958,7 +1948,6 @@ static const grpc_event_engine_vtable vtable = {
.pollset_init = pollset_init,
.pollset_shutdown = pollset_shutdown,
- .pollset_reset = pollset_reset,
.pollset_destroy = pollset_destroy,
.pollset_work = pollset_work,
.pollset_kick = pollset_kick,
diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c
index ca129322194fdd8c052e63a652bc16befa87ac59..21eb62753efbedcff818fc45031d16f7ecad4572 100644
--- a/src/core/lib/iomgr/ev_poll_posix.c
+++ b/src/core/lib/iomgr/ev_poll_posix.c
@@ -815,16 +815,6 @@ static void pollset_destroy(grpc_pollset *pollset) {
gpr_mu_destroy(&pollset->mu);
}
-static void pollset_reset(grpc_pollset *pollset) {
- GPR_ASSERT(pollset->shutting_down);
- GPR_ASSERT(!pollset_has_workers(pollset));
- GPR_ASSERT(pollset->idle_jobs.head == pollset->idle_jobs.tail);
- GPR_ASSERT(pollset->fd_count == 0);
- pollset->shutting_down = 0;
- pollset->called_shutdown = 0;
- pollset->kicked_without_pollers = 0;
-}
-
static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_fd *fd) {
gpr_mu_lock(&pollset->mu);
@@ -1514,7 +1504,6 @@ static const grpc_event_engine_vtable vtable = {
.pollset_init = pollset_init,
.pollset_shutdown = pollset_shutdown,
- .pollset_reset = pollset_reset,
.pollset_destroy = pollset_destroy,
.pollset_work = pollset_work,
.pollset_kick = pollset_kick,
diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c
index 5bb55631d69dae86aada76c681b043fbd1bd8289..95b1d99d7577773612aa09f766e1e4e5a0aa5102 100644
--- a/src/core/lib/iomgr/ev_posix.c
+++ b/src/core/lib/iomgr/ev_posix.c
@@ -191,10 +191,6 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
g_event_engine->pollset_shutdown(exec_ctx, pollset, closure);
}
-void grpc_pollset_reset(grpc_pollset *pollset) {
- g_event_engine->pollset_reset(pollset);
-}
-
void grpc_pollset_destroy(grpc_pollset *pollset) {
g_event_engine->pollset_destroy(pollset);
}
diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h
index a589efdeece264c814fa1d49b405e6978b6d75d4..1aea7d61f32e65b1720a2cbb4df46abb6ef28ffe 100644
--- a/src/core/lib/iomgr/ev_posix.h
+++ b/src/core/lib/iomgr/ev_posix.h
@@ -64,7 +64,6 @@ typedef struct grpc_event_engine_vtable {
void (*pollset_init)(grpc_pollset *pollset, gpr_mu **mu);
void (*pollset_shutdown)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure);
- void (*pollset_reset)(grpc_pollset *pollset);
void (*pollset_destroy)(grpc_pollset *pollset);
grpc_error *(*pollset_work)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker, gpr_timespec now,
diff --git a/src/core/lib/iomgr/pollset.h b/src/core/lib/iomgr/pollset.h
index 8d9edc8406b887a82be4be750bf52b1732fb0b6e..c4b62a279042f2fa50a1a89418dbafea8b76f5bc 100644
--- a/src/core/lib/iomgr/pollset.h
+++ b/src/core/lib/iomgr/pollset.h
@@ -58,9 +58,6 @@ void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu);
* pollset's mutex must be held */
void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure);
-/** Reset the pollset to its initial state (perhaps with some cached objects);
- * must have been previously shutdown */
-void grpc_pollset_reset(grpc_pollset *pollset);
void grpc_pollset_destroy(grpc_pollset *pollset);
/* Do some work on a pollset.
diff --git a/src/core/lib/iomgr/pollset_uv.c b/src/core/lib/iomgr/pollset_uv.c
index ed3edeee9493ca2a8bc69a917b39f2ad818b580f..5847771108a50ee2641d283af89a2dd5a60da7fe 100644
--- a/src/core/lib/iomgr/pollset_uv.c
+++ b/src/core/lib/iomgr/pollset_uv.c
@@ -97,11 +97,6 @@ void grpc_pollset_destroy(grpc_pollset *pollset) {
}
}
-void grpc_pollset_reset(grpc_pollset *pollset) {
- GPR_ASSERT(pollset->shutting_down);
- pollset->shutting_down = 0;
-}
-
static void timer_run_cb(uv_timer_t *timer) {}
grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
diff --git a/src/core/lib/iomgr/pollset_windows.c b/src/core/lib/iomgr/pollset_windows.c
index 2a45e708df6a499c3b4fa9560cbff19b88e2df38..9be73a7ce8e92edbec0964fa7c03191e6849d81b 100644
--- a/src/core/lib/iomgr/pollset_windows.c
+++ b/src/core/lib/iomgr/pollset_windows.c
@@ -117,16 +117,6 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
void grpc_pollset_destroy(grpc_pollset *pollset) {}
-void grpc_pollset_reset(grpc_pollset *pollset) {
- GPR_ASSERT(pollset->shutting_down);
- GPR_ASSERT(
- !has_workers(&pollset->root_worker, GRPC_POLLSET_WORKER_LINK_POLLSET));
- pollset->shutting_down = 0;
- pollset->is_iocp_worker = 0;
- pollset->kicked_without_pollers = 0;
- pollset->on_shutdown = NULL;
-}
-
grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker_hdl,
gpr_timespec now, gpr_timespec deadline) {
diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c
index 429dbad7c7787896e02046d4f76bcb46bcec4b17..d6acd392c1a5f55f43fbe1f8a797833fcc3dcec8 100644
--- a/src/core/lib/surface/channel.c
+++ b/src/core/lib/surface/channel.c
@@ -128,7 +128,8 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
}
channel->default_authority = grpc_mdelem_from_slices(
exec_ctx, GRPC_MDSTR_AUTHORITY,
- grpc_slice_from_copied_string(args->args[i].value.string));
+ grpc_slice_intern(
+ grpc_slice_from_static_string(args->args[i].value.string)));
}
} else if (0 ==
strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) {
@@ -144,7 +145,8 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
} else {
channel->default_authority = grpc_mdelem_from_slices(
exec_ctx, GRPC_MDSTR_AUTHORITY,
- grpc_slice_from_copied_string(args->args[i].value.string));
+ grpc_slice_intern(
+ grpc_slice_from_static_string(args->args[i].value.string)));
}
}
} else if (0 == strcmp(args->args[i].key,
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index 1830842d00b0d93e606188637f2713c639aa8834..b80656aa8de508df9c00ad2b9c90678ed36ada8d 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -96,9 +96,6 @@ struct grpc_completion_queue {
#define POLLSET_FROM_CQ(cq) ((grpc_pollset *)(cq + 1))
#define CQ_FROM_POLLSET(ps) (((grpc_completion_queue *)ps) - 1)
-static gpr_mu g_freelist_mu;
-static grpc_completion_queue *g_freelist;
-
int grpc_cq_pluck_trace;
int grpc_cq_event_timeout_trace;
@@ -113,21 +110,6 @@ int grpc_cq_event_timeout_trace;
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
grpc_error *error);
-void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); }
-
-void grpc_cq_global_shutdown(void) {
- gpr_mu_destroy(&g_freelist_mu);
- while (g_freelist) {
- grpc_completion_queue *next = g_freelist->next_free;
- grpc_pollset_destroy(POLLSET_FROM_CQ(g_freelist));
-#ifndef NDEBUG
- gpr_free(g_freelist->outstanding_tags);
-#endif
- gpr_free(g_freelist);
- g_freelist = next;
- }
-}
-
grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
grpc_completion_queue *cc;
GPR_ASSERT(!reserved);
@@ -136,22 +118,12 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved));
- gpr_mu_lock(&g_freelist_mu);
- if (g_freelist == NULL) {
- gpr_mu_unlock(&g_freelist_mu);
-
- cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size());
- grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu);
+ cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size());
+ grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu);
#ifndef NDEBUG
- cc->outstanding_tags = NULL;
- cc->outstanding_tag_capacity = 0;
+ cc->outstanding_tags = NULL;
+ cc->outstanding_tag_capacity = 0;
#endif
- } else {
- cc = g_freelist;
- g_freelist = g_freelist->next_free;
- gpr_mu_unlock(&g_freelist_mu);
- /* pollset already initialized */
- }
/* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cc->pending_events, 1);
@@ -203,11 +175,11 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) {
#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head);
- grpc_pollset_reset(POLLSET_FROM_CQ(cc));
- gpr_mu_lock(&g_freelist_mu);
- cc->next_free = g_freelist;
- g_freelist = cc;
- gpr_mu_unlock(&g_freelist_mu);
+ grpc_pollset_destroy(POLLSET_FROM_CQ(cc));
+#ifndef NDEBUG
+ gpr_free(cc->outstanding_tags);
+#endif
+ gpr_free(cc);
}
}
diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h
index c1cafba5f2cb791cb1c8c1d8e465c72b6aad48b6..5d73dd7216e449ef741735fa7e1dcc4c957b2e42 100644
--- a/src/core/lib/surface/completion_queue.h
+++ b/src/core/lib/surface/completion_queue.h
@@ -99,7 +99,4 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
int grpc_cq_is_server_cq(grpc_completion_queue *cc);
-void grpc_cq_global_init(void);
-void grpc_cq_global_shutdown(void);
-
#endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */
diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c
index b338ac4c486b10e23b04aeff41355ab990af27f1..91bd014a0e5c6f0a8928466775ae91ef21e616d6 100644
--- a/src/core/lib/surface/init.c
+++ b/src/core/lib/surface/init.c
@@ -209,7 +209,6 @@ void grpc_init(void) {
grpc_iomgr_init();
grpc_executor_init();
gpr_timers_global_init();
- grpc_cq_global_init();
grpc_handshaker_factory_registry_init();
grpc_security_init();
for (i = 0; i < g_number_of_plugins; i++) {
@@ -236,7 +235,6 @@ void grpc_shutdown(void) {
gpr_mu_lock(&g_init_mu);
if (--g_initializations == 0) {
grpc_executor_shutdown(&exec_ctx);
- grpc_cq_global_shutdown();
grpc_iomgr_shutdown(&exec_ctx);
gpr_timers_global_destroy();
grpc_tracer_shutdown();
diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py
index 691e96a32286b3389d2801dbede758c56b540bdf..26d93faf75287e6d91ece853118257a5ad8426a4 100644
--- a/src/python/grpcio/grpc/_channel.py
+++ b/src/python/grpcio/grpc/_channel.py
@@ -444,10 +444,10 @@ def _start_unary_request(request, timeout, request_serializer):
return deadline, deadline_timespec, serialized_request, None
-def _end_unary_response_blocking(state, with_call, deadline):
+def _end_unary_response_blocking(state, call, with_call, deadline):
if state.code is grpc.StatusCode.OK:
if with_call:
- rendezvous = _Rendezvous(state, None, None, deadline)
+ rendezvous = _Rendezvous(state, call, None, deadline)
return state.response, rendezvous
else:
return state.response
@@ -499,17 +499,17 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
_check_call_error(call_error, metadata)
_handle_event(completion_queue.poll(), state,
self._response_deserializer)
- return state, deadline
+ return state, call, deadline
def __call__(self, request, timeout=None, metadata=None, credentials=None):
- state, deadline, = self._blocking(request, timeout, metadata,
- credentials)
- return _end_unary_response_blocking(state, False, deadline)
+ state, call, deadline = self._blocking(request, timeout, metadata,
+ credentials)
+ return _end_unary_response_blocking(state, call, False, deadline)
def with_call(self, request, timeout=None, metadata=None, credentials=None):
- state, deadline, = self._blocking(request, timeout, metadata,
- credentials)
- return _end_unary_response_blocking(state, True, deadline)
+ state, call, deadline = self._blocking(request, timeout, metadata,
+ credentials)
+ return _end_unary_response_blocking(state, call, True, deadline)
def future(self, request, timeout=None, metadata=None, credentials=None):
state, operations, deadline, deadline_timespec, rendezvous = self._prepare(
@@ -619,25 +619,25 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable):
state.condition.notify_all()
if not state.due:
break
- return state, deadline
+ return state, call, deadline
def __call__(self,
request_iterator,
timeout=None,
metadata=None,
credentials=None):
- state, deadline, = self._blocking(request_iterator, timeout, metadata,
- credentials)
- return _end_unary_response_blocking(state, False, deadline)
+ state, call, deadline = self._blocking(request_iterator, timeout,
+ metadata, credentials)
+ return _end_unary_response_blocking(state, call, False, deadline)
def with_call(self,
request_iterator,
timeout=None,
metadata=None,
credentials=None):
- state, deadline, = self._blocking(request_iterator, timeout, metadata,
- credentials)
- return _end_unary_response_blocking(state, True, deadline)
+ state, call, deadline = self._blocking(request_iterator, timeout,
+ metadata, credentials)
+ return _end_unary_response_blocking(state, call, True, deadline)
def future(self,
request_iterator,
diff --git a/test/core/internal_api_canaries/iomgr.c b/test/core/internal_api_canaries/iomgr.c
index d73d5c175c9ad2db438406e510c59b54bf0164bb..6fdaf1f98063b1d83ad399fdec889f8e837c1ec4 100644
--- a/test/core/internal_api_canaries/iomgr.c
+++ b/test/core/internal_api_canaries/iomgr.c
@@ -105,7 +105,6 @@ static void test_code(void) {
grpc_pollset_size();
grpc_pollset_init(NULL, NULL);
grpc_pollset_shutdown(NULL, NULL, NULL);
- grpc_pollset_reset(NULL);
grpc_pollset_destroy(NULL);
GRPC_ERROR_UNREF(grpc_pollset_work(NULL, NULL, NULL,
gpr_now(GPR_CLOCK_REALTIME),
diff --git a/test/cpp/microbenchmarks/bm_fullstack.cc b/test/cpp/microbenchmarks/bm_fullstack.cc
index 5c8567461135a267c5df22fc1fdfeb7984d913d3..c63de0ce0ab35c484d14e66dbea4c8596c885167 100644
--- a/test/cpp/microbenchmarks/bm_fullstack.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack.cc
@@ -698,6 +698,7 @@ static void BM_StreamingPingPongMsgs(benchmark::State& state) {
}
while (state.KeepRunning()) {
+ GPR_TIMER_SCOPE("BenchmarkCycle", 0);
request_rw->Write(send_request, tag(0)); // Start client send
response_rw.Read(&recv_request, tag(1)); // Start server recv
request_rw->Read(&recv_response, tag(2)); // Start client recv
diff --git a/tools/internal_ci/linux/grpc_interop.cfg b/tools/internal_ci/linux/grpc_interop.cfg
new file mode 100644
index 0000000000000000000000000000000000000000..9259faf34d5a366c7deb7c97d12d9f8160181830
--- /dev/null
+++ b/tools/internal_ci/linux/grpc_interop.cfg
@@ -0,0 +1,41 @@
+#!/bin/bash
+# Copyright 2017, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+# * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+# Config file for the internal CI (in protobuf text format)
+
+# Location of the continuous shell script in repository.
+build_file: "grpc/tools/internal_ci/linux/grpc_interop.sh"
+# grpc_interop tests can take 6+ hours to complete.
+timeout_mins: 480
+action {
+ define_artifacts {
+ regex: "**/sponge_log.xml"
+ }
+}
diff --git a/tools/internal_ci/linux/grpc_interop.sh b/tools/internal_ci/linux/grpc_interop.sh
new file mode 100755
index 0000000000000000000000000000000000000000..68bb4199e3dae6bd549ff1538025643414bcc81e
--- /dev/null
+++ b/tools/internal_ci/linux/grpc_interop.sh
@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# Copyright 2017, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+# * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+set -ex
+
+export LANG=en_US.UTF-8
+
+# Enter the gRPC repo root
+cd $(dirname $0)/../../..
+
+git submodule update --init
+
+tools/run_tests/run_interop_tests.py -l all -s all --cloud_to_prod --cloud_to_prod_auth --use_docker --http2_interop -t -j 12 $@ || FAILED="true"
+tools/run_tests/run_interop_tests.py -l java --use_docker --http2_badserver_interop $@ || FAILED="true"
+tools/run_tests/run_interop_tests.py -l python --use_docker --http2_badserver_interop $@ || FAILED="true"
diff --git a/tools/run_tests/python_utils/start_port_server.py b/tools/run_tests/python_utils/start_port_server.py
index d521aa6a9d940040ee529e373677f37c559d37e9..8ee7080168a5b84fa14c578ee26c68c33dd80422 100644
--- a/tools/run_tests/python_utils/start_port_server.py
+++ b/tools/run_tests/python_utils/start_port_server.py
@@ -36,6 +36,7 @@ import tempfile
import sys
import time
import jobset
+import socket
def start_port_server(port_server_port):
# check if a compatible port server is running
diff --git a/tools/run_tests/run_microbenchmark.py b/tools/run_tests/run_microbenchmark.py
index d748e5a0ce462dd1d36b065e091169daf549edf0..4884911c995e679e29b6aea9d7d7a11a609ba83b 100755
--- a/tools/run_tests/run_microbenchmark.py
+++ b/tools/run_tests/run_microbenchmark.py
@@ -126,23 +126,45 @@ def collect_perf(bm_name, args):
subprocess.check_call(
['make', bm_name,
'CONFIG=mutrace', '-j', '%d' % multiprocessing.cpu_count()])
+ benchmarks = []
+ profile_analysis = []
+ cleanup = []
for line in subprocess.check_output(['bins/mutrace/%s' % bm_name,
'--benchmark_list_tests']).splitlines():
- subprocess.check_call(['perf', 'record', '-o', '%s-perf.data' % fnize(line),
- '-g', '-c', '1000',
- 'bins/mutrace/%s' % bm_name,
- '--benchmark_filter=^%s$' % line,
- '--benchmark_min_time=10'])
- env = os.environ.copy()
- env.update({
- 'PERF_BASE_NAME': fnize(line),
- 'OUTPUT_DIR': 'reports',
- 'OUTPUT_FILENAME': fnize(line),
- })
- subprocess.check_call(['tools/run_tests/performance/process_local_perf_flamegraphs.sh'],
- env=env)
- subprocess.check_call(['rm', '%s-perf.data' % fnize(line)])
- subprocess.check_call(['rm', '%s-out.perf' % fnize(line)])
+ link(line, '%s.svg' % fnize(line))
+ benchmarks.append(
+ jobset.JobSpec(['perf', 'record', '-o', '%s-perf.data' % fnize(line),
+ '-g', '-F', '997',
+ 'bins/mutrace/%s' % bm_name,
+ '--benchmark_filter=^%s$' % line,
+ '--benchmark_min_time=10']))
+ profile_analysis.append(
+ jobset.JobSpec(['tools/run_tests/performance/process_local_perf_flamegraphs.sh'],
+ environ = {
+ 'PERF_BASE_NAME': fnize(line),
+ 'OUTPUT_DIR': 'reports',
+ 'OUTPUT_FILENAME': fnize(line),
+ }))
+ cleanup.append(jobset.JobSpec(['rm', '%s-perf.data' % fnize(line)]))
+ cleanup.append(jobset.JobSpec(['rm', '%s-out.perf' % fnize(line)]))
+ # periodically flush out the list of jobs: temporary space required for this
+ # processing is large
+ if len(benchmarks) >= 20:
+ # run up to half the cpu count: each benchmark can use up to two cores
+ # (one for the microbenchmark, one for the data flush)
+ jobset.run(benchmarks, maxjobs=1,
+ add_env={'GRPC_TEST_PORT_SERVER': 'localhost:%d' % port_server_port})
+ jobset.run(profile_analysis, maxjobs=multiprocessing.cpu_count())
+ jobset.run(cleanup, maxjobs=multiprocessing.cpu_count())
+ benchmarks = []
+ profile_analysis = []
+ cleanup = []
+ # run the remaining benchmarks that weren't flushed
+ if len(benchmarks):
+ jobset.run(benchmarks, maxjobs=1,
+ add_env={'GRPC_TEST_PORT_SERVER': 'localhost:%d' % port_server_port})
+ jobset.run(profile_analysis, maxjobs=multiprocessing.cpu_count())
+ jobset.run(cleanup, maxjobs=multiprocessing.cpu_count())
def collect_summary(bm_name, args):
heading('Summary: %s' % bm_name)