client_config bugfixes

pull/2303/head
Craig Tiller 10 years ago
parent b49736829a
commit 87cc0848ce
  1. 17
      src/core/client_config/lb_policies/pick_first.c
  2. 7
      src/core/client_config/subchannel.c
  3. 2
      src/core/client_config/subchannel.h
  4. 3
      src/core/iomgr/fd_posix.c
  5. 8
      src/core/iomgr/iomgr.c
  6. 1
      src/core/surface/channel_create.c

@ -176,15 +176,20 @@ loop:
del_interested_parties_locked(p); del_interested_parties_locked(p);
GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel], GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel],
p->subchannels[p->num_subchannels - 1]); p->subchannels[p->num_subchannels - 1]);
p->checking_subchannel %= p->num_subchannels;
p->checking_connectivity = grpc_subchannel_check_connectivity(
p->subchannels[p->checking_subchannel]);
p->num_subchannels--; p->num_subchannels--;
GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first");
add_interested_parties_locked(p);
if (p->num_subchannels == 0) { if (p->num_subchannels == 0) {
abort(); while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
*pp->target = NULL;
grpc_iomgr_add_delayed_callback(pp->on_complete, 1);
gpr_free(pp);
}
} else { } else {
p->checking_subchannel %= p->num_subchannels;
p->checking_connectivity = grpc_subchannel_check_connectivity(
p->subchannels[p->checking_subchannel]);
GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first");
add_interested_parties_locked(p);
goto loop; goto loop;
} }
} }

@ -546,6 +546,7 @@ static void on_alarm(void *arg, int iomgr_success) {
grpc_subchannel *c = arg; grpc_subchannel *c = arg;
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
c->have_alarm = 0; c->have_alarm = 0;
connectivity_state_changed_locked(c);
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
if (iomgr_success) { if (iomgr_success) {
continue_connect(c); continue_connect(c);
@ -560,6 +561,7 @@ static void subchannel_connected(void *arg, int iomgr_success) {
publish_transport(c); publish_transport(c);
} else { } else {
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
connectivity_state_changed_locked(c);
GPR_ASSERT(!c->have_alarm); GPR_ASSERT(!c->have_alarm);
c->have_alarm = 1; c->have_alarm = 1;
c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta); c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta);
@ -570,7 +572,7 @@ static void subchannel_connected(void *arg, int iomgr_success) {
} }
static gpr_timespec compute_connect_deadline(grpc_subchannel *c) { static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
return gpr_time_add(gpr_now(), gpr_time_from_seconds(60)); return gpr_time_add(c->next_attempt, c->backoff_delta);
} }
static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) { static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
@ -578,6 +580,9 @@ static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
return GRPC_CHANNEL_FATAL_FAILURE; return GRPC_CHANNEL_FATAL_FAILURE;
} }
if (c->connecting) { if (c->connecting) {
if (c->have_alarm) {
return GRPC_CHANNEL_TRANSIENT_FAILURE;
}
return GRPC_CHANNEL_CONNECTING; return GRPC_CHANNEL_CONNECTING;
} }
if (c->active) { if (c->active) {

@ -43,8 +43,6 @@ typedef struct grpc_subchannel grpc_subchannel;
typedef struct grpc_subchannel_call grpc_subchannel_call; typedef struct grpc_subchannel_call grpc_subchannel_call;
typedef struct grpc_subchannel_args grpc_subchannel_args; typedef struct grpc_subchannel_args grpc_subchannel_args;
#define GRPC_SUBCHANNEL_REFCOUNT_DEBUG
#ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG #ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG
#define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p), __FILE__, __LINE__, (r)) #define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p), __FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_UNREF(p, r) grpc_subchannel_unref((p), __FILE__, __LINE__, (r)) #define GRPC_SUBCHANNEL_UNREF(p, r) grpc_subchannel_unref((p), __FILE__, __LINE__, (r))

@ -115,7 +115,7 @@ static void destroy(grpc_fd *fd) {
#define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__) #define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__)
static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file, static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file,
int line) { int line) {
gpr_log(GPR_DEBUG, "FD %d %p ref %d %d -> %d [%s; %s:%d]", fd->fd, fd, n, gpr_log(GPR_DEBUG, "FD %d %p ref %d %d -> %d [%s; %s:%d]", fd->fd, fd, n,
gpr_atm_no_barrier_load(&fd->refst), gpr_atm_no_barrier_load(&fd->refst),
gpr_atm_no_barrier_load(&fd->refst) + n, reason, file, line); gpr_atm_no_barrier_load(&fd->refst) + n, reason, file, line);
#else #else
@ -159,6 +159,7 @@ void grpc_fd_global_shutdown(void) {
grpc_fd *grpc_fd_create(int fd, const char *name) { grpc_fd *grpc_fd_create(int fd, const char *name) {
grpc_fd *r = alloc_fd(fd); grpc_fd *r = alloc_fd(fd);
gpr_log(GPR_DEBUG, "FD %d %p create", r->fd, r);
grpc_iomgr_register_object(&r->iomgr_object, name); grpc_iomgr_register_object(&r->iomgr_object, name);
return r; return r;
} }

@ -158,7 +158,7 @@ void grpc_iomgr_shutdown(void) {
"memory leaks are likely", "memory leaks are likely",
count_objects()); count_objects());
for (obj = g_root_object.next; obj != &g_root_object; obj = obj->next) { for (obj = g_root_object.next; obj != &g_root_object; obj = obj->next) {
gpr_log(GPR_DEBUG, "LEAKED OBJECT: %s", obj->name); gpr_log(GPR_DEBUG, "LEAKED OBJECT: %s %p", obj->name, obj);
} }
break; break;
} }
@ -177,8 +177,9 @@ void grpc_iomgr_shutdown(void) {
} }
void grpc_iomgr_register_object(grpc_iomgr_object *obj, const char *name) { void grpc_iomgr_register_object(grpc_iomgr_object *obj, const char *name) {
gpr_mu_lock(&g_mu);
obj->name = gpr_strdup(name); obj->name = gpr_strdup(name);
gpr_log(GPR_DEBUG, "register: %s %p", obj->name, obj);
gpr_mu_lock(&g_mu);
obj->next = &g_root_object; obj->next = &g_root_object;
obj->prev = obj->next->prev; obj->prev = obj->next->prev;
obj->next->prev = obj->prev->next = obj; obj->next->prev = obj->prev->next = obj;
@ -186,12 +187,13 @@ void grpc_iomgr_register_object(grpc_iomgr_object *obj, const char *name) {
} }
void grpc_iomgr_unregister_object(grpc_iomgr_object *obj) { void grpc_iomgr_unregister_object(grpc_iomgr_object *obj) {
gpr_log(GPR_DEBUG, "unregister: %s %p", obj->name, obj);
gpr_mu_lock(&g_mu); gpr_mu_lock(&g_mu);
obj->next->prev = obj->prev; obj->next->prev = obj->prev;
obj->prev->next = obj->next; obj->prev->next = obj->next;
gpr_free(obj->name);
gpr_cv_signal(&g_rcv); gpr_cv_signal(&g_rcv);
gpr_mu_unlock(&g_mu); gpr_mu_unlock(&g_mu);
gpr_free(obj->name);
} }
void grpc_iomgr_closure_init(grpc_iomgr_closure *closure, grpc_iomgr_cb_func cb, void grpc_iomgr_closure_init(grpc_iomgr_closure *closure, grpc_iomgr_cb_func cb,

@ -73,6 +73,7 @@ static void connected(void *arg, grpc_endpoint *tcp) {
if (tcp != NULL) { if (tcp != NULL) {
c->result->transport = c->result->transport =
grpc_create_chttp2_transport(c->args.channel_args, tcp, NULL, 0, c->args.metadata_context, 1); grpc_create_chttp2_transport(c->args.channel_args, tcp, NULL, 0, c->args.metadata_context, 1);
GPR_ASSERT(c->result->transport);
c->result->filters = gpr_malloc(sizeof(grpc_channel_filter*)); c->result->filters = gpr_malloc(sizeof(grpc_channel_filter*));
c->result->filters[0] = &grpc_http_client_filter; c->result->filters[0] = &grpc_http_client_filter;
c->result->num_filters = 1; c->result->num_filters = 1;

Loading…
Cancel
Save