blob: 05fd02fb0ab39f146c26011fbf66f42266426e0a [file] [log] [blame]
Craig Tilleraf691802015-06-23 14:57:07 -07001/*
2 *
3 * Copyright 2015, Google Inc.
4 * All rights reserved.
5 *
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions are
8 * met:
9 *
10 * * Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * * Redistributions in binary form must reproduce the above
13 * copyright notice, this list of conditions and the following disclaimer
14 * in the documentation and/or other materials provided with the
15 * distribution.
16 * * Neither the name of Google Inc. nor the names of its
17 * contributors may be used to endorse or promote products derived from
18 * this software without specific prior written permission.
19 *
20 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 *
32 */
33
34#include "src/core/client_config/subchannel.h"
Craig Tiller2595ab72015-06-25 15:26:00 -070035
Craig Tillerf7afa1f2015-06-26 09:02:20 -070036#include <string.h>
37
Craig Tiller2595ab72015-06-25 15:26:00 -070038#include <grpc/support/alloc.h>
39
Craig Tillereb3b12e2015-06-26 14:42:49 -070040#include "src/core/channel/channel_args.h"
Craig Tillerff54c922015-06-26 16:57:20 -070041#include "src/core/channel/connected_channel.h"
Craig Tillerff3ae682015-06-29 17:44:04 -070042#include "src/core/iomgr/alarm.h"
Craig Tiller08a1cf82015-06-29 09:37:52 -070043#include "src/core/transport/connectivity_state.h"
Craig Tillereb3b12e2015-06-26 14:42:49 -070044
45typedef struct {
Craig Tiller4ab82d22015-06-29 09:40:33 -070046 /* all fields protected by subchannel->mu */
47 /** refcount */
48 int refs;
49 /** parent subchannel */
Craig Tillereb3b12e2015-06-26 14:42:49 -070050 grpc_subchannel *subchannel;
51} connection;
52
Craig Tillerdf91ba52015-06-29 10:55:46 -070053typedef struct {
54 grpc_iomgr_closure closure;
55 size_t version;
56 grpc_subchannel *subchannel;
57 grpc_connectivity_state connectivity_state;
58} state_watcher;
59
Craig Tiller5f84c842015-06-26 16:08:21 -070060typedef struct waiting_for_connect {
61 struct waiting_for_connect *next;
62 grpc_iomgr_closure *notify;
Craig Tillerabf36382015-06-29 16:13:27 -070063 grpc_pollset *pollset;
Craig Tiller5f84c842015-06-26 16:08:21 -070064 grpc_subchannel_call **target;
Craig Tillerdf91ba52015-06-29 10:55:46 -070065 grpc_subchannel *subchannel;
66 grpc_iomgr_closure continuation;
Craig Tiller5f84c842015-06-26 16:08:21 -070067} waiting_for_connect;
68
Craig Tiller2595ab72015-06-25 15:26:00 -070069struct grpc_subchannel {
Craig Tiller91624662015-06-25 16:31:02 -070070 grpc_connector *connector;
Craig Tillerf7afa1f2015-06-26 09:02:20 -070071
72 /** non-transport related channel filters */
73 const grpc_channel_filter **filters;
Craig Tiller5945ee12015-06-27 10:36:09 -070074 size_t num_filters;
Craig Tillerf7afa1f2015-06-26 09:02:20 -070075 /** channel arguments */
76 grpc_channel_args *args;
77 /** address to connect to */
78 struct sockaddr *addr;
79 size_t addr_len;
Craig Tiller5f84c842015-06-26 16:08:21 -070080 /** metadata context */
81 grpc_mdctx *mdctx;
Craig Tiller98465032015-06-29 14:36:42 -070082 /** master channel */
83 grpc_channel *master;
Craig Tillerb6fbf1d2015-06-29 15:25:49 -070084 /** have we seen a disconnection? */
85 int disconnected;
Craig Tillereb3b12e2015-06-26 14:42:49 -070086
87 /** set during connection */
Craig Tiller04c5d4b2015-06-26 17:21:41 -070088 grpc_connect_out_args connecting_result;
Craig Tillereb3b12e2015-06-26 14:42:49 -070089
90 /** callback for connection finishing */
91 grpc_iomgr_closure connected;
92
Craig Tiller5f84c842015-06-26 16:08:21 -070093 /** pollset_set tracking who's interested in a connection
94 being setup */
95 grpc_pollset_set pollset_set;
96
Craig Tillereb3b12e2015-06-26 14:42:49 -070097 /** mutex protecting remaining elements */
98 gpr_mu mu;
99
100 /** active connection */
101 connection *active;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700102 /** version number for the active connection */
103 size_t active_version;
Craig Tillerd7b68e72015-06-28 11:41:09 -0700104 /** refcount */
105 int refs;
Craig Tillereb3b12e2015-06-26 14:42:49 -0700106 /** are we connecting */
107 int connecting;
Craig Tiller5f84c842015-06-26 16:08:21 -0700108 /** things waiting for a connection */
109 waiting_for_connect *waiting;
Craig Tillerc7b5f762015-06-27 11:48:42 -0700110 /** connectivity state tracking */
111 grpc_connectivity_state_tracker state_tracker;
Craig Tillerff3ae682015-06-29 17:44:04 -0700112
113 /** next connect attempt time */
114 gpr_timespec next_attempt;
115 /** amount to backoff each failure */
116 gpr_timespec backoff_delta;
117 /** do we have an active alarm? */
118 int have_alarm;
119 /** our alarm */
120 grpc_alarm alarm;
Craig Tiller2595ab72015-06-25 15:26:00 -0700121};
122
123struct grpc_subchannel_call {
Craig Tillereb3b12e2015-06-26 14:42:49 -0700124 connection *connection;
Craig Tiller2595ab72015-06-25 15:26:00 -0700125 gpr_refcount refs;
126};
127
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700128#define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
129#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)((con) + 1))
Craig Tiller2595ab72015-06-25 15:26:00 -0700130
Craig Tillerabf36382015-06-29 16:13:27 -0700131static grpc_subchannel_call *create_call(connection *con);
Craig Tiller5f84c842015-06-26 16:08:21 -0700132static void connectivity_state_changed_locked(grpc_subchannel *c);
133static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c);
134static gpr_timespec compute_connect_deadline(grpc_subchannel *c);
Craig Tillerff54c922015-06-26 16:57:20 -0700135static void subchannel_connected(void *subchannel, int iomgr_success);
Craig Tillereb3b12e2015-06-26 14:42:49 -0700136
Craig Tiller079a11b2015-06-30 10:07:15 -0700137static void subchannel_ref_locked(
138 grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
139static int subchannel_unref_locked(
140 grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT;
Craig Tillerc3967532015-06-29 14:59:38 -0700141static void connection_ref_locked(connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
Craig Tiller079a11b2015-06-30 10:07:15 -0700142static grpc_subchannel *connection_unref_locked(
143 connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT;
Craig Tillerd7b68e72015-06-28 11:41:09 -0700144static void subchannel_destroy(grpc_subchannel *c);
145
Craig Tillerc3967532015-06-29 14:59:38 -0700146#ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG
Craig Tiller079a11b2015-06-30 10:07:15 -0700147#define SUBCHANNEL_REF_LOCKED(p, r) \
148 subchannel_ref_locked((p), __FILE__, __LINE__, (r))
149#define SUBCHANNEL_UNREF_LOCKED(p, r) \
150 subchannel_unref_locked((p), __FILE__, __LINE__, (r))
151#define CONNECTION_REF_LOCKED(p, r) \
152 connection_ref_locked((p), __FILE__, __LINE__, (r))
153#define CONNECTION_UNREF_LOCKED(p, r) \
154 connection_unref_locked((p), __FILE__, __LINE__, (r))
Craig Tillerc3967532015-06-29 14:59:38 -0700155#define REF_PASS_ARGS , file, line, reason
Craig Tiller079a11b2015-06-30 10:07:15 -0700156#define REF_LOG(name, p) \
157 gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p ref %d -> %d %s", \
158 (name), (p), (p)->refs, (p)->refs + 1, reason)
159#define UNREF_LOG(name, p) \
160 gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p unref %d -> %d %s", \
161 (name), (p), (p)->refs, (p)->refs - 1, reason)
Craig Tillerc3967532015-06-29 14:59:38 -0700162#else
163#define SUBCHANNEL_REF_LOCKED(p, r) subchannel_ref_locked((p))
164#define SUBCHANNEL_UNREF_LOCKED(p, r) subchannel_unref_locked((p))
Craig Tiller11bf14e2015-06-29 16:35:41 -0700165#define CONNECTION_REF_LOCKED(p, r) connection_ref_locked((p))
166#define CONNECTION_UNREF_LOCKED(p, r) connection_unref_locked((p))
Craig Tillerc3967532015-06-29 14:59:38 -0700167#define REF_PASS_ARGS
Craig Tiller11bf14e2015-06-29 16:35:41 -0700168#define REF_LOG(name, p) \
169 do { \
170 } while (0)
171#define UNREF_LOG(name, p) \
172 do { \
173 } while (0)
Craig Tillerc3967532015-06-29 14:59:38 -0700174#endif
175
Craig Tiller2595ab72015-06-25 15:26:00 -0700176/*
Craig Tillerca3e9d32015-06-27 18:37:27 -0700177 * connection implementation
178 */
179
Craig Tillerd7b68e72015-06-28 11:41:09 -0700180static void connection_destroy(connection *c) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700181 GPR_ASSERT(c->refs == 0);
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700182 gpr_log(GPR_DEBUG, "CONNECTION_DESTROY %p", c);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700183 grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CONNECTION(c));
Craig Tillerd7b68e72015-06-28 11:41:09 -0700184 gpr_free(c);
Craig Tillerca3e9d32015-06-27 18:37:27 -0700185}
186
Craig Tiller079a11b2015-06-30 10:07:15 -0700187static void connection_ref_locked(
188 connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tillerc3967532015-06-29 14:59:38 -0700189 REF_LOG("CONNECTION", c);
190 subchannel_ref_locked(c->subchannel REF_PASS_ARGS);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700191 ++c->refs;
Craig Tillerd7b68e72015-06-28 11:41:09 -0700192}
193
Craig Tiller079a11b2015-06-30 10:07:15 -0700194static grpc_subchannel *connection_unref_locked(
195 connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700196 grpc_subchannel *destroy = NULL;
Craig Tillerc3967532015-06-29 14:59:38 -0700197 UNREF_LOG("CONNECTION", c);
198 if (subchannel_unref_locked(c->subchannel REF_PASS_ARGS)) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700199 destroy = c->subchannel;
200 }
Craig Tillerd7b68e72015-06-28 11:41:09 -0700201 if (--c->refs == 0 && c->subchannel->active != c) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700202 connection_destroy(c);
Craig Tillerca3e9d32015-06-27 18:37:27 -0700203 }
Craig Tillerd7b68e72015-06-28 11:41:09 -0700204 return destroy;
Craig Tillerca3e9d32015-06-27 18:37:27 -0700205}
206
207/*
Craig Tiller2595ab72015-06-25 15:26:00 -0700208 * grpc_subchannel implementation
209 */
210
Craig Tiller079a11b2015-06-30 10:07:15 -0700211static void subchannel_ref_locked(
212 grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tillerc3967532015-06-29 14:59:38 -0700213 REF_LOG("SUBCHANNEL", c);
Craig Tiller079a11b2015-06-30 10:07:15 -0700214 ++c->refs;
Craig Tillerc3967532015-06-29 14:59:38 -0700215}
Craig Tiller2595ab72015-06-25 15:26:00 -0700216
Craig Tiller079a11b2015-06-30 10:07:15 -0700217static int subchannel_unref_locked(
218 grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tillerc3967532015-06-29 14:59:38 -0700219 UNREF_LOG("SUBCHANNEL", c);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700220 return --c->refs == 0;
Craig Tillerd7b68e72015-06-28 11:41:09 -0700221}
222
Craig Tillerc3967532015-06-29 14:59:38 -0700223void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700224 gpr_mu_lock(&c->mu);
Craig Tillerc3967532015-06-29 14:59:38 -0700225 subchannel_ref_locked(c REF_PASS_ARGS);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700226 gpr_mu_unlock(&c->mu);
Craig Tillerd7b68e72015-06-28 11:41:09 -0700227}
228
Craig Tillerc3967532015-06-29 14:59:38 -0700229void grpc_subchannel_unref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700230 int destroy;
231 gpr_mu_lock(&c->mu);
Craig Tillerc3967532015-06-29 14:59:38 -0700232 destroy = subchannel_unref_locked(c REF_PASS_ARGS);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700233 gpr_mu_unlock(&c->mu);
234 if (destroy) subchannel_destroy(c);
Craig Tillerd7b68e72015-06-28 11:41:09 -0700235}
236
237static void subchannel_destroy(grpc_subchannel *c) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700238 if (c->active != NULL) {
239 connection_destroy(c->active);
240 }
Craig Tillerd7b68e72015-06-28 11:41:09 -0700241 gpr_free(c->filters);
242 grpc_channel_args_destroy(c->args);
243 gpr_free(c->addr);
244 grpc_mdctx_unref(c->mdctx);
245 grpc_pollset_set_destroy(&c->pollset_set);
246 grpc_connectivity_state_destroy(&c->state_tracker);
Craig Tillerc3967532015-06-29 14:59:38 -0700247 grpc_connector_unref(c->connector);
Craig Tillerd7b68e72015-06-28 11:41:09 -0700248 gpr_free(c);
Craig Tiller2595ab72015-06-25 15:26:00 -0700249}
250
Craig Tiller5f84c842015-06-26 16:08:21 -0700251void grpc_subchannel_add_interested_party(grpc_subchannel *c,
252 grpc_pollset *pollset) {
253 grpc_pollset_set_add_pollset(&c->pollset_set, pollset);
254}
255
256void grpc_subchannel_del_interested_party(grpc_subchannel *c,
257 grpc_pollset *pollset) {
258 grpc_pollset_set_del_pollset(&c->pollset_set, pollset);
259}
260
Craig Tillerf7afa1f2015-06-26 09:02:20 -0700261grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
262 grpc_subchannel_args *args) {
263 grpc_subchannel *c = gpr_malloc(sizeof(*c));
264 memset(c, 0, sizeof(*c));
Craig Tillerd7b68e72015-06-28 11:41:09 -0700265 c->refs = 1;
Craig Tillerf7afa1f2015-06-26 09:02:20 -0700266 c->connector = connector;
267 grpc_connector_ref(c->connector);
Craig Tiller5945ee12015-06-27 10:36:09 -0700268 c->num_filters = args->filter_count;
269 c->filters = gpr_malloc(sizeof(grpc_channel_filter *) * c->num_filters);
Craig Tillerf7afa1f2015-06-26 09:02:20 -0700270 memcpy(c->filters, args->filters,
Craig Tiller5945ee12015-06-27 10:36:09 -0700271 sizeof(grpc_channel_filter *) * c->num_filters);
Craig Tillerf7afa1f2015-06-26 09:02:20 -0700272 c->addr = gpr_malloc(args->addr_len);
273 memcpy(c->addr, args->addr, args->addr_len);
274 c->addr_len = args->addr_len;
Craig Tillereb3b12e2015-06-26 14:42:49 -0700275 c->args = grpc_channel_args_copy(args->args);
Craig Tiller5f84c842015-06-26 16:08:21 -0700276 c->mdctx = args->mdctx;
Craig Tiller98465032015-06-29 14:36:42 -0700277 c->master = args->master;
Craig Tiller5f84c842015-06-26 16:08:21 -0700278 grpc_mdctx_ref(c->mdctx);
Craig Tillerff54c922015-06-26 16:57:20 -0700279 grpc_pollset_set_init(&c->pollset_set);
280 grpc_iomgr_closure_init(&c->connected, subchannel_connected, c);
Craig Tillerc7b5f762015-06-27 11:48:42 -0700281 grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE);
Craig Tillereb3b12e2015-06-26 14:42:49 -0700282 gpr_mu_init(&c->mu);
Craig Tillerf7afa1f2015-06-26 09:02:20 -0700283 return c;
284}
285
Craig Tillerff3ae682015-06-29 17:44:04 -0700286static void continue_connect(grpc_subchannel *c) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700287 grpc_connect_in_args args;
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700288
Craig Tiller4ab82d22015-06-29 09:40:33 -0700289 args.interested_parties = &c->pollset_set;
290 args.addr = c->addr;
291 args.addr_len = c->addr_len;
292 args.deadline = compute_connect_deadline(c);
293 args.channel_args = c->args;
294 args.metadata_context = c->mdctx;
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700295
Craig Tiller4ab82d22015-06-29 09:40:33 -0700296 grpc_connector_connect(c->connector, &args, &c->connecting_result,
297 &c->connected);
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700298}
299
Craig Tillerff3ae682015-06-29 17:44:04 -0700300static void start_connect(grpc_subchannel *c) {
301 gpr_timespec now = gpr_now();
302 c->next_attempt = now;
303 c->backoff_delta = gpr_time_from_seconds(1);
304
305 continue_connect(c);
306}
307
Craig Tillerdf91ba52015-06-29 10:55:46 -0700308static void continue_creating_call(void *arg, int iomgr_success) {
309 waiting_for_connect *w4c = arg;
Craig Tillerabf36382015-06-29 16:13:27 -0700310 grpc_subchannel_create_call(w4c->subchannel, w4c->pollset, w4c->target,
Craig Tillerf62d6fc2015-06-29 10:55:59 -0700311 w4c->notify);
Craig Tillerc3967532015-06-29 14:59:38 -0700312 GRPC_SUBCHANNEL_UNREF(w4c->subchannel, "waiting_for_connect");
Craig Tillerdf91ba52015-06-29 10:55:46 -0700313 gpr_free(w4c);
314}
315
Craig Tillerabf36382015-06-29 16:13:27 -0700316void grpc_subchannel_create_call(grpc_subchannel *c, grpc_pollset *pollset,
Craig Tillereb3b12e2015-06-26 14:42:49 -0700317 grpc_subchannel_call **target,
318 grpc_iomgr_closure *notify) {
319 connection *con;
320 gpr_mu_lock(&c->mu);
321 if (c->active != NULL) {
322 con = c->active;
Craig Tillerc3967532015-06-29 14:59:38 -0700323 CONNECTION_REF_LOCKED(con, "call");
Craig Tillereb3b12e2015-06-26 14:42:49 -0700324 gpr_mu_unlock(&c->mu);
325
Craig Tillerabf36382015-06-29 16:13:27 -0700326 *target = create_call(con);
Craig Tillereb3b12e2015-06-26 14:42:49 -0700327 notify->cb(notify->cb_arg, 1);
328 } else {
Craig Tiller5f84c842015-06-26 16:08:21 -0700329 waiting_for_connect *w4c = gpr_malloc(sizeof(*w4c));
330 w4c->next = c->waiting;
331 w4c->notify = notify;
Craig Tillerabf36382015-06-29 16:13:27 -0700332 w4c->pollset = pollset;
Craig Tiller5f84c842015-06-26 16:08:21 -0700333 w4c->target = target;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700334 w4c->subchannel = c;
Craig Tiller98465032015-06-29 14:36:42 -0700335 /* released when clearing w4c */
Craig Tillerc3967532015-06-29 14:59:38 -0700336 SUBCHANNEL_REF_LOCKED(c, "waiting_for_connect");
Craig Tillerdf91ba52015-06-29 10:55:46 -0700337 grpc_iomgr_closure_init(&w4c->continuation, continue_creating_call, w4c);
Craig Tiller5f84c842015-06-26 16:08:21 -0700338 c->waiting = w4c;
Craig Tillerabf36382015-06-29 16:13:27 -0700339 grpc_subchannel_add_interested_party(c, pollset);
Craig Tillereb3b12e2015-06-26 14:42:49 -0700340 if (!c->connecting) {
341 c->connecting = 1;
Craig Tiller5f84c842015-06-26 16:08:21 -0700342 connectivity_state_changed_locked(c);
Craig Tiller98465032015-06-29 14:36:42 -0700343 /* released by connection */
Craig Tillerc3967532015-06-29 14:59:38 -0700344 SUBCHANNEL_REF_LOCKED(c, "connecting");
Craig Tillereb3b12e2015-06-26 14:42:49 -0700345 gpr_mu_unlock(&c->mu);
346
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700347 start_connect(c);
Craig Tillereb3b12e2015-06-26 14:42:49 -0700348 } else {
349 gpr_mu_unlock(&c->mu);
350 }
351 }
352}
353
Craig Tiller5f84c842015-06-26 16:08:21 -0700354grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
355 grpc_connectivity_state state;
356 gpr_mu_lock(&c->mu);
Craig Tillerc7b5f762015-06-27 11:48:42 -0700357 state = grpc_connectivity_state_check(&c->state_tracker);
Craig Tiller5f84c842015-06-26 16:08:21 -0700358 gpr_mu_unlock(&c->mu);
359 return state;
360}
361
362void grpc_subchannel_notify_on_state_change(grpc_subchannel *c,
363 grpc_connectivity_state *state,
364 grpc_iomgr_closure *notify) {
Craig Tiller5f84c842015-06-26 16:08:21 -0700365 int do_connect = 0;
Craig Tiller5f84c842015-06-26 16:08:21 -0700366 gpr_mu_lock(&c->mu);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700367 if (grpc_connectivity_state_notify_on_state_change(&c->state_tracker, state,
368 notify)) {
369 do_connect = 1;
Craig Tiller5f84c842015-06-26 16:08:21 -0700370 c->connecting = 1;
Craig Tiller98465032015-06-29 14:36:42 -0700371 /* released by connection */
Craig Tillerc3967532015-06-29 14:59:38 -0700372 SUBCHANNEL_REF_LOCKED(c, "connecting");
Craig Tiller11bf14e2015-06-29 16:35:41 -0700373 connectivity_state_changed_locked(c);
Craig Tiller5f84c842015-06-26 16:08:21 -0700374 }
Craig Tillerc7b5f762015-06-27 11:48:42 -0700375 gpr_mu_unlock(&c->mu);
Craig Tiller5f84c842015-06-26 16:08:21 -0700376 if (do_connect) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700377 start_connect(c);
Craig Tiller5f84c842015-06-26 16:08:21 -0700378 }
379}
380
Craig Tiller4ab82d22015-06-29 09:40:33 -0700381void grpc_subchannel_process_transport_op(grpc_subchannel *c,
382 grpc_transport_op *op) {
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700383 connection *con = NULL;
384 grpc_subchannel *destroy;
Craig Tillerff3ae682015-06-29 17:44:04 -0700385 int cancel_alarm = 0;
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700386 gpr_mu_lock(&c->mu);
387 if (op->disconnect) {
388 c->disconnected = 1;
Craig Tiller11bf14e2015-06-29 16:35:41 -0700389 connectivity_state_changed_locked(c);
Craig Tillerff3ae682015-06-29 17:44:04 -0700390 if (c->have_alarm) {
391 cancel_alarm = 1;
392 }
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700393 }
394 if (c->active != NULL) {
395 con = c->active;
396 CONNECTION_REF_LOCKED(con, "transport-op");
397 }
398 gpr_mu_unlock(&c->mu);
399
400 if (con != NULL) {
401 grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con);
Craig Tiller079a11b2015-06-30 10:07:15 -0700402 grpc_channel_element *top_elem =
403 grpc_channel_stack_element(channel_stack, 0);
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700404 top_elem->filter->start_transport_op(top_elem, op);
405
406 gpr_mu_lock(&c->mu);
407 destroy = CONNECTION_UNREF_LOCKED(con, "transport-op");
408 gpr_mu_unlock(&c->mu);
409 if (destroy) {
410 subchannel_destroy(destroy);
411 }
412 }
Craig Tillerff3ae682015-06-29 17:44:04 -0700413
414 if (cancel_alarm) {
415 grpc_alarm_cancel(&c->alarm);
416 }
Craig Tillerdf91ba52015-06-29 10:55:46 -0700417}
418
419static void on_state_changed(void *p, int iomgr_success) {
420 state_watcher *sw = p;
421 grpc_subchannel *c = sw->subchannel;
422 gpr_mu *mu = &c->mu;
423 int destroy;
424 grpc_transport_op op;
425 grpc_channel_element *elem;
426 connection *destroy_connection = NULL;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700427
428 gpr_mu_lock(mu);
429
430 /* if we failed or there is a version number mismatch, just leave
431 this closure */
432 if (!iomgr_success || sw->subchannel->active_version != sw->version) {
433 goto done;
434 }
435
Craig Tiller11bf14e2015-06-29 16:35:41 -0700436 gpr_log(GPR_DEBUG, "TRANSPORT STATE: %d", sw->connectivity_state);
437
Craig Tillerdf91ba52015-06-29 10:55:46 -0700438 switch (sw->connectivity_state) {
439 case GRPC_CHANNEL_CONNECTING:
440 case GRPC_CHANNEL_READY:
441 case GRPC_CHANNEL_IDLE:
442 /* all is still good: keep watching */
443 memset(&op, 0, sizeof(op));
444 op.connectivity_state = &sw->connectivity_state;
445 op.on_connectivity_state_change = &sw->closure;
Craig Tillerf62d6fc2015-06-29 10:55:59 -0700446 elem = grpc_channel_stack_element(
447 CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
Craig Tillerdf91ba52015-06-29 10:55:46 -0700448 elem->filter->start_transport_op(elem, &op);
449 /* early out */
450 gpr_mu_unlock(mu);
451 return;
452 case GRPC_CHANNEL_FATAL_FAILURE:
Craig Tiller49924e02015-06-29 22:42:33 -0700453 case GRPC_CHANNEL_TRANSIENT_FAILURE:
Craig Tillerdf91ba52015-06-29 10:55:46 -0700454 /* things have gone wrong, deactivate and enter idle */
455 if (sw->subchannel->active->refs == 0) {
456 destroy_connection = sw->subchannel->active;
457 }
458 sw->subchannel->active = NULL;
Craig Tiller11bf14e2015-06-29 16:35:41 -0700459 grpc_connectivity_state_set(&c->state_tracker,
460 GRPC_CHANNEL_TRANSIENT_FAILURE);
Craig Tillerdf91ba52015-06-29 10:55:46 -0700461 break;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700462 }
463
464done:
Craig Tiller11bf14e2015-06-29 16:35:41 -0700465 connectivity_state_changed_locked(c);
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700466 destroy = SUBCHANNEL_UNREF_LOCKED(c, "state_watcher");
Craig Tillerdf91ba52015-06-29 10:55:46 -0700467 gpr_free(sw);
468 gpr_mu_unlock(mu);
Craig Tillerdf91ba52015-06-29 10:55:46 -0700469 if (destroy) {
470 subchannel_destroy(c);
471 }
472 if (destroy_connection != NULL) {
473 connection_destroy(destroy_connection);
474 }
Craig Tillerc7b5f762015-06-27 11:48:42 -0700475}
476
Craig Tillerff54c922015-06-26 16:57:20 -0700477static void publish_transport(grpc_subchannel *c) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700478 size_t channel_stack_size;
479 connection *con;
480 grpc_channel_stack *stk;
481 size_t num_filters;
482 const grpc_channel_filter **filters;
483 waiting_for_connect *w4c;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700484 grpc_transport_op op;
485 state_watcher *sw;
486 connection *destroy_connection = NULL;
487 grpc_channel_element *elem;
Craig Tiller5945ee12015-06-27 10:36:09 -0700488
Craig Tillerdf91ba52015-06-29 10:55:46 -0700489 /* build final filter list */
Craig Tiller4ab82d22015-06-29 09:40:33 -0700490 num_filters = c->num_filters + c->connecting_result.num_filters + 1;
491 filters = gpr_malloc(sizeof(*filters) * num_filters);
492 memcpy(filters, c->filters, sizeof(*filters) * c->num_filters);
493 memcpy(filters + c->num_filters, c->connecting_result.filters,
494 sizeof(*filters) * c->connecting_result.num_filters);
495 filters[num_filters - 1] = &grpc_connected_channel_filter;
Craig Tiller5945ee12015-06-27 10:36:09 -0700496
Craig Tillerdf91ba52015-06-29 10:55:46 -0700497 /* construct channel stack */
Craig Tiller4ab82d22015-06-29 09:40:33 -0700498 channel_stack_size = grpc_channel_stack_size(filters, num_filters);
499 con = gpr_malloc(sizeof(connection) + channel_stack_size);
500 stk = (grpc_channel_stack *)(con + 1);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700501 con->refs = 0;
502 con->subchannel = c;
Craig Tiller079a11b2015-06-30 10:07:15 -0700503 grpc_channel_stack_init(filters, num_filters, c->master, c->args, c->mdctx,
504 stk);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700505 grpc_connected_channel_bind_transport(stk, c->connecting_result.transport);
Craig Tiller98465032015-06-29 14:36:42 -0700506 gpr_free(c->connecting_result.filters);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700507 memset(&c->connecting_result, 0, sizeof(c->connecting_result));
Craig Tillerff54c922015-06-26 16:57:20 -0700508
Craig Tillerdf91ba52015-06-29 10:55:46 -0700509 /* initialize state watcher */
510 sw = gpr_malloc(sizeof(*sw));
511 grpc_iomgr_closure_init(&sw->closure, on_state_changed, sw);
512 sw->subchannel = c;
513 sw->connectivity_state = GRPC_CHANNEL_READY;
514
Craig Tiller4ab82d22015-06-29 09:40:33 -0700515 gpr_mu_lock(&c->mu);
Craig Tillerdf91ba52015-06-29 10:55:46 -0700516
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700517 if (c->disconnected) {
518 gpr_mu_unlock(&c->mu);
519 gpr_free(sw);
520 gpr_free(filters);
521 grpc_channel_stack_destroy(stk);
522 return;
523 }
524
Craig Tillerdf91ba52015-06-29 10:55:46 -0700525 /* publish */
526 if (c->active != NULL && c->active->refs == 0) {
527 destroy_connection = c->active;
528 }
Craig Tiller4ab82d22015-06-29 09:40:33 -0700529 c->active = con;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700530 c->active_version++;
531 sw->version = c->active_version;
Craig Tiller4ab82d22015-06-29 09:40:33 -0700532 c->connecting = 0;
Craig Tillerdf91ba52015-06-29 10:55:46 -0700533
534 /* watch for changes; subchannel ref for connecting is donated
535 to the state watcher */
536 memset(&op, 0, sizeof(op));
537 op.connectivity_state = &sw->connectivity_state;
538 op.on_connectivity_state_change = &sw->closure;
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700539 SUBCHANNEL_REF_LOCKED(c, "state_watcher");
540 GPR_ASSERT(!SUBCHANNEL_UNREF_LOCKED(c, "connecting"));
Craig Tillerf62d6fc2015-06-29 10:55:59 -0700541 elem =
542 grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
Craig Tillerdf91ba52015-06-29 10:55:46 -0700543 elem->filter->start_transport_op(elem, &op);
544
545 /* signal completion */
Craig Tiller4ab82d22015-06-29 09:40:33 -0700546 connectivity_state_changed_locked(c);
547 while ((w4c = c->waiting)) {
Craig Tillerdf91ba52015-06-29 10:55:46 -0700548 c->waiting = w4c->next;
549 grpc_iomgr_add_callback(&w4c->continuation);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700550 }
Craig Tillerdf91ba52015-06-29 10:55:46 -0700551
Craig Tiller4ab82d22015-06-29 09:40:33 -0700552 gpr_mu_unlock(&c->mu);
Craig Tiller5945ee12015-06-27 10:36:09 -0700553
Craig Tiller4ab82d22015-06-29 09:40:33 -0700554 gpr_free(filters);
Craig Tillerd7b68e72015-06-28 11:41:09 -0700555
Craig Tillerdf91ba52015-06-29 10:55:46 -0700556 if (destroy_connection != NULL) {
557 connection_destroy(destroy_connection);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700558 }
559}
Craig Tillerff54c922015-06-26 16:57:20 -0700560
Craig Tillerff3ae682015-06-29 17:44:04 -0700561static void on_alarm(void *arg, int iomgr_success) {
562 grpc_subchannel *c = arg;
563 gpr_mu_lock(&c->mu);
564 c->have_alarm = 0;
Craig Tillerd2cc4592015-07-01 07:50:47 -0700565 if (c->disconnected) {
566 iomgr_success = 0;
567 }
Craig Tiller87cc0842015-06-30 08:15:55 -0700568 connectivity_state_changed_locked(c);
Craig Tillerff3ae682015-06-29 17:44:04 -0700569 gpr_mu_unlock(&c->mu);
570 if (iomgr_success) {
571 continue_connect(c);
572 } else {
573 GRPC_SUBCHANNEL_UNREF(c, "connecting");
574 }
575}
576
Craig Tillerff54c922015-06-26 16:57:20 -0700577static void subchannel_connected(void *arg, int iomgr_success) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700578 grpc_subchannel *c = arg;
Craig Tiller11bf14e2015-06-29 16:35:41 -0700579 if (c->connecting_result.transport != NULL) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700580 publish_transport(c);
581 } else {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700582 gpr_mu_lock(&c->mu);
Craig Tiller87cc0842015-06-30 08:15:55 -0700583 connectivity_state_changed_locked(c);
Craig Tillerff3ae682015-06-29 17:44:04 -0700584 GPR_ASSERT(!c->have_alarm);
585 c->have_alarm = 1;
586 c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta);
587 c->backoff_delta = gpr_time_add(c->backoff_delta, c->backoff_delta);
588 grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now());
Craig Tiller4ab82d22015-06-29 09:40:33 -0700589 gpr_mu_unlock(&c->mu);
Craig Tiller4ab82d22015-06-29 09:40:33 -0700590 }
Craig Tillerff54c922015-06-26 16:57:20 -0700591}
592
Craig Tiller5f84c842015-06-26 16:08:21 -0700593static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
Craig Tiller87cc0842015-06-30 08:15:55 -0700594 return gpr_time_add(c->next_attempt, c->backoff_delta);
Craig Tiller5f84c842015-06-26 16:08:21 -0700595}
596
597static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
Craig Tillerb6fbf1d2015-06-29 15:25:49 -0700598 if (c->disconnected) {
599 return GRPC_CHANNEL_FATAL_FAILURE;
600 }
Craig Tiller5f84c842015-06-26 16:08:21 -0700601 if (c->connecting) {
Craig Tiller87cc0842015-06-30 08:15:55 -0700602 if (c->have_alarm) {
603 return GRPC_CHANNEL_TRANSIENT_FAILURE;
604 }
Craig Tiller5f84c842015-06-26 16:08:21 -0700605 return GRPC_CHANNEL_CONNECTING;
606 }
607 if (c->active) {
608 return GRPC_CHANNEL_READY;
609 }
610 return GRPC_CHANNEL_IDLE;
611}
612
613static void connectivity_state_changed_locked(grpc_subchannel *c) {
614 grpc_connectivity_state current = compute_connectivity_locked(c);
Craig Tillerc7b5f762015-06-27 11:48:42 -0700615 grpc_connectivity_state_set(&c->state_tracker, current);
Craig Tiller5f84c842015-06-26 16:08:21 -0700616}
617
Craig Tiller2595ab72015-06-25 15:26:00 -0700618/*
619 * grpc_subchannel_call implementation
620 */
621
Craig Tiller079a11b2015-06-30 10:07:15 -0700622void grpc_subchannel_call_ref(
623 grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
624 gpr_ref(&c->refs);
Craig Tillerc3967532015-06-29 14:59:38 -0700625}
Craig Tiller2595ab72015-06-25 15:26:00 -0700626
Craig Tiller079a11b2015-06-30 10:07:15 -0700627void grpc_subchannel_call_unref(
628 grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
Craig Tillerca3e9d32015-06-27 18:37:27 -0700629 if (gpr_unref(&c->refs)) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700630 gpr_mu *mu = &c->connection->subchannel->mu;
631 grpc_subchannel *destroy;
Craig Tillerca3e9d32015-06-27 18:37:27 -0700632 grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(c));
Craig Tillerd7b68e72015-06-28 11:41:09 -0700633 gpr_mu_lock(mu);
Craig Tillerc3967532015-06-29 14:59:38 -0700634 destroy = CONNECTION_UNREF_LOCKED(c->connection, "call");
Craig Tillerd7b68e72015-06-28 11:41:09 -0700635 gpr_mu_unlock(mu);
Craig Tillerca3e9d32015-06-27 18:37:27 -0700636 gpr_free(c);
Craig Tillerc3967532015-06-29 14:59:38 -0700637 if (destroy != NULL) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700638 subchannel_destroy(destroy);
Craig Tillerd7b68e72015-06-28 11:41:09 -0700639 }
Craig Tiller2595ab72015-06-25 15:26:00 -0700640 }
641}
642
643void grpc_subchannel_call_process_op(grpc_subchannel_call *call,
644 grpc_transport_stream_op *op) {
645 grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call);
646 grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0);
647 top_elem->filter->start_transport_stream_op(top_elem, op);
648}
Craig Tillereb3b12e2015-06-26 14:42:49 -0700649
Craig Tillerabf36382015-06-29 16:13:27 -0700650grpc_subchannel_call *create_call(connection *con) {
Craig Tiller4ab82d22015-06-29 09:40:33 -0700651 grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
652 grpc_subchannel_call *call =
653 gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700654 grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call);
655 call->connection = con;
656 gpr_ref_init(&call->refs, 1);
Craig Tillerabf36382015-06-29 16:13:27 -0700657 grpc_call_stack_init(chanstk, NULL, NULL, callstk);
Craig Tiller04c5d4b2015-06-26 17:21:41 -0700658 return call;
Craig Tillereb3b12e2015-06-26 14:42:49 -0700659}