mirror of https://github.com/grpc/grpc.git
Merge pull request #12732 from y-zeng/connectivity_watcher
Client channel backup pollerpull/13069/head
Yuchen Zeng
7 years ago
committed by
GitHub
commit
9e6837d397
21 changed files with 293 additions and 190 deletions
@ -0,0 +1,158 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include "src/core/ext/filters/client_channel/backup_poller.h" |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/sync.h> |
||||
#include "src/core/ext/filters/client_channel/client_channel.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/iomgr/pollset.h" |
||||
#include "src/core/lib/iomgr/timer.h" |
||||
#include "src/core/lib/support/env.h" |
||||
#include "src/core/lib/support/string.h" |
||||
#include "src/core/lib/surface/channel.h" |
||||
#include "src/core/lib/surface/completion_queue.h" |
||||
|
||||
#define DEFAULT_POLL_INTERVAL_MS 5000 |
||||
|
||||
typedef struct backup_poller { |
||||
grpc_timer polling_timer; |
||||
grpc_closure run_poller_closure; |
||||
grpc_closure shutdown_closure; |
||||
gpr_mu* pollset_mu; |
||||
grpc_pollset* pollset; // guarded by pollset_mu
|
||||
bool shutting_down; // guarded by pollset_mu
|
||||
gpr_refcount refs; |
||||
gpr_refcount shutdown_refs; |
||||
} backup_poller; |
||||
|
||||
static gpr_once g_once = GPR_ONCE_INIT; |
||||
static gpr_mu g_poller_mu; |
||||
static backup_poller* g_poller = NULL; // guarded by g_poller_mu
|
||||
// g_poll_interval_ms is set only once at the first time
|
||||
// grpc_client_channel_start_backup_polling() is called, after that it is
|
||||
// treated as const.
|
||||
static int g_poll_interval_ms = DEFAULT_POLL_INTERVAL_MS; |
||||
|
||||
static void init_globals() { |
||||
gpr_mu_init(&g_poller_mu); |
||||
char* env = gpr_getenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS"); |
||||
if (env != NULL) { |
||||
int poll_interval_ms = gpr_parse_nonnegative_int(env); |
||||
if (poll_interval_ms == -1) { |
||||
gpr_log(GPR_ERROR, |
||||
"Invalid GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS: %s, " |
||||
"default value %d will be used.", |
||||
env, g_poll_interval_ms); |
||||
} else { |
||||
g_poll_interval_ms = poll_interval_ms; |
||||
} |
||||
} |
||||
gpr_free(env); |
||||
} |
||||
|
||||
static void backup_poller_shutdown_unref(grpc_exec_ctx* exec_ctx, |
||||
backup_poller* p) { |
||||
if (gpr_unref(&p->shutdown_refs)) { |
||||
grpc_pollset_destroy(exec_ctx, p->pollset); |
||||
gpr_free(p->pollset); |
||||
gpr_free(p); |
||||
} |
||||
} |
||||
|
||||
static void done_poller(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { |
||||
backup_poller_shutdown_unref(exec_ctx, (backup_poller*)arg); |
||||
} |
||||
|
||||
static void g_poller_unref(grpc_exec_ctx* exec_ctx) { |
||||
if (gpr_unref(&g_poller->refs)) { |
||||
gpr_mu_lock(&g_poller_mu); |
||||
backup_poller* p = g_poller; |
||||
g_poller = NULL; |
||||
gpr_mu_unlock(&g_poller_mu); |
||||
gpr_mu_lock(p->pollset_mu); |
||||
p->shutting_down = true; |
||||
grpc_pollset_shutdown(exec_ctx, p->pollset, |
||||
GRPC_CLOSURE_INIT(&p->shutdown_closure, done_poller, |
||||
p, grpc_schedule_on_exec_ctx)); |
||||
gpr_mu_unlock(p->pollset_mu); |
||||
grpc_timer_cancel(exec_ctx, &p->polling_timer); |
||||
} |
||||
} |
||||
|
||||
static void run_poller(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { |
||||
backup_poller* p = (backup_poller*)arg; |
||||
if (error != GRPC_ERROR_NONE) { |
||||
if (error != GRPC_ERROR_CANCELLED) { |
||||
GRPC_LOG_IF_ERROR("run_poller", GRPC_ERROR_REF(error)); |
||||
} |
||||
backup_poller_shutdown_unref(exec_ctx, p); |
||||
return; |
||||
} |
||||
gpr_mu_lock(p->pollset_mu); |
||||
if (p->shutting_down) { |
||||
gpr_mu_unlock(p->pollset_mu); |
||||
backup_poller_shutdown_unref(exec_ctx, p); |
||||
return; |
||||
} |
||||
grpc_error* err = grpc_pollset_work(exec_ctx, p->pollset, NULL, |
||||
grpc_exec_ctx_now(exec_ctx)); |
||||
gpr_mu_unlock(p->pollset_mu); |
||||
GRPC_LOG_IF_ERROR("Run client channel backup poller", err); |
||||
grpc_timer_init(exec_ctx, &p->polling_timer, |
||||
grpc_exec_ctx_now(exec_ctx) + g_poll_interval_ms, |
||||
&p->run_poller_closure); |
||||
} |
||||
|
||||
void grpc_client_channel_start_backup_polling( |
||||
grpc_exec_ctx* exec_ctx, grpc_pollset_set* interested_parties) { |
||||
gpr_once_init(&g_once, init_globals); |
||||
if (g_poll_interval_ms == 0) { |
||||
return; |
||||
} |
||||
gpr_mu_lock(&g_poller_mu); |
||||
if (g_poller == NULL) { |
||||
g_poller = (backup_poller*)gpr_zalloc(sizeof(backup_poller)); |
||||
g_poller->pollset = (grpc_pollset*)gpr_zalloc(grpc_pollset_size()); |
||||
g_poller->shutting_down = false; |
||||
grpc_pollset_init(g_poller->pollset, &g_poller->pollset_mu); |
||||
gpr_ref_init(&g_poller->refs, 0); |
||||
// one for timer cancellation, one for pollset shutdown
|
||||
gpr_ref_init(&g_poller->shutdown_refs, 2); |
||||
GRPC_CLOSURE_INIT(&g_poller->run_poller_closure, run_poller, g_poller, |
||||
grpc_schedule_on_exec_ctx); |
||||
grpc_timer_init(exec_ctx, &g_poller->polling_timer, |
||||
grpc_exec_ctx_now(exec_ctx) + g_poll_interval_ms, |
||||
&g_poller->run_poller_closure); |
||||
} |
||||
gpr_ref(&g_poller->refs); |
||||
gpr_mu_unlock(&g_poller_mu); |
||||
grpc_pollset_set_add_pollset(exec_ctx, interested_parties, g_poller->pollset); |
||||
} |
||||
|
||||
void grpc_client_channel_stop_backup_polling( |
||||
grpc_exec_ctx* exec_ctx, grpc_pollset_set* interested_parties) { |
||||
if (g_poll_interval_ms == 0) { |
||||
return; |
||||
} |
||||
grpc_pollset_set_del_pollset(exec_ctx, interested_parties, g_poller->pollset); |
||||
g_poller_unref(exec_ctx); |
||||
} |
||||
@ -0,0 +1,34 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_BACKUP_POLLER_H |
||||
#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_BACKUP_POLLER_H |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" |
||||
|
||||
/* Start polling \a interested_parties periodically in the timer thread */ |
||||
void grpc_client_channel_start_backup_polling( |
||||
grpc_exec_ctx* exec_ctx, grpc_pollset_set* interested_parties); |
||||
|
||||
/* Stop polling \a interested_parties */ |
||||
void grpc_client_channel_stop_backup_polling( |
||||
grpc_exec_ctx* exec_ctx, grpc_pollset_set* interested_parties); |
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_BACKUP_POLLER_H */ |
||||
Loading…
Reference in new issue