Further client config work
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 037f0c0..9637cf3 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -37,6 +37,13 @@
#include <grpc/support/alloc.h>
+#include "src/core/channel/channel_args.h"
+
+typedef struct {
+ gpr_refcount refs;
+ grpc_subchannel *subchannel;
+} connection;
+
struct grpc_subchannel {
gpr_refcount refs;
grpc_connector *connector;
@@ -49,24 +56,45 @@
/** address to connect to */
struct sockaddr *addr;
size_t addr_len;
+
+ /** set during connection */
+ grpc_transport *connecting_transport;
+
+ /** callback for connection finishing */
+ grpc_iomgr_closure connected;
+
+ /** mutex protecting remaining elements */
+ gpr_mu mu;
+
+ /** active connection */
+ connection *active;
+ /** are we connecting */
+ int connecting;
+ /** closures waiting for a connection */
+ grpc_iomgr_closure *waiting;
};
struct grpc_subchannel_call {
- grpc_subchannel *subchannel;
+ connection *connection;
gpr_refcount refs;
};
#define SUBCHANNEL_CALL_TO_CALL_STACK(call) (((grpc_call_stack *)(call)) + 1)
+static grpc_subchannel_call *create_call(connection *con, grpc_transport_stream_op *initial_op);
+
/*
* grpc_subchannel implementation
*/
-void grpc_subchannel_ref(grpc_subchannel *channel) { gpr_ref(&channel->refs); }
+void grpc_subchannel_ref(grpc_subchannel *c) { gpr_ref(&c->refs); }
-void grpc_subchannel_unref(grpc_subchannel *channel) {
- if (gpr_unref(&channel->refs)) {
- gpr_free(channel);
+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);
+ gpr_free(c);
}
}
@@ -84,9 +112,39 @@
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);
+ gpr_mu_init(&c->mu);
return c;
}
+void grpc_subchannel_create_call(grpc_subchannel *c,
+ grpc_mdctx *mdctx,
+ 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 {
+ notify->next = c->waiting;
+ c->waiting = notify;
+ if (!c->connecting) {
+ c->connecting = 1;
+ gpr_mu_unlock(&c->mu);
+
+ grpc_connector_connect(c->connector, c->args, mdctx, &c->connecting_transport, &c->connected);
+ } else {
+ gpr_mu_unlock(&c->mu);
+ }
+ }
+}
+
/*
* grpc_subchannel_call implementation
*/
@@ -98,7 +156,9 @@
void grpc_subchannel_call_unref(grpc_subchannel_call *call) {
if (gpr_unref(&call->refs)) {
grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(call));
- grpc_subchannel_unref(call->subchannel);
+ if (gpr_unref(&call->connection->refs)) {
+ gpr_free(call->connection);
+ }
gpr_free(call);
}
}
@@ -109,3 +169,8 @@
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) {
+ abort();
+ return NULL;
+}