Merge branch 'races' into enable-epoll1

reviewable/pr11816/r1
Craig Tiller 8 years ago
commit 06833e9bec
  1. 2
      src/core/ext/filters/client_channel/subchannel.c
  2. 12
      src/core/ext/filters/client_channel/subchannel_index.c
  3. 19
      test/core/end2end/fixtures/http_proxy_fixture.c
  4. 2
      test/core/surface/concurrent_connectivity_test.c

@ -283,6 +283,7 @@ static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
gpr_atm old_refs;
// add a weak ref and subtract a strong ref (atomically)
old_refs = ref_mutate(c, (gpr_atm)1 - (gpr_atm)(1 << INTERNAL_REF_BITS),
1 REF_MUTATE_PURPOSE("STRONG_UNREF"));
if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) {
@ -656,7 +657,6 @@ static bool publish_transport_locked(grpc_exec_ctx *exec_ctx,
gpr_free(sw_subchannel);
grpc_channel_stack_destroy(exec_ctx, stk);
gpr_free(con);
GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
return false;
}

@ -183,8 +183,11 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
enter_ctx(exec_ctx);
grpc_subchannel *c = NULL;
bool need_to_unref_constructed;
while (c == NULL) {
need_to_unref_constructed = false;
// Compare and swap loop:
// - take a reference to the current index
gpr_mu_lock(&g_mu);
@ -193,9 +196,12 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
// - Check to see if a subchannel already exists
c = gpr_avl_get(index, key);
if (c != NULL) {
c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "index_register");
}
if (c != NULL) {
// yes -> we're done
GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, constructed, "index_register");
need_to_unref_constructed = true;
} else {
// no -> update the avl and compare/swap
gpr_avl updated =
@ -219,6 +225,10 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
leave_ctx(exec_ctx);
if (need_to_unref_constructed) {
GRPC_SUBCHANNEL_UNREF(exec_ctx, constructed, "index_register");
}
return c;
}

@ -60,6 +60,7 @@
#include "src/core/lib/iomgr/tcp_client.h"
#include "src/core/lib/iomgr/tcp_server.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/slice/slice_internal.h"
#include "test/core/util/port.h"
@ -71,6 +72,8 @@ struct grpc_end2end_http_proxy {
gpr_mu* mu;
grpc_pollset* pollset;
gpr_refcount users;
grpc_combiner *combiner;
};
//
@ -400,19 +403,19 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg,
grpc_pollset_set_add_pollset(exec_ctx, conn->pollset_set, proxy->pollset);
grpc_endpoint_add_to_pollset_set(exec_ctx, endpoint, conn->pollset_set);
grpc_closure_init(&conn->on_read_request_done, on_read_request_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_closure_init(&conn->on_server_connect_done, on_server_connect_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_closure_init(&conn->on_write_response_done, on_write_response_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_closure_init(&conn->on_client_read_done, on_client_read_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_closure_init(&conn->on_client_write_done, on_client_write_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_closure_init(&conn->on_server_read_done, on_server_read_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_closure_init(&conn->on_server_write_done, on_server_write_done, conn,
grpc_schedule_on_exec_ctx);
grpc_combiner_scheduler(conn->proxy->combiner, false));
grpc_slice_buffer_init(&conn->client_read_buffer);
grpc_slice_buffer_init(&conn->client_deferred_write_buffer);
grpc_slice_buffer_init(&conn->client_write_buffer);
@ -453,6 +456,7 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) {
grpc_end2end_http_proxy* proxy =
(grpc_end2end_http_proxy*)gpr_malloc(sizeof(*proxy));
memset(proxy, 0, sizeof(*proxy));
proxy->combiner = grpc_combiner_create(NULL);
gpr_ref_init(&proxy->users, 1);
// Construct proxy address.
const int proxy_port = grpc_pick_unused_port_or_die();
@ -504,6 +508,7 @@ void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy) {
grpc_pollset_shutdown(&exec_ctx, proxy->pollset,
grpc_closure_create(destroy_pollset, proxy->pollset,
grpc_schedule_on_exec_ctx));
grpc_combiner_unref(&exec_ctx, proxy->combiner);
gpr_free(proxy);
grpc_exec_ctx_finish(&exec_ctx);
}

@ -112,7 +112,9 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *vargs, grpc_endpoint *tcp,
grpc_endpoint_shutdown(exec_ctx, tcp,
GRPC_ERROR_CREATE_FROM_STATIC_STRING("Connected"));
grpc_endpoint_destroy(exec_ctx, tcp);
gpr_mu_lock(args->mu);
GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(args->pollset, NULL));
gpr_mu_unlock(args->mu);
}
void bad_server_thread(void *vargs) {

Loading…
Cancel
Save