1 /*
2 *
3 * Copyright 2017 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include <grpc/support/port_platform.h>
20
21 #include "src/core/ext/filters/client_channel/backup_poller.h"
22
23 #include <grpc/grpc.h>
24 #include <grpc/support/alloc.h>
25 #include <grpc/support/log.h>
26 #include <grpc/support/sync.h>
27 #include "src/core/ext/filters/client_channel/client_channel.h"
28 #include "src/core/lib/gpr/env.h"
29 #include "src/core/lib/gpr/string.h"
30 #include "src/core/lib/iomgr/error.h"
31 #include "src/core/lib/iomgr/pollset.h"
32 #include "src/core/lib/iomgr/timer.h"
33 #include "src/core/lib/surface/channel.h"
34 #include "src/core/lib/surface/completion_queue.h"
35
36 #define DEFAULT_POLL_INTERVAL_MS 5000
37
38 namespace {
39 struct backup_poller {
40 grpc_timer polling_timer;
41 grpc_closure run_poller_closure;
42 grpc_closure shutdown_closure;
43 gpr_mu* pollset_mu;
44 grpc_pollset* pollset; // guarded by pollset_mu
45 bool shutting_down; // guarded by pollset_mu
46 gpr_refcount refs;
47 gpr_refcount shutdown_refs;
48 };
49 } // namespace
50
51 static gpr_once g_once = GPR_ONCE_INIT;
52 static gpr_mu g_poller_mu;
53 static backup_poller* g_poller = nullptr; // guarded by g_poller_mu
54 // g_poll_interval_ms is set only once at the first time
55 // grpc_client_channel_start_backup_polling() is called, after that it is
56 // treated as const.
57 static int g_poll_interval_ms = DEFAULT_POLL_INTERVAL_MS;
58
init_globals()59 static void init_globals() {
60 gpr_mu_init(&g_poller_mu);
61 char* env = gpr_getenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS");
62 if (env != nullptr) {
63 int poll_interval_ms = gpr_parse_nonnegative_int(env);
64 if (poll_interval_ms == -1) {
65 gpr_log(GPR_ERROR,
66 "Invalid GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS: %s, "
67 "default value %d will be used.",
68 env, g_poll_interval_ms);
69 } else {
70 g_poll_interval_ms = poll_interval_ms;
71 }
72 }
73 gpr_free(env);
74 }
75
backup_poller_shutdown_unref(backup_poller * p)76 static void backup_poller_shutdown_unref(backup_poller* p) {
77 if (gpr_unref(&p->shutdown_refs)) {
78 grpc_pollset_destroy(p->pollset);
79 gpr_free(p->pollset);
80 gpr_free(p);
81 }
82 }
83
done_poller(void * arg,grpc_error * error)84 static void done_poller(void* arg, grpc_error* error) {
85 backup_poller_shutdown_unref(static_cast<backup_poller*>(arg));
86 }
87
g_poller_unref()88 static void g_poller_unref() {
89 gpr_mu_lock(&g_poller_mu);
90 if (gpr_unref(&g_poller->refs)) {
91 backup_poller* p = g_poller;
92 g_poller = nullptr;
93 gpr_mu_unlock(&g_poller_mu);
94 gpr_mu_lock(p->pollset_mu);
95 p->shutting_down = true;
96 grpc_pollset_shutdown(
97 p->pollset, GRPC_CLOSURE_INIT(&p->shutdown_closure, done_poller, p,
98 grpc_schedule_on_exec_ctx));
99 gpr_mu_unlock(p->pollset_mu);
100 grpc_timer_cancel(&p->polling_timer);
101 } else {
102 gpr_mu_unlock(&g_poller_mu);
103 }
104 }
105
run_poller(void * arg,grpc_error * error)106 static void run_poller(void* arg, grpc_error* error) {
107 backup_poller* p = static_cast<backup_poller*>(arg);
108 if (error != GRPC_ERROR_NONE) {
109 if (error != GRPC_ERROR_CANCELLED) {
110 GRPC_LOG_IF_ERROR("run_poller", GRPC_ERROR_REF(error));
111 }
112 backup_poller_shutdown_unref(p);
113 return;
114 }
115 gpr_mu_lock(p->pollset_mu);
116 if (p->shutting_down) {
117 gpr_mu_unlock(p->pollset_mu);
118 backup_poller_shutdown_unref(p);
119 return;
120 }
121 grpc_error* err =
122 grpc_pollset_work(p->pollset, nullptr, grpc_core::ExecCtx::Get()->Now());
123 gpr_mu_unlock(p->pollset_mu);
124 GRPC_LOG_IF_ERROR("Run client channel backup poller", err);
125 grpc_timer_init(&p->polling_timer,
126 grpc_core::ExecCtx::Get()->Now() + g_poll_interval_ms,
127 &p->run_poller_closure);
128 }
129
g_poller_init_locked()130 static void g_poller_init_locked() {
131 if (g_poller == nullptr) {
132 g_poller = static_cast<backup_poller*>(gpr_zalloc(sizeof(backup_poller)));
133 g_poller->pollset =
134 static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
135 g_poller->shutting_down = false;
136 grpc_pollset_init(g_poller->pollset, &g_poller->pollset_mu);
137 gpr_ref_init(&g_poller->refs, 0);
138 // one for timer cancellation, one for pollset shutdown
139 gpr_ref_init(&g_poller->shutdown_refs, 2);
140 GRPC_CLOSURE_INIT(&g_poller->run_poller_closure, run_poller, g_poller,
141 grpc_schedule_on_exec_ctx);
142 grpc_timer_init(&g_poller->polling_timer,
143 grpc_core::ExecCtx::Get()->Now() + g_poll_interval_ms,
144 &g_poller->run_poller_closure);
145 }
146 }
147
grpc_client_channel_start_backup_polling(grpc_pollset_set * interested_parties)148 void grpc_client_channel_start_backup_polling(
149 grpc_pollset_set* interested_parties) {
150 gpr_once_init(&g_once, init_globals);
151 if (g_poll_interval_ms == 0) {
152 return;
153 }
154 gpr_mu_lock(&g_poller_mu);
155 g_poller_init_locked();
156 gpr_ref(&g_poller->refs);
157 /* Get a reference to g_poller->pollset before releasing g_poller_mu to make
158 * TSAN happy. Otherwise, reading from g_poller (i.e g_poller->pollset) after
159 * releasing the lock and setting g_poller to NULL in g_poller_unref() is
160 * being flagged as a data-race by TSAN */
161 grpc_pollset* pollset = g_poller->pollset;
162 gpr_mu_unlock(&g_poller_mu);
163
164 grpc_pollset_set_add_pollset(interested_parties, pollset);
165 }
166
grpc_client_channel_stop_backup_polling(grpc_pollset_set * interested_parties)167 void grpc_client_channel_stop_backup_polling(
168 grpc_pollset_set* interested_parties) {
169 if (g_poll_interval_ms == 0) {
170 return;
171 }
172 grpc_pollset_set_del_pollset(interested_parties, g_poller->pollset);
173 g_poller_unref();
174 }
175