src/core/client_config/subchannel.c - platform/external/grpc-grpc - Gitiles

 /*
  *
  * Copyright 2015, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met:
  *
  *     * Redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer.
  *     * Redistributions in binary form must reproduce the above
  * copyright notice, this list of conditions and the following disclaimer
  * in the documentation and/or other materials provided with the
  * distribution.
  *     * Neither the name of Google Inc. nor the names of its
  * contributors may be used to endorse or promote products derived from
  * this software without specific prior written permission.
  *
  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  *
  */

 #include "src/core/client_config/subchannel.h"

 #include <string.h>

 #include <grpc/support/alloc.h>

 #include "src/core/channel/channel_args.h"
 #include "src/core/channel/connected_channel.h"

 typedef struct {
   gpr_refcount refs;
   grpc_subchannel *subchannel;
 } connection;

 typedef struct waiting_for_connect {
   struct waiting_for_connect *next;
   grpc_iomgr_closure *notify;
   grpc_transport_stream_op *initial_op;
   grpc_subchannel_call **target;
 } waiting_for_connect;

 typedef struct connectivity_state_watcher {
   struct connectivity_state_watcher *next;
   grpc_iomgr_closure *notify;
   grpc_connectivity_state *current;
 } connectivity_state_watcher;

 struct grpc_subchannel {
   gpr_refcount refs;
   grpc_connector *connector;

   /** non-transport related channel filters */
   const grpc_channel_filter **filters;
   size_t filter_count;
   /** channel arguments */
   grpc_channel_args *args;
   /** address to connect to */
   struct sockaddr *addr;
   size_t addr_len;
   /** metadata context */
   grpc_mdctx *mdctx;

   /** set during connection */
   grpc_connect_out_args connecting_result;

   /** callback for connection finishing */
   grpc_iomgr_closure connected;

   /** pollset_set tracking who's interested in a connection
       being setup */
   grpc_pollset_set pollset_set;

   /** mutex protecting remaining elements */
   gpr_mu mu;

   /** active connection */
   connection *active;
   /** are we connecting */
   int connecting;
   /** things waiting for a connection */
   waiting_for_connect *waiting;
   /** things watching the connectivity state */
   connectivity_state_watcher *watchers;
 };

 struct grpc_subchannel_call {
   connection *connection;
   gpr_refcount refs;
 };

 #define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
 #define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)((con) + 1))

 static grpc_subchannel_call *create_call(connection *con, grpc_transport_stream_op *initial_op);
 static void connectivity_state_changed_locked(grpc_subchannel *c);
 static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c);
 static gpr_timespec compute_connect_deadline(grpc_subchannel *c);
 static void subchannel_connected(void *subchannel, int iomgr_success);

 /*
  * grpc_subchannel implementation
  */

 void grpc_subchannel_ref(grpc_subchannel *c) { gpr_ref(&c->refs); }

 void grpc_subchannel_unref(grpc_subchannel *c) {
   if (gpr_unref(&c->refs)) {
     gpr_free(c->filters);
     grpc_channel_args_destroy(c->args);
     gpr_free(c->addr);
     grpc_mdctx_unref(c->mdctx);
     grpc_pollset_set_destroy(&c->pollset_set);
     gpr_free(c);
   }
 }

 void grpc_subchannel_add_interested_party(grpc_subchannel *c,
                                           grpc_pollset *pollset) {
   grpc_pollset_set_add_pollset(&c->pollset_set, pollset);
 }

 void grpc_subchannel_del_interested_party(grpc_subchannel *c,
                                           grpc_pollset *pollset) {
   grpc_pollset_set_del_pollset(&c->pollset_set, pollset);
 }

 grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
                                         grpc_subchannel_args *args) {
   grpc_subchannel *c = gpr_malloc(sizeof(*c));
   memset(c, 0, sizeof(*c));
   gpr_ref_init(&c->refs, 1);
   c->connector = connector;
   grpc_connector_ref(c->connector);
   c->filter_count = args->filter_count + 1;
   c->filters = gpr_malloc(sizeof(grpc_channel_filter *) * c->filter_count);
   memcpy(c->filters, args->filters,
          sizeof(grpc_channel_filter *) * args->filter_count);
   c->filters[c->filter_count - 1] = &grpc_connected_channel_filter;
   c->addr = gpr_malloc(args->addr_len);
   memcpy(c->addr, args->addr, args->addr_len);
   c->addr_len = args->addr_len;
   c->args = grpc_channel_args_copy(args->args);
   c->mdctx = args->mdctx;
   grpc_mdctx_ref(c->mdctx);
   grpc_pollset_set_init(&c->pollset_set);
   grpc_iomgr_closure_init(&c->connected, subchannel_connected, c);
   gpr_mu_init(&c->mu);
   return c;
 }

 static void start_connect(grpc_subchannel *c) {
 	grpc_connect_in_args args;

 	args.interested_parties = &c->pollset_set;
 	args.addr = c->addr;
 	args.addr_len = c->addr_len;
 	args.deadline = compute_connect_deadline(c);
 	args.channel_args = c->args;
 	args.metadata_context = c->mdctx;

   grpc_connector_connect(c->connector, &args, &c->connecting_result, &c->connected);
 }

 void grpc_subchannel_create_call(grpc_subchannel *c,
                                  grpc_transport_stream_op *initial_op,
                                  grpc_subchannel_call **target,
                                  grpc_iomgr_closure *notify) {
   connection *con;
   gpr_mu_lock(&c->mu);
   if (c->active != NULL) {
     con = c->active;
     gpr_ref(&con->refs);
     gpr_mu_unlock(&c->mu);

     *target = create_call(con, initial_op);
     notify->cb(notify->cb_arg, 1);
   } else {
     waiting_for_connect *w4c = gpr_malloc(sizeof(*w4c));
     w4c->next = c->waiting;
     w4c->notify = notify;
     w4c->initial_op = initial_op;
     w4c->target = target;
     c->waiting = w4c;
     grpc_subchannel_add_interested_party(c, initial_op->bind_pollset);
     if (!c->connecting) {
       c->connecting = 1;
       connectivity_state_changed_locked(c);
       grpc_subchannel_ref(c);
       gpr_mu_unlock(&c->mu);

       start_connect(c);
     } else {
       gpr_mu_unlock(&c->mu);
     }
   }
 }

 grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
   grpc_connectivity_state state;
   gpr_mu_lock(&c->mu);
   state = compute_connectivity_locked(c);
   gpr_mu_unlock(&c->mu);
   return state;
 }

 void grpc_subchannel_notify_on_state_change(grpc_subchannel *c,
                                             grpc_connectivity_state *state,
                                             grpc_iomgr_closure *notify) {
   grpc_connectivity_state current;
   int do_connect = 0;
   connectivity_state_watcher *w = gpr_malloc(sizeof(*w));
   w->current = state;
   w->notify = notify;
   gpr_mu_lock(&c->mu);
   current = compute_connectivity_locked(c);
   if (current == GRPC_CHANNEL_IDLE) {
     current = GRPC_CHANNEL_CONNECTING;
     c->connecting = 1;
     do_connect = 1;
     grpc_subchannel_ref(c);
     connectivity_state_changed_locked(c);
   }
   if (current != *state) {
     gpr_mu_unlock(&c->mu);
     *state = current;
     grpc_iomgr_add_callback(notify);
     gpr_free(w);
   } else {
     w->next = c->watchers;
     c->watchers = w;
     gpr_mu_unlock(&c->mu);
   }
   if (do_connect) {
   	start_connect(c);
   }
 }

 static void publish_transport(grpc_subchannel *c) {
 	size_t channel_stack_size = grpc_channel_stack_size(c->filters, c->filter_count);
 	connection *con = gpr_malloc(sizeof(connection) + channel_stack_size);
 	grpc_channel_stack *stk = (grpc_channel_stack *)(con + 1);
 	waiting_for_connect *w4c;
 	gpr_ref_init(&con->refs, 1);
 	con->subchannel = c;
 	grpc_channel_stack_init(c->filters, c->filter_count, c->args, c->mdctx, stk);
 	grpc_connected_channel_bind_transport(stk, c->connecting_result.transport);
 	memset(&c->connecting_result, 0, sizeof(c->connecting_result));

 	gpr_mu_lock(&c->mu);
 	GPR_ASSERT(c->active == NULL);
 	c->active = con;
 	c->connecting = 0;
 	connectivity_state_changed_locked(c);
 	while ((w4c = c->waiting)) {
 		abort(); /* not implemented */
 	}
 	gpr_mu_unlock(&c->mu);
 }

 static void subchannel_connected(void *arg, int iomgr_success) {
 	grpc_subchannel *c = arg;
 	if (c->connecting_result.transport) {
 		publish_transport(c);
 	} else {
 		grpc_subchannel_unref(c);
 		/* TODO(ctiller): retry after sleeping */
 		abort();
 	}
 }

 static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
   return gpr_time_add(gpr_now(), gpr_time_from_seconds(60));
 }

 static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
   if (c->connecting) {
     return GRPC_CHANNEL_CONNECTING;
   }
   if (c->active) {
     return GRPC_CHANNEL_READY;
   }
   return GRPC_CHANNEL_IDLE;
 }

 static void connectivity_state_changed_locked(grpc_subchannel *c) {
   grpc_connectivity_state current = compute_connectivity_locked(c);
   connectivity_state_watcher *new = NULL;
   connectivity_state_watcher *w;
   while ((w = c->watchers)) {
     c->watchers = w->next;

     if (current != *w->current) {
       *w->current = current;
       grpc_iomgr_add_callback(w->notify);
       gpr_free(w);
     } else {
       w->next = new;
       new = w;
     }
   }
   c->watchers = new;
 }

 /*
  * grpc_subchannel_call implementation
  */

 void grpc_subchannel_call_ref(grpc_subchannel_call *call) {
   gpr_ref(&call->refs);
 }

 void grpc_subchannel_call_unref(grpc_subchannel_call *call) {
   if (gpr_unref(&call->refs)) {
     grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(call));
     if (gpr_unref(&call->connection->refs)) {
       gpr_free(call->connection);
     }
     gpr_free(call);
   }
 }

 void grpc_subchannel_call_process_op(grpc_subchannel_call *call,
                                      grpc_transport_stream_op *op) {
   grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call);
   grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0);
   top_elem->filter->start_transport_stream_op(top_elem, op);
 }

 grpc_subchannel_call *create_call(connection *con, grpc_transport_stream_op *initial_op) {
 	grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
   grpc_subchannel_call *call = gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
   grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call);
   call->connection = con;
   gpr_ref_init(&call->refs, 1);
   grpc_call_stack_init(chanstk, NULL, initial_op, callstk);
   return call;
 }
	/*
	*
	* Copyright 2015, Google Inc.
	* All rights reserved.
	*
	* Redistribution and use in source and binary forms, with or without
	* modification, are permitted provided that the following conditions are
	* met:
	*
	* * Redistributions of source code must retain the above copyright
	* notice, this list of conditions and the following disclaimer.
	* * Redistributions in binary form must reproduce the above
	* copyright notice, this list of conditions and the following disclaimer
	* in the documentation and/or other materials provided with the
	* distribution.
	* * Neither the name of Google Inc. nor the names of its
	* contributors may be used to endorse or promote products derived from
	* this software without specific prior written permission.
	*
	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
	* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
	* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
	* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
	* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
	* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	*
	*/

	#include "src/core/client_config/subchannel.h"

	#include <string.h>

	#include <grpc/support/alloc.h>

	#include "src/core/channel/channel_args.h"
	#include "src/core/channel/connected_channel.h"

	typedef struct {
	gpr_refcount refs;
	grpc_subchannel *subchannel;
	} connection;

	typedef struct waiting_for_connect {
	struct waiting_for_connect *next;
	grpc_iomgr_closure *notify;
	grpc_transport_stream_op *initial_op;
	grpc_subchannel_call **target;
	} waiting_for_connect;

	typedef struct connectivity_state_watcher {
	struct connectivity_state_watcher *next;
	grpc_iomgr_closure *notify;
	grpc_connectivity_state *current;
	} connectivity_state_watcher;

	struct grpc_subchannel {
	gpr_refcount refs;
	grpc_connector *connector;

	/** non-transport related channel filters */
	const grpc_channel_filter **filters;
	size_t filter_count;
	/** channel arguments */
	grpc_channel_args *args;
	/** address to connect to */
	struct sockaddr *addr;
	size_t addr_len;
	/** metadata context */
	grpc_mdctx *mdctx;

	/** set during connection */
	grpc_connect_out_args connecting_result;

	/** callback for connection finishing */
	grpc_iomgr_closure connected;

	/** pollset_set tracking who's interested in a connection
	being setup */
	grpc_pollset_set pollset_set;

	/** mutex protecting remaining elements */
	gpr_mu mu;

	/** active connection */
	connection *active;
	/** are we connecting */
	int connecting;
	/** things waiting for a connection */
	waiting_for_connect *waiting;
	/** things watching the connectivity state */
	connectivity_state_watcher *watchers;
	};

	struct grpc_subchannel_call {
	connection *connection;
	gpr_refcount refs;
	};

	#define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
	#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)((con) + 1))

	static grpc_subchannel_call create_call(connection con, grpc_transport_stream_op *initial_op);
	static void connectivity_state_changed_locked(grpc_subchannel *c);
	static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c);
	static gpr_timespec compute_connect_deadline(grpc_subchannel *c);
	static void subchannel_connected(void *subchannel, int iomgr_success);

	/*
	* grpc_subchannel implementation
	*/

	void grpc_subchannel_ref(grpc_subchannel *c) { gpr_ref(&c->refs); }

	void grpc_subchannel_unref(grpc_subchannel *c) {
	if (gpr_unref(&c->refs)) {
	gpr_free(c->filters);
	grpc_channel_args_destroy(c->args);
	gpr_free(c->addr);
	grpc_mdctx_unref(c->mdctx);
	grpc_pollset_set_destroy(&c->pollset_set);
	gpr_free(c);
	}
	}

	void grpc_subchannel_add_interested_party(grpc_subchannel *c,
	grpc_pollset *pollset) {
	grpc_pollset_set_add_pollset(&c->pollset_set, pollset);
	}

	void grpc_subchannel_del_interested_party(grpc_subchannel *c,
	grpc_pollset *pollset) {
	grpc_pollset_set_del_pollset(&c->pollset_set, pollset);
	}

	grpc_subchannel grpc_subchannel_create(grpc_connector connector,
	grpc_subchannel_args *args) {
	grpc_subchannel c = gpr_malloc(sizeof(c));
	memset(c, 0, sizeof(*c));
	gpr_ref_init(&c->refs, 1);
	c->connector = connector;
	grpc_connector_ref(c->connector);
	c->filter_count = args->filter_count + 1;
	c->filters = gpr_malloc(sizeof(grpc_channel_filter ) c->filter_count);
	memcpy(c->filters, args->filters,
	sizeof(grpc_channel_filter ) args->filter_count);
	c->filters[c->filter_count - 1] = &grpc_connected_channel_filter;
	c->addr = gpr_malloc(args->addr_len);
	memcpy(c->addr, args->addr, args->addr_len);
	c->addr_len = args->addr_len;
	c->args = grpc_channel_args_copy(args->args);
	c->mdctx = args->mdctx;
	grpc_mdctx_ref(c->mdctx);
	grpc_pollset_set_init(&c->pollset_set);
	grpc_iomgr_closure_init(&c->connected, subchannel_connected, c);
	gpr_mu_init(&c->mu);
	return c;
	}

	static void start_connect(grpc_subchannel *c) {
	grpc_connect_in_args args;

	args.interested_parties = &c->pollset_set;
	args.addr = c->addr;
	args.addr_len = c->addr_len;
	args.deadline = compute_connect_deadline(c);
	args.channel_args = c->args;
	args.metadata_context = c->mdctx;

	grpc_connector_connect(c->connector, &args, &c->connecting_result, &c->connected);
	}

	void grpc_subchannel_create_call(grpc_subchannel *c,
	grpc_transport_stream_op *initial_op,
	grpc_subchannel_call **target,
	grpc_iomgr_closure *notify) {
	connection *con;
	gpr_mu_lock(&c->mu);
	if (c->active != NULL) {
	con = c->active;
	gpr_ref(&con->refs);
	gpr_mu_unlock(&c->mu);

	*target = create_call(con, initial_op);
	notify->cb(notify->cb_arg, 1);
	} else {
	waiting_for_connect w4c = gpr_malloc(sizeof(w4c));
	w4c->next = c->waiting;
	w4c->notify = notify;
	w4c->initial_op = initial_op;
	w4c->target = target;
	c->waiting = w4c;
	grpc_subchannel_add_interested_party(c, initial_op->bind_pollset);
	if (!c->connecting) {
	c->connecting = 1;
	connectivity_state_changed_locked(c);
	grpc_subchannel_ref(c);
	gpr_mu_unlock(&c->mu);

	start_connect(c);
	} else {
	gpr_mu_unlock(&c->mu);
	}
	}
	}

	grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
	grpc_connectivity_state state;
	gpr_mu_lock(&c->mu);
	state = compute_connectivity_locked(c);
	gpr_mu_unlock(&c->mu);
	return state;
	}

	void grpc_subchannel_notify_on_state_change(grpc_subchannel *c,
	grpc_connectivity_state *state,
	grpc_iomgr_closure *notify) {
	grpc_connectivity_state current;
	int do_connect = 0;
	connectivity_state_watcher w = gpr_malloc(sizeof(w));
	w->current = state;
	w->notify = notify;
	gpr_mu_lock(&c->mu);
	current = compute_connectivity_locked(c);
	if (current == GRPC_CHANNEL_IDLE) {
	current = GRPC_CHANNEL_CONNECTING;
	c->connecting = 1;
	do_connect = 1;
	grpc_subchannel_ref(c);
	connectivity_state_changed_locked(c);
	}
	if (current != *state) {
	gpr_mu_unlock(&c->mu);
	*state = current;
	grpc_iomgr_add_callback(notify);
	gpr_free(w);
	} else {
	w->next = c->watchers;
	c->watchers = w;
	gpr_mu_unlock(&c->mu);
	}
	if (do_connect) {
	start_connect(c);
	}
	}

	static void publish_transport(grpc_subchannel *c) {
	size_t channel_stack_size = grpc_channel_stack_size(c->filters, c->filter_count);
	connection *con = gpr_malloc(sizeof(connection) + channel_stack_size);
	grpc_channel_stack stk = (grpc_channel_stack )(con + 1);
	waiting_for_connect *w4c;
	gpr_ref_init(&con->refs, 1);
	con->subchannel = c;
	grpc_channel_stack_init(c->filters, c->filter_count, c->args, c->mdctx, stk);
	grpc_connected_channel_bind_transport(stk, c->connecting_result.transport);
	memset(&c->connecting_result, 0, sizeof(c->connecting_result));

	gpr_mu_lock(&c->mu);
	GPR_ASSERT(c->active == NULL);
	c->active = con;
	c->connecting = 0;
	connectivity_state_changed_locked(c);
	while ((w4c = c->waiting)) {
	abort(); /* not implemented */
	}
	gpr_mu_unlock(&c->mu);
	}

	static void subchannel_connected(void *arg, int iomgr_success) {
	grpc_subchannel *c = arg;
	if (c->connecting_result.transport) {
	publish_transport(c);
	} else {
	grpc_subchannel_unref(c);
	/* TODO(ctiller): retry after sleeping */
	abort();
	}
	}

	static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
	return gpr_time_add(gpr_now(), gpr_time_from_seconds(60));
	}

	static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
	if (c->connecting) {
	return GRPC_CHANNEL_CONNECTING;
	}
	if (c->active) {
	return GRPC_CHANNEL_READY;
	}
	return GRPC_CHANNEL_IDLE;
	}

	static void connectivity_state_changed_locked(grpc_subchannel *c) {
	grpc_connectivity_state current = compute_connectivity_locked(c);
	connectivity_state_watcher *new = NULL;
	connectivity_state_watcher *w;
	while ((w = c->watchers)) {
	c->watchers = w->next;

	if (current != *w->current) {
	*w->current = current;
	grpc_iomgr_add_callback(w->notify);
	gpr_free(w);
	} else {
	w->next = new;
	new = w;
	}
	}
	c->watchers = new;
	}

	/*
	* grpc_subchannel_call implementation
	*/

	void grpc_subchannel_call_ref(grpc_subchannel_call *call) {
	gpr_ref(&call->refs);
	}

	void grpc_subchannel_call_unref(grpc_subchannel_call *call) {
	if (gpr_unref(&call->refs)) {
	grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(call));
	if (gpr_unref(&call->connection->refs)) {
	gpr_free(call->connection);
	}
	gpr_free(call);
	}
	}

	void grpc_subchannel_call_process_op(grpc_subchannel_call *call,
	grpc_transport_stream_op *op) {
	grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call);
	grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0);
	top_elem->filter->start_transport_stream_op(top_elem, op);
	}

	grpc_subchannel_call create_call(connection con, grpc_transport_stream_op *initial_op) {
	grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
	grpc_subchannel_call *call = gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
	grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call);
	call->connection = con;
	gpr_ref_init(&call->refs, 1);
	grpc_call_stack_init(chanstk, NULL, initial_op, callstk);
	return call;
	}