aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/ext/transport
diff options
context:
space:
mode:
authorGravatar Craig Tiller <ctiller@google.com>2016-12-06 20:27:38 -0800
committerGravatar Craig Tiller <ctiller@google.com>2016-12-06 20:27:38 -0800
commit385dd9a46294c5230dc22ce587050b30adc5f90f (patch)
treeacd6a645bacbf4089f83a5b07b0c3a2821830317 /src/core/ext/transport
parentddea41e666e2c23a3f1a38e0452e541baa6dddd3 (diff)
parent35769dc29c5fd55ccaf8a234d8704fe7dd7c7fdb (diff)
Merge branch 'slice_interning' into metadata_filter
Diffstat (limited to 'src/core/ext/transport')
-rw-r--r--src/core/ext/transport/chttp2/client/chttp2_connector.c270
-rw-r--r--src/core/ext/transport/chttp2/client/chttp2_connector.h52
-rw-r--r--src/core/ext/transport/chttp2/client/insecure/channel_create.c153
-rw-r--r--src/core/ext/transport/chttp2/client/secure/secure_channel_create.c221
-rw-r--r--src/core/ext/transport/chttp2/server/chttp2_server.c355
-rw-r--r--src/core/ext/transport/chttp2/server/chttp2_server.h78
-rw-r--r--src/core/ext/transport/chttp2/server/insecure/server_chttp2.c172
-rw-r--r--src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c337
-rw-r--r--src/core/ext/transport/chttp2/transport/chttp2_transport.c55
-rw-r--r--src/core/ext/transport/chttp2/transport/internal.h7
-rw-r--r--src/core/ext/transport/cronet/transport/cronet_transport.c8
11 files changed, 897 insertions, 811 deletions
diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.c b/src/core/ext/transport/chttp2/client/chttp2_connector.c
new file mode 100644
index 0000000000..f311b14f2b
--- /dev/null
+++ b/src/core/ext/transport/chttp2/client/chttp2_connector.c
@@ -0,0 +1,270 @@
+/*
+ *
+ * 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/ext/transport/chttp2/client/chttp2_connector.h"
+
+#include <grpc/grpc.h>
+
+#include <string.h>
+
+#include <grpc/slice_buffer.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/string_util.h>
+
+#include "src/core/ext/client_channel/connector.h"
+#include "src/core/ext/client_channel/http_connect_handshaker.h"
+#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/channel/handshaker.h"
+#include "src/core/lib/iomgr/tcp_client.h"
+#include "src/core/lib/security/transport/security_connector.h"
+#include "src/core/lib/slice/slice_internal.h"
+
+typedef struct {
+ grpc_connector base;
+
+ gpr_mu mu;
+ gpr_refcount refs;
+
+ bool shutdown;
+ bool connecting;
+
+ char *server_name;
+ grpc_chttp2_create_handshakers_func create_handshakers;
+ void *create_handshakers_user_data;
+
+ grpc_closure *notify;
+ grpc_connect_in_args args;
+ grpc_connect_out_args *result;
+ grpc_closure initial_string_sent;
+ grpc_slice_buffer initial_string_buffer;
+
+ grpc_endpoint *endpoint; // Non-NULL until handshaking starts.
+
+ grpc_closure connected;
+
+ grpc_handshake_manager *handshake_mgr;
+} chttp2_connector;
+
+static void chttp2_connector_ref(grpc_connector *con) {
+ chttp2_connector *c = (chttp2_connector *)con;
+ gpr_ref(&c->refs);
+}
+
+static void chttp2_connector_unref(grpc_exec_ctx *exec_ctx,
+ grpc_connector *con) {
+ chttp2_connector *c = (chttp2_connector *)con;
+ if (gpr_unref(&c->refs)) {
+ /* c->initial_string_buffer does not need to be destroyed */
+ gpr_mu_destroy(&c->mu);
+ // If handshaking is not yet in progress, destroy the endpoint.
+ // Otherwise, the handshaker will do this for us.
+ if (c->endpoint != NULL) grpc_endpoint_destroy(exec_ctx, c->endpoint);
+ gpr_free(c->server_name);
+ gpr_free(c);
+ }
+}
+
+static void chttp2_connector_shutdown(grpc_exec_ctx *exec_ctx,
+ grpc_connector *con) {
+ chttp2_connector *c = (chttp2_connector *)con;
+ gpr_mu_lock(&c->mu);
+ c->shutdown = true;
+ if (c->handshake_mgr != NULL) {
+ grpc_handshake_manager_shutdown(exec_ctx, c->handshake_mgr);
+ }
+ // If handshaking is not yet in progress, shutdown the endpoint.
+ // Otherwise, the handshaker will do this for us.
+ if (!c->connecting && c->endpoint != NULL) {
+ grpc_endpoint_shutdown(exec_ctx, c->endpoint);
+ }
+ gpr_mu_unlock(&c->mu);
+}
+
+static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ grpc_handshaker_args *args = arg;
+ chttp2_connector *c = args->user_data;
+ gpr_mu_lock(&c->mu);
+ if (error != GRPC_ERROR_NONE || c->shutdown) {
+ if (error == GRPC_ERROR_NONE) {
+ error = GRPC_ERROR_CREATE("connector shutdown");
+ // We were shut down after handshaking completed successfully, so
+ // destroy the endpoint here.
+ // TODO(ctiller): It is currently necessary to shutdown endpoints
+ // before destroying them, even if we know that there are no
+ // pending read/write callbacks. This should be fixed, at which
+ // point this can be removed.
+ grpc_endpoint_shutdown(exec_ctx, args->endpoint);
+ grpc_endpoint_destroy(exec_ctx, args->endpoint);
+ grpc_channel_args_destroy(exec_ctx, args->args);
+ grpc_slice_buffer_destroy_internal(exec_ctx, args->read_buffer);
+ gpr_free(args->read_buffer);
+ } else {
+ error = GRPC_ERROR_REF(error);
+ }
+ memset(c->result, 0, sizeof(*c->result));
+ } else {
+ c->result->transport =
+ grpc_create_chttp2_transport(exec_ctx, args->args, args->endpoint, 1);
+ GPR_ASSERT(c->result->transport);
+ grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport,
+ args->read_buffer);
+ c->result->channel_args = args->args;
+ }
+ grpc_closure *notify = c->notify;
+ c->notify = NULL;
+ grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
+ grpc_handshake_manager_destroy(exec_ctx, c->handshake_mgr);
+ c->handshake_mgr = NULL;
+ gpr_mu_unlock(&c->mu);
+ chttp2_connector_unref(exec_ctx, (grpc_connector *)c);
+}
+
+static void start_handshake_locked(grpc_exec_ctx *exec_ctx,
+ chttp2_connector *c) {
+ c->handshake_mgr = grpc_handshake_manager_create();
+ char *proxy_name = grpc_get_http_proxy_server();
+ if (proxy_name != NULL) {
+ grpc_handshake_manager_add(
+ c->handshake_mgr,
+ grpc_http_connect_handshaker_create(proxy_name, c->server_name));
+ gpr_free(proxy_name);
+ }
+ if (c->create_handshakers != NULL) {
+ c->create_handshakers(exec_ctx, c->create_handshakers_user_data,
+ c->handshake_mgr);
+ }
+ grpc_handshake_manager_do_handshake(
+ exec_ctx, c->handshake_mgr, c->endpoint, c->args.channel_args,
+ c->args.deadline, NULL /* acceptor */, on_handshake_done, c);
+ c->endpoint = NULL; // Endpoint handed off to handshake manager.
+}
+
+static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ chttp2_connector *c = arg;
+ gpr_mu_lock(&c->mu);
+ if (error != GRPC_ERROR_NONE || c->shutdown) {
+ if (error == GRPC_ERROR_NONE) {
+ error = GRPC_ERROR_CREATE("connector shutdown");
+ } else {
+ error = GRPC_ERROR_REF(error);
+ }
+ memset(c->result, 0, sizeof(*c->result));
+ grpc_closure *notify = c->notify;
+ c->notify = NULL;
+ grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
+ gpr_mu_unlock(&c->mu);
+ chttp2_connector_unref(exec_ctx, arg);
+ } else {
+ start_handshake_locked(exec_ctx, c);
+ gpr_mu_unlock(&c->mu);
+ }
+}
+
+static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
+ chttp2_connector *c = arg;
+ gpr_mu_lock(&c->mu);
+ GPR_ASSERT(c->connecting);
+ c->connecting = false;
+ if (error != GRPC_ERROR_NONE || c->shutdown) {
+ if (error == GRPC_ERROR_NONE) {
+ error = GRPC_ERROR_CREATE("connector shutdown");
+ } else {
+ error = GRPC_ERROR_REF(error);
+ }
+ memset(c->result, 0, sizeof(*c->result));
+ grpc_closure *notify = c->notify;
+ c->notify = NULL;
+ grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
+ if (c->endpoint != NULL) grpc_endpoint_shutdown(exec_ctx, c->endpoint);
+ gpr_mu_unlock(&c->mu);
+ chttp2_connector_unref(exec_ctx, arg);
+ } else {
+ GPR_ASSERT(c->endpoint != NULL);
+ if (!GRPC_SLICE_IS_EMPTY(c->args.initial_connect_string)) {
+ grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,
+ c);
+ grpc_slice_buffer_init(&c->initial_string_buffer);
+ grpc_slice_buffer_add(&c->initial_string_buffer,
+ c->args.initial_connect_string);
+ grpc_endpoint_write(exec_ctx, c->endpoint, &c->initial_string_buffer,
+ &c->initial_string_sent);
+ } else {
+ start_handshake_locked(exec_ctx, c);
+ }
+ gpr_mu_unlock(&c->mu);
+ }
+}
+
+static void chttp2_connector_connect(grpc_exec_ctx *exec_ctx,
+ grpc_connector *con,
+ const grpc_connect_in_args *args,
+ grpc_connect_out_args *result,
+ grpc_closure *notify) {
+ chttp2_connector *c = (chttp2_connector *)con;
+ gpr_mu_lock(&c->mu);
+ GPR_ASSERT(c->notify == NULL);
+ c->notify = notify;
+ c->args = *args;
+ c->result = result;
+ GPR_ASSERT(c->endpoint == NULL);
+ chttp2_connector_ref(con); // Ref taken for callback.
+ grpc_closure_init(&c->connected, connected, c);
+ GPR_ASSERT(!c->connecting);
+ c->connecting = true;
+ grpc_tcp_client_connect(exec_ctx, &c->connected, &c->endpoint,
+ args->interested_parties, args->channel_args,
+ args->addr, args->deadline);
+ gpr_mu_unlock(&c->mu);
+}
+
+static const grpc_connector_vtable chttp2_connector_vtable = {
+ chttp2_connector_ref, chttp2_connector_unref, chttp2_connector_shutdown,
+ chttp2_connector_connect};
+
+grpc_connector *grpc_chttp2_connector_create(
+ grpc_exec_ctx *exec_ctx, const char *server_name,
+ grpc_chttp2_create_handshakers_func create_handshakers,
+ void *create_handshakers_user_data) {
+ chttp2_connector *c = gpr_malloc(sizeof(*c));
+ memset(c, 0, sizeof(*c));
+ c->base.vtable = &chttp2_connector_vtable;
+ gpr_mu_init(&c->mu);
+ gpr_ref_init(&c->refs, 1);
+ c->server_name = gpr_strdup(server_name);
+ c->create_handshakers = create_handshakers;
+ c->create_handshakers_user_data = create_handshakers_user_data;
+ return &c->base;
+}
diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.h b/src/core/ext/transport/chttp2/client/chttp2_connector.h
new file mode 100644
index 0000000000..6c34ce1af1
--- /dev/null
+++ b/src/core/ext/transport/chttp2/client/chttp2_connector.h
@@ -0,0 +1,52 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_TRANSPORT_CHTTP2_CLIENT_CHTTP2_CONNECTOR_H
+#define GRPC_CORE_EXT_TRANSPORT_CHTTP2_CLIENT_CHTTP2_CONNECTOR_H
+
+#include "src/core/ext/client_channel/connector.h"
+#include "src/core/lib/channel/handshaker.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+
+typedef void (*grpc_chttp2_create_handshakers_func)(
+ grpc_exec_ctx* exec_ctx, void* user_data,
+ grpc_handshake_manager* handshake_mgr);
+
+/// If \a create_handshakers is non-NULL, it will be called with
+/// \a create_handshakers_user_data to add handshakers.
+grpc_connector* grpc_chttp2_connector_create(
+ grpc_exec_ctx* exec_ctx, const char* server_name,
+ grpc_chttp2_create_handshakers_func create_handshakers,
+ void* create_handshakers_user_data);
+
+#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_CLIENT_CHTTP2_CONNECTOR_H */
diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create.c b/src/core/ext/transport/chttp2/client/insecure/channel_create.c
index ad1dd29241..6aff9e75c8 100644
--- a/src/core/ext/transport/chttp2/client/insecure/channel_create.c
+++ b/src/core/ext/transport/chttp2/client/insecure/channel_create.c
@@ -33,138 +33,17 @@
#include <grpc/grpc.h>
-#include <stdlib.h>
#include <string.h>
-#include <grpc/slice.h>
-#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
+#include <grpc/support/string_util.h>
#include "src/core/ext/client_channel/client_channel.h"
-#include "src/core/ext/client_channel/http_connect_handshaker.h"
#include "src/core/ext/client_channel/resolver_registry.h"
-#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
-#include "src/core/lib/channel/channel_args.h"
-#include "src/core/lib/channel/compress_filter.h"
-#include "src/core/lib/channel/handshaker.h"
-#include "src/core/lib/channel/http_client_filter.h"
-#include "src/core/lib/iomgr/tcp_client.h"
+#include "src/core/ext/transport/chttp2/client/chttp2_connector.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/channel.h"
-//
-// connector
-//
-
-typedef struct {
- grpc_connector base;
- gpr_refcount refs;
-
- grpc_closure *notify;
- grpc_connect_in_args args;
- grpc_connect_out_args *result;
- grpc_closure initial_string_sent;
- grpc_slice_buffer initial_string_buffer;
-
- grpc_endpoint *tcp;
-
- grpc_closure connected;
-
- grpc_handshake_manager *handshake_mgr;
-} connector;
-
-static void connector_ref(grpc_connector *con) {
- connector *c = (connector *)con;
- gpr_ref(&c->refs);
-}
-
-static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
- connector *c = (connector *)con;
- if (gpr_unref(&c->refs)) {
- /* c->initial_string_buffer does not need to be destroyed */
- grpc_handshake_manager_destroy(exec_ctx, c->handshake_mgr);
- gpr_free(c);
- }
-}
-
-static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- connector_unref(exec_ctx, arg);
-}
-
-static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
- grpc_channel_args *args,
- grpc_slice_buffer *read_buffer, void *user_data,
- grpc_error *error) {
- connector *c = user_data;
- if (error != GRPC_ERROR_NONE) {
- grpc_channel_args_destroy(exec_ctx, args);
- gpr_free(read_buffer);
- } else {
- c->result->transport =
- grpc_create_chttp2_transport(exec_ctx, args, endpoint, 1);
- GPR_ASSERT(c->result->transport);
- grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport,
- read_buffer);
- c->result->channel_args = args;
- }
- grpc_closure *notify = c->notify;
- c->notify = NULL;
- grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
-}
-
-static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
- connector *c = arg;
- grpc_endpoint *tcp = c->tcp;
- if (tcp != NULL) {
- if (!GRPC_SLICE_IS_EMPTY(c->args.initial_connect_string)) {
- grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,
- c);
- grpc_slice_buffer_init(&c->initial_string_buffer);
- grpc_slice_buffer_add(&c->initial_string_buffer,
- c->args.initial_connect_string);
- connector_ref(arg);
- grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer,
- &c->initial_string_sent);
- } else {
- grpc_handshake_manager_do_handshake(
- exec_ctx, c->handshake_mgr, tcp, c->args.channel_args,
- c->args.deadline, NULL /* acceptor */, on_handshake_done, c);
- }
- } else {
- memset(c->result, 0, sizeof(*c->result));
- grpc_closure *notify = c->notify;
- c->notify = NULL;
- grpc_exec_ctx_sched(exec_ctx, notify, GRPC_ERROR_REF(error), NULL);
- }
-}
-
-static void connector_shutdown(grpc_exec_ctx *exec_ctx, grpc_connector *con) {}
-
-static void connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *con,
- const grpc_connect_in_args *args,
- grpc_connect_out_args *result,
- grpc_closure *notify) {
- connector *c = (connector *)con;
- GPR_ASSERT(c->notify == NULL);
- GPR_ASSERT(notify->cb);
- c->notify = notify;
- c->args = *args;
- c->result = result;
- c->tcp = NULL;
- grpc_closure_init(&c->connected, connected, c);
- grpc_tcp_client_connect(exec_ctx, &c->connected, &c->tcp,
- args->interested_parties, args->channel_args,
- args->addr, args->deadline);
-}
-
-static const grpc_connector_vtable connector_vtable = {
- connector_ref, connector_unref, connector_shutdown, connector_connect};
-
-//
-// client_channel_factory
-//
-
static void client_channel_factory_ref(
grpc_client_channel_factory *cc_factory) {}
@@ -174,20 +53,11 @@ static void client_channel_factory_unref(
static grpc_subchannel *client_channel_factory_create_subchannel(
grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
const grpc_subchannel_args *args) {
- connector *c = gpr_malloc(sizeof(*c));
- memset(c, 0, sizeof(*c));
- c->base.vtable = &connector_vtable;
- gpr_ref_init(&c->refs, 1);
- c->handshake_mgr = grpc_handshake_manager_create();
- char *proxy_name = grpc_get_http_proxy_server();
- if (proxy_name != NULL) {
- grpc_handshake_manager_add(
- c->handshake_mgr,
- grpc_http_connect_handshaker_create(proxy_name, args->server_name));
- gpr_free(proxy_name);
- }
- grpc_subchannel *s = grpc_subchannel_create(exec_ctx, &c->base, args);
- grpc_connector_unref(exec_ctx, &c->base);
+ grpc_connector *connector = grpc_chttp2_connector_create(
+ exec_ctx, args->server_name, NULL /* create_handshakers */,
+ NULL /* user_data */);
+ grpc_subchannel *s = grpc_subchannel_create(exec_ctx, connector, args);
+ grpc_connector_unref(exec_ctx, connector);
return s;
}
@@ -198,16 +68,14 @@ static grpc_channel *client_channel_factory_create_channel(
grpc_channel *channel =
grpc_channel_create(exec_ctx, target, args, GRPC_CLIENT_CHANNEL, NULL);
grpc_resolver *resolver = grpc_resolver_create(exec_ctx, target, args);
- if (!resolver) {
+ if (resolver == NULL) {
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel,
"client_channel_factory_create_channel");
return NULL;
}
-
grpc_client_channel_finish_initialization(
exec_ctx, grpc_channel_get_channel_stack(channel), resolver, cc_factory);
GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create_channel");
-
return channel;
}
@@ -230,16 +98,13 @@ grpc_channel *grpc_insecure_channel_create(const char *target,
GRPC_API_TRACE(
"grpc_insecure_channel_create(target=%p, args=%p, reserved=%p)", 3,
(target, args, reserved));
- GPR_ASSERT(!reserved);
-
+ GPR_ASSERT(reserved == NULL);
grpc_client_channel_factory *factory =
(grpc_client_channel_factory *)&client_channel_factory;
grpc_channel *channel = client_channel_factory_create_channel(
&exec_ctx, factory, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, args);
-
grpc_client_channel_factory_unref(&exec_ctx, factory);
grpc_exec_ctx_finish(&exec_ctx);
-
return channel != NULL ? channel : grpc_lame_client_channel_create(
target, GRPC_STATUS_INTERNAL,
"Failed to create client channel");
diff --git a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
index 92e7fb17f9..a39f73e90f 100644
--- a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
+++ b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
@@ -33,196 +33,19 @@
#include <grpc/grpc.h>
-#include <stdlib.h>
#include <string.h>
-#include <grpc/slice.h>
-#include <grpc/slice_buffer.h>
#include <grpc/support/alloc.h>
+#include <grpc/support/string_util.h>
#include "src/core/ext/client_channel/client_channel.h"
-#include "src/core/ext/client_channel/http_connect_handshaker.h"
#include "src/core/ext/client_channel/resolver_registry.h"
-#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/ext/transport/chttp2/client/chttp2_connector.h"
#include "src/core/lib/channel/channel_args.h"
-#include "src/core/lib/channel/handshaker.h"
-#include "src/core/lib/iomgr/tcp_client.h"
-#include "src/core/lib/security/context/security_context.h"
#include "src/core/lib/security/credentials/credentials.h"
-#include "src/core/lib/security/transport/auth_filters.h"
+#include "src/core/lib/security/transport/security_connector.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/channel.h"
-#include "src/core/lib/tsi/transport_security_interface.h"
-
-//
-// connector
-//
-
-typedef struct {
- grpc_connector base;
- gpr_refcount refs;
-
- grpc_channel_security_connector *security_connector;
-
- grpc_closure *notify;
- grpc_connect_in_args args;
- grpc_connect_out_args *result;
- grpc_closure initial_string_sent;
- grpc_slice_buffer initial_string_buffer;
-
- gpr_mu mu;
- grpc_endpoint *connecting_endpoint;
- grpc_endpoint *newly_connecting_endpoint;
-
- grpc_closure connected_closure;
-
- grpc_handshake_manager *handshake_mgr;
-
- // TODO(roth): Remove once we eliminate on_secure_handshake_done().
- grpc_channel_args *tmp_args;
-} connector;
-
-static void connector_ref(grpc_connector *con) {
- connector *c = (connector *)con;
- gpr_ref(&c->refs);
-}
-
-static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
- connector *c = (connector *)con;
- if (gpr_unref(&c->refs)) {
- /* c->initial_string_buffer does not need to be destroyed */
- grpc_channel_args_destroy(exec_ctx, c->tmp_args);
- grpc_handshake_manager_destroy(exec_ctx, c->handshake_mgr);
- gpr_free(c);
- }
-}
-
-static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_security_status status,
- grpc_endpoint *secure_endpoint,
- grpc_auth_context *auth_context) {
- connector *c = arg;
- gpr_mu_lock(&c->mu);
- grpc_error *error = GRPC_ERROR_NONE;
- if (c->connecting_endpoint == NULL) {
- memset(c->result, 0, sizeof(*c->result));
- gpr_mu_unlock(&c->mu);
- } else if (status != GRPC_SECURITY_OK) {
- error = grpc_error_set_int(GRPC_ERROR_CREATE("Secure handshake failed"),
- GRPC_ERROR_INT_SECURITY_STATUS, status);
- memset(c->result, 0, sizeof(*c->result));
- c->connecting_endpoint = NULL;
- gpr_mu_unlock(&c->mu);
- } else {
- grpc_arg auth_context_arg;
- c->connecting_endpoint = NULL;
- gpr_mu_unlock(&c->mu);
- c->result->transport = grpc_create_chttp2_transport(
- exec_ctx, c->args.channel_args, secure_endpoint, 1);
- grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, NULL);
- auth_context_arg = grpc_auth_context_to_arg(auth_context);
- c->result->channel_args =
- grpc_channel_args_copy_and_add(c->tmp_args, &auth_context_arg, 1);
- }
- grpc_closure *notify = c->notify;
- c->notify = NULL;
- grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
-}
-
-static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
- grpc_channel_args *args,
- grpc_slice_buffer *read_buffer, void *user_data,
- grpc_error *error) {
- connector *c = user_data;
- c->tmp_args = args;
- if (error != GRPC_ERROR_NONE) {
- gpr_free(read_buffer);
- grpc_closure *notify = c->notify;
- c->notify = NULL;
- grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
- } else {
- // TODO(roth, jboeuf): Convert security connector handshaking to use new
- // handshake API, and then move the code from on_secure_handshake_done()
- // into this function.
- grpc_channel_security_connector_do_handshake(
- exec_ctx, c->security_connector, endpoint, read_buffer,
- c->args.deadline, on_secure_handshake_done, c);
- }
-}
-
-static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- connector *c = arg;
- grpc_handshake_manager_do_handshake(
- exec_ctx, c->handshake_mgr, c->connecting_endpoint, c->args.channel_args,
- c->args.deadline, NULL /* acceptor */, on_handshake_done, c);
-}
-
-static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
- connector *c = arg;
- grpc_endpoint *tcp = c->newly_connecting_endpoint;
- if (tcp != NULL) {
- gpr_mu_lock(&c->mu);
- GPR_ASSERT(c->connecting_endpoint == NULL);
- c->connecting_endpoint = tcp;
- gpr_mu_unlock(&c->mu);
- if (!GRPC_SLICE_IS_EMPTY(c->args.initial_connect_string)) {
- grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,
- c);
- grpc_slice_buffer_init(&c->initial_string_buffer);
- grpc_slice_buffer_add(&c->initial_string_buffer,
- c->args.initial_connect_string);
- grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer,
- &c->initial_string_sent);
- } else {
- grpc_handshake_manager_do_handshake(
- exec_ctx, c->handshake_mgr, tcp, c->args.channel_args,
- c->args.deadline, NULL /* acceptor */, on_handshake_done, c);
- }
- } else {
- memset(c->result, 0, sizeof(*c->result));
- grpc_closure *notify = c->notify;
- c->notify = NULL;
- grpc_exec_ctx_sched(exec_ctx, notify, GRPC_ERROR_REF(error), NULL);
- }
-}
-
-static void connector_shutdown(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
- connector *c = (connector *)con;
- grpc_endpoint *ep;
- gpr_mu_lock(&c->mu);
- ep = c->connecting_endpoint;
- c->connecting_endpoint = NULL;
- gpr_mu_unlock(&c->mu);
- if (ep) {
- grpc_endpoint_shutdown(exec_ctx, ep);
- }
-}
-
-static void connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *con,
- const grpc_connect_in_args *args,
- grpc_connect_out_args *result,
- grpc_closure *notify) {
- connector *c = (connector *)con;
- GPR_ASSERT(c->notify == NULL);
- c->notify = notify;
- c->args = *args;
- c->result = result;
- gpr_mu_lock(&c->mu);
- GPR_ASSERT(c->connecting_endpoint == NULL);
- gpr_mu_unlock(&c->mu);
- grpc_closure_init(&c->connected_closure, connected, c);
- grpc_tcp_client_connect(
- exec_ctx, &c->connected_closure, &c->newly_connecting_endpoint,
- args->interested_parties, args->channel_args, args->addr, args->deadline);
-}
-
-static const grpc_connector_vtable connector_vtable = {
- connector_ref, connector_unref, connector_shutdown, connector_connect};
-
-//
-// client_channel_factory
-//
typedef struct {
grpc_client_channel_factory base;
@@ -246,26 +69,21 @@ static void client_channel_factory_unref(
}
}
+static void create_handshakers(grpc_exec_ctx *exec_ctx,
+ void *security_connector,
+ grpc_handshake_manager *handshake_mgr) {
+ grpc_channel_security_connector_create_handshakers(
+ exec_ctx, security_connector, handshake_mgr);
+}
+
static grpc_subchannel *client_channel_factory_create_subchannel(
grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
const grpc_subchannel_args *args) {
client_channel_factory *f = (client_channel_factory *)cc_factory;
- connector *c = gpr_malloc(sizeof(*c));
- memset(c, 0, sizeof(*c));
- c->base.vtable = &connector_vtable;
- c->security_connector = f->security_connector;
- c->handshake_mgr = grpc_handshake_manager_create();
- char *proxy_name = grpc_get_http_proxy_server();
- if (proxy_name != NULL) {
- grpc_handshake_manager_add(
- c->handshake_mgr,
- grpc_http_connect_handshaker_create(proxy_name, args->server_name));
- gpr_free(proxy_name);
- }
- gpr_mu_init(&c->mu);
- gpr_ref_init(&c->refs, 1);
- grpc_subchannel *s = grpc_subchannel_create(exec_ctx, &c->base, args);
- grpc_connector_unref(exec_ctx, &c->base);
+ grpc_connector *connector = grpc_chttp2_connector_create(
+ exec_ctx, args->server_name, create_handshakers, f->security_connector);
+ grpc_subchannel *s = grpc_subchannel_create(exec_ctx, connector, args);
+ grpc_connector_unref(exec_ctx, connector);
return s;
}
@@ -277,15 +95,14 @@ static grpc_channel *client_channel_factory_create_channel(
grpc_channel *channel =
grpc_channel_create(exec_ctx, target, args, GRPC_CLIENT_CHANNEL, NULL);
grpc_resolver *resolver = grpc_resolver_create(exec_ctx, target, args);
- if (resolver != NULL) {
- grpc_client_channel_finish_initialization(
- exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base);
- GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create");
- } else {
+ if (resolver == NULL) {
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel,
"client_channel_factory_create_channel");
- channel = NULL;
+ return NULL;
}
+ grpc_client_channel_finish_initialization(
+ exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base);
+ GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create_channel");
return channel;
}
diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.c b/src/core/ext/transport/chttp2/server/chttp2_server.c
new file mode 100644
index 0000000000..4319454407
--- /dev/null
+++ b/src/core/ext/transport/chttp2/server/chttp2_server.c
@@ -0,0 +1,355 @@
+/*
+ *
+ * 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/ext/transport/chttp2/server/chttp2_server.h"
+
+#include <grpc/grpc.h>
+
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/useful.h>
+
+#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/channel/handshaker.h"
+#include "src/core/lib/channel/http_server_filter.h"
+#include "src/core/lib/iomgr/endpoint.h"
+#include "src/core/lib/iomgr/resolve_address.h"
+#include "src/core/lib/iomgr/tcp_server.h"
+#include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/surface/api_trace.h"
+#include "src/core/lib/surface/server.h"
+
+void grpc_chttp2_server_handshaker_factory_create_handshakers(
+ grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory,
+ grpc_handshake_manager *handshake_mgr) {
+ if (handshaker_factory != NULL) {
+ handshaker_factory->vtable->create_handshakers(exec_ctx, handshaker_factory,
+ handshake_mgr);
+ }
+}
+
+void grpc_chttp2_server_handshaker_factory_destroy(
+ grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory) {
+ if (handshaker_factory != NULL) {
+ handshaker_factory->vtable->destroy(exec_ctx, handshaker_factory);
+ }
+}
+
+typedef struct pending_handshake_manager_node {
+ grpc_handshake_manager *handshake_mgr;
+ struct pending_handshake_manager_node *next;
+} pending_handshake_manager_node;
+
+typedef struct {
+ grpc_server *server;
+ grpc_tcp_server *tcp_server;
+ grpc_channel_args *args;
+ grpc_chttp2_server_handshaker_factory *handshaker_factory;
+ gpr_mu mu;
+ bool shutdown;
+ grpc_closure tcp_server_shutdown_complete;
+ grpc_closure *server_destroy_listener_done;
+ pending_handshake_manager_node *pending_handshake_mgrs;
+} server_state;
+
+typedef struct {
+ server_state *server_state;
+ grpc_pollset *accepting_pollset;
+ grpc_tcp_server_acceptor *acceptor;
+ grpc_handshake_manager *handshake_mgr;
+} server_connection_state;
+
+static void pending_handshake_manager_add_locked(
+ server_state *state, grpc_handshake_manager *handshake_mgr) {
+ pending_handshake_manager_node *node = gpr_malloc(sizeof(*node));
+ node->handshake_mgr = handshake_mgr;
+ node->next = state->pending_handshake_mgrs;
+ state->pending_handshake_mgrs = node;
+}
+
+static void pending_handshake_manager_remove_locked(
+ server_state *state, grpc_handshake_manager *handshake_mgr) {
+ pending_handshake_manager_node **prev_node = &state->pending_handshake_mgrs;
+ for (pending_handshake_manager_node *node = state->pending_handshake_mgrs;
+ node != NULL; node = node->next) {
+ if (node->handshake_mgr == handshake_mgr) {
+ *prev_node = node->next;
+ gpr_free(node);
+ break;
+ }
+ prev_node = &node->next;
+ }
+}
+
+static void pending_handshake_manager_shutdown_locked(grpc_exec_ctx *exec_ctx,
+ server_state *state) {
+ pending_handshake_manager_node *prev_node = NULL;
+ for (pending_handshake_manager_node *node = state->pending_handshake_mgrs;
+ node != NULL; node = node->next) {
+ grpc_handshake_manager_shutdown(exec_ctx, node->handshake_mgr);
+ gpr_free(prev_node);
+ prev_node = node;
+ }
+ gpr_free(prev_node);
+ state->pending_handshake_mgrs = NULL;
+}
+
+static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ grpc_handshaker_args *args = arg;
+ server_connection_state *connection_state = args->user_data;
+ gpr_mu_lock(&connection_state->server_state->mu);
+ if (error != GRPC_ERROR_NONE || connection_state->server_state->shutdown) {
+ const char *error_str = grpc_error_string(error);
+ gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str);
+ grpc_error_free_string(error_str);
+ if (error == GRPC_ERROR_NONE) {
+ // We were shut down after handshaking completed successfully, so
+ // destroy the endpoint here.
+ // TODO(ctiller): It is currently necessary to shutdown endpoints
+ // before destroying them, even if we know that there are no
+ // pending read/write callbacks. This should be fixed, at which
+ // point this can be removed.
+ grpc_endpoint_shutdown(exec_ctx, args->endpoint);
+ grpc_endpoint_destroy(exec_ctx, args->endpoint);
+ grpc_channel_args_destroy(exec_ctx, args->args);
+ grpc_slice_buffer_destroy_internal(exec_ctx, args->read_buffer);
+ gpr_free(args->read_buffer);
+ }
+ } else {
+ grpc_transport *transport =
+ grpc_create_chttp2_transport(exec_ctx, args->args, args->endpoint, 0);
+ grpc_server_setup_transport(
+ exec_ctx, connection_state->server_state->server, transport,
+ connection_state->accepting_pollset, args->args);
+ grpc_chttp2_transport_start_reading(exec_ctx, transport, args->read_buffer);
+ grpc_channel_args_destroy(exec_ctx, args->args);
+ }
+ pending_handshake_manager_remove_locked(connection_state->server_state,
+ connection_state->handshake_mgr);
+ gpr_mu_unlock(&connection_state->server_state->mu);
+ grpc_handshake_manager_destroy(exec_ctx, connection_state->handshake_mgr);
+ grpc_tcp_server_unref(exec_ctx, connection_state->server_state->tcp_server);
+ gpr_free(connection_state);
+}
+
+static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp,
+ grpc_pollset *accepting_pollset,
+ grpc_tcp_server_acceptor *acceptor) {
+ server_state *state = arg;
+ gpr_mu_lock(&state->mu);
+ if (state->shutdown) {
+ gpr_mu_unlock(&state->mu);
+ grpc_endpoint_destroy(exec_ctx, tcp);
+ return;
+ }
+ grpc_handshake_manager *handshake_mgr = grpc_handshake_manager_create();
+ pending_handshake_manager_add_locked(state, handshake_mgr);
+ gpr_mu_unlock(&state->mu);
+ grpc_tcp_server_ref(state->tcp_server);
+ server_connection_state *connection_state =
+ gpr_malloc(sizeof(*connection_state));
+ connection_state->server_state = state;
+ connection_state->accepting_pollset = accepting_pollset;
+ connection_state->acceptor = acceptor;
+ connection_state->handshake_mgr = handshake_mgr;
+ grpc_chttp2_server_handshaker_factory_create_handshakers(
+ exec_ctx, state->handshaker_factory, connection_state->handshake_mgr);
+ // TODO(roth): We should really get this timeout value from channel
+ // args instead of hard-coding it.
+ const gpr_timespec deadline = gpr_time_add(
+ gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(120, GPR_TIMESPAN));
+ grpc_handshake_manager_do_handshake(exec_ctx, connection_state->handshake_mgr,
+ tcp, state->args, deadline, acceptor,
+ on_handshake_done, connection_state);
+}
+
+/* Server callback: start listening on our ports */
+static void server_start_listener(grpc_exec_ctx *exec_ctx, grpc_server *server,
+ void *arg, grpc_pollset **pollsets,
+ size_t pollset_count) {
+ server_state *state = arg;
+ gpr_mu_lock(&state->mu);
+ state->shutdown = false;
+ gpr_mu_unlock(&state->mu);
+ grpc_tcp_server_start(exec_ctx, state->tcp_server, pollsets, pollset_count,
+ on_accept, state);
+}
+
+static void tcp_server_shutdown_complete(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_error *error) {
+ server_state *state = arg;
+ /* ensure all threads have unlocked */
+ gpr_mu_lock(&state->mu);
+ grpc_closure *destroy_done = state->server_destroy_listener_done;
+ GPR_ASSERT(state->shutdown);
+ pending_handshake_manager_shutdown_locked(exec_ctx, state);
+ gpr_mu_unlock(&state->mu);
+ // Flush queued work before destroying handshaker factory, since that
+ // may do a synchronous unref.
+ grpc_exec_ctx_flush(exec_ctx);
+ grpc_chttp2_server_handshaker_factory_destroy(exec_ctx,
+ state->handshaker_factory);
+ if (destroy_done != NULL) {
+ destroy_done->cb(exec_ctx, destroy_done->cb_arg, GRPC_ERROR_REF(error));
+ grpc_exec_ctx_flush(exec_ctx);
+ }
+ grpc_channel_args_destroy(exec_ctx, state->args);
+ gpr_mu_destroy(&state->mu);
+ gpr_free(state);
+}
+
+/* Server callback: destroy the tcp listener (so we don't generate further
+ callbacks) */
+static void server_destroy_listener(grpc_exec_ctx *exec_ctx,
+ grpc_server *server, void *arg,
+ grpc_closure *destroy_done) {
+ server_state *state = arg;
+ gpr_mu_lock(&state->mu);
+ state->shutdown = true;
+ state->server_destroy_listener_done = destroy_done;
+ grpc_tcp_server *tcp_server = state->tcp_server;
+ gpr_mu_unlock(&state->mu);
+ grpc_tcp_server_shutdown_listeners(exec_ctx, tcp_server);
+ grpc_tcp_server_unref(exec_ctx, tcp_server);
+}
+
+grpc_error *grpc_chttp2_server_add_port(
+ grpc_exec_ctx *exec_ctx, grpc_server *server, const char *addr,
+ grpc_channel_args *args,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory, int *port_num) {
+ grpc_resolved_addresses *resolved = NULL;
+ grpc_tcp_server *tcp_server = NULL;
+ size_t i;
+ size_t count = 0;
+ int port_temp;
+ grpc_error *err = GRPC_ERROR_NONE;
+ server_state *state = NULL;
+ grpc_error **errors = NULL;
+
+ *port_num = -1;
+
+ /* resolve address */
+ err = grpc_blocking_resolve_address(addr, "https", &resolved);
+ if (err != GRPC_ERROR_NONE) {
+ goto error;
+ }
+ state = gpr_malloc(sizeof(*state));
+ memset(state, 0, sizeof(*state));
+ grpc_closure_init(&state->tcp_server_shutdown_complete,
+ tcp_server_shutdown_complete, state);
+ err = grpc_tcp_server_create(exec_ctx, &state->tcp_server_shutdown_complete,
+ args, &tcp_server);
+ if (err != GRPC_ERROR_NONE) {
+ goto error;
+ }
+
+ state->server = server;
+ state->tcp_server = tcp_server;
+ state->args = args;
+ state->handshaker_factory = handshaker_factory;
+ state->shutdown = true;
+ gpr_mu_init(&state->mu);
+
+ const size_t naddrs = resolved->naddrs;
+ errors = gpr_malloc(sizeof(*errors) * naddrs);
+ for (i = 0; i < naddrs; i++) {
+ errors[i] =
+ grpc_tcp_server_add_port(tcp_server, &resolved->addrs[i], &port_temp);
+ if (errors[i] == GRPC_ERROR_NONE) {
+ if (*port_num == -1) {
+ *port_num = port_temp;
+ } else {
+ GPR_ASSERT(*port_num == port_temp);
+ }
+ count++;
+ }
+ }
+ if (count == 0) {
+ char *msg;
+ gpr_asprintf(&msg, "No address added out of total %" PRIuPTR " resolved",
+ naddrs);
+ err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, naddrs);
+ gpr_free(msg);
+ goto error;
+ } else if (count != naddrs) {
+ char *msg;
+ gpr_asprintf(&msg, "Only %" PRIuPTR
+ " addresses added out of total %" PRIuPTR " resolved",
+ count, naddrs);
+ err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, naddrs);
+ gpr_free(msg);
+
+ const char *warning_message = grpc_error_string(err);
+ gpr_log(GPR_INFO, "WARNING: %s", warning_message);
+ grpc_error_free_string(warning_message);
+ /* we managed to bind some addresses: continue */
+ }
+ grpc_resolved_addresses_destroy(resolved);
+
+ /* Register with the server only upon success */
+ grpc_server_add_listener(exec_ctx, server, state, server_start_listener,
+ server_destroy_listener);
+ goto done;
+
+/* Error path: cleanup and return */
+error:
+ GPR_ASSERT(err != GRPC_ERROR_NONE);
+ if (resolved) {
+ grpc_resolved_addresses_destroy(resolved);
+ }
+ if (tcp_server) {
+ grpc_tcp_server_unref(exec_ctx, tcp_server);
+ } else {
+ grpc_channel_args_destroy(exec_ctx, args);
+ grpc_chttp2_server_handshaker_factory_destroy(exec_ctx, handshaker_factory);
+ gpr_free(state);
+ }
+ *port_num = 0;
+
+done:
+ if (errors != NULL) {
+ for (i = 0; i < naddrs; i++) {
+ GRPC_ERROR_UNREF(errors[i]);
+ }
+ gpr_free(errors);
+ }
+ return err;
+}
diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.h b/src/core/ext/transport/chttp2/server/chttp2_server.h
new file mode 100644
index 0000000000..3073399267
--- /dev/null
+++ b/src/core/ext/transport/chttp2/server/chttp2_server.h
@@ -0,0 +1,78 @@
+/*
+ *
+ * Copyright 2016, 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.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_TRANSPORT_CHTTP2_SERVER_CHTTP2_SERVER_H
+#define GRPC_CORE_EXT_TRANSPORT_CHTTP2_SERVER_CHTTP2_SERVER_H
+
+#include <grpc/impl/codegen/grpc_types.h>
+
+#include "src/core/lib/channel/handshaker.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+
+/// A server handshaker factory is used to create handshakers for server
+/// connections.
+typedef struct grpc_chttp2_server_handshaker_factory
+ grpc_chttp2_server_handshaker_factory;
+
+typedef struct {
+ void (*create_handshakers)(
+ grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory,
+ grpc_handshake_manager *handshake_mgr);
+ void (*destroy)(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory);
+} grpc_chttp2_server_handshaker_factory_vtable;
+
+struct grpc_chttp2_server_handshaker_factory {
+ const grpc_chttp2_server_handshaker_factory_vtable *vtable;
+};
+
+void grpc_chttp2_server_handshaker_factory_create_handshakers(
+ grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory,
+ grpc_handshake_manager *handshake_mgr);
+
+void grpc_chttp2_server_handshaker_factory_destroy(
+ grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory);
+
+/// Adds a port to \a server. Sets \a port_num to the port number.
+/// If \a handshaker_factory is not NULL, it will be used to create
+/// handshakers for the port.
+/// Takes ownership of \a args and \a handshaker_factory.
+grpc_error *grpc_chttp2_server_add_port(
+ grpc_exec_ctx *exec_ctx, grpc_server *server, const char *addr,
+ grpc_channel_args *args,
+ grpc_chttp2_server_handshaker_factory *handshaker_factory, int *port_num);
+
+#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_SERVER_CHTTP2_SERVER_H */
diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
index ce7cd7586b..7e286d4e46 100644
--- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
+++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
@@ -33,180 +33,28 @@
#include <grpc/grpc.h>
-#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
-#include <grpc/support/string_util.h>
-#include <grpc/support/useful.h>
-#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
+#include "src/core/ext/transport/chttp2/server/chttp2_server.h"
#include "src/core/lib/channel/channel_args.h"
-#include "src/core/lib/channel/handshaker.h"
-#include "src/core/lib/channel/http_server_filter.h"
-#include "src/core/lib/iomgr/resolve_address.h"
-#include "src/core/lib/iomgr/tcp_server.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/server.h"
-typedef struct server_connect_state {
- grpc_server *server;
- grpc_pollset *accepting_pollset;
- grpc_tcp_server_acceptor *acceptor;
- grpc_handshake_manager *handshake_mgr;
-} server_connect_state;
-
-static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
- grpc_channel_args *args,
- grpc_slice_buffer *read_buffer, void *user_data,
- grpc_error *error) {
- server_connect_state *state = user_data;
- if (error != GRPC_ERROR_NONE) {
- const char *error_str = grpc_error_string(error);
- gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str);
- grpc_error_free_string(error_str);
- GRPC_ERROR_UNREF(error);
- grpc_handshake_manager_shutdown(exec_ctx, state->handshake_mgr);
- gpr_free(read_buffer);
- } else {
- // Beware that the call to grpc_create_chttp2_transport() has to happen
- // before grpc_tcp_server_destroy(). This is fine here, but similar code
- // asynchronously doing a handshake instead of calling
- // grpc_tcp_server_start() (as in server_secure_chttp2.c) needs to add
- // synchronization to avoid this case.
- grpc_transport *transport =
- grpc_create_chttp2_transport(exec_ctx, args, endpoint, 0);
- grpc_server_setup_transport(exec_ctx, state->server, transport,
- state->accepting_pollset,
- grpc_server_get_channel_args(state->server));
- grpc_chttp2_transport_start_reading(exec_ctx, transport, read_buffer);
- }
- // Clean up.
- grpc_channel_args_destroy(exec_ctx, args);
- grpc_handshake_manager_destroy(exec_ctx, state->handshake_mgr);
- gpr_free(state);
-}
-
-static void on_accept(grpc_exec_ctx *exec_ctx, void *server, grpc_endpoint *tcp,
- grpc_pollset *accepting_pollset,
- grpc_tcp_server_acceptor *acceptor) {
- server_connect_state *state = gpr_malloc(sizeof(server_connect_state));
- state->server = server;
- state->accepting_pollset = accepting_pollset;
- state->acceptor = acceptor;
- state->handshake_mgr = grpc_handshake_manager_create();
- // TODO(roth): We should really get this timeout value from channel
- // args instead of hard-coding it.
- const gpr_timespec deadline = gpr_time_add(
- gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(120, GPR_TIMESPAN));
- grpc_handshake_manager_do_handshake(
- exec_ctx, state->handshake_mgr, tcp, grpc_server_get_channel_args(server),
- deadline, acceptor, on_handshake_done, state);
-}
-
-/* Server callback: start listening on our ports */
-static void start(grpc_exec_ctx *exec_ctx, grpc_server *server, void *tcpp,
- grpc_pollset **pollsets, size_t pollset_count) {
- grpc_tcp_server *tcp = tcpp;
- grpc_tcp_server_start(exec_ctx, tcp, pollsets, pollset_count, on_accept,
- server);
-}
-
-/* Server callback: destroy the tcp listener (so we don't generate further
- callbacks) */
-static void destroy(grpc_exec_ctx *exec_ctx, grpc_server *server, void *tcpp,
- grpc_closure *destroy_done) {
- grpc_tcp_server *tcp = tcpp;
- grpc_tcp_server_shutdown_listeners(exec_ctx, tcp);
- grpc_tcp_server_unref(exec_ctx, tcp);
- grpc_exec_ctx_sched(exec_ctx, destroy_done, GRPC_ERROR_NONE, NULL);
-}
-
int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) {
- grpc_resolved_addresses *resolved = NULL;
- grpc_tcp_server *tcp = NULL;
- size_t i;
- size_t count = 0;
- int port_num = -1;
- int port_temp;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- grpc_error *err = GRPC_ERROR_NONE;
-
+ int port_num = 0;
GRPC_API_TRACE("grpc_server_add_insecure_http2_port(server=%p, addr=%s)", 2,
(server, addr));
-
- grpc_error **errors = NULL;
- err = grpc_blocking_resolve_address(addr, "https", &resolved);
- if (err != GRPC_ERROR_NONE) {
- goto error;
- }
-
- err = grpc_tcp_server_create(&exec_ctx, NULL,
- grpc_server_get_channel_args(server), &tcp);
+ grpc_error *err = grpc_chttp2_server_add_port(
+ &exec_ctx, server, addr,
+ grpc_channel_args_copy(grpc_server_get_channel_args(server)),
+ NULL /* handshaker_factory */, &port_num);
if (err != GRPC_ERROR_NONE) {
- goto error;
- }
-
- const size_t naddrs = resolved->naddrs;
- errors = gpr_malloc(sizeof(*errors) * naddrs);
- for (i = 0; i < naddrs; i++) {
- errors[i] = grpc_tcp_server_add_port(tcp, &resolved->addrs[i], &port_temp);
- if (errors[i] == GRPC_ERROR_NONE) {
- if (port_num == -1) {
- port_num = port_temp;
- } else {
- GPR_ASSERT(port_num == port_temp);
- }
- count++;
- }
+ const char *msg = grpc_error_string(err);
+ gpr_log(GPR_ERROR, "%s", msg);
+ grpc_error_free_string(msg);
+ GRPC_ERROR_UNREF(err);
}
- if (count == 0) {
- char *msg;
- gpr_asprintf(&msg, "No address added out of total %" PRIuPTR " resolved",
- naddrs);
- err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, naddrs);
- gpr_free(msg);
- goto error;
- } else if (count != naddrs) {
- char *msg;
- gpr_asprintf(&msg, "Only %" PRIuPTR
- " addresses added out of total %" PRIuPTR " resolved",
- count, naddrs);
- err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, naddrs);
- gpr_free(msg);
-
- const char *warning_message = grpc_error_string(err);
- gpr_log(GPR_INFO, "WARNING: %s", warning_message);
- grpc_error_free_string(warning_message);
- /* we managed to bind some addresses: continue */
- }
- grpc_resolved_addresses_destroy(resolved);
-
- /* Register with the server only upon success */
- grpc_server_add_listener(&exec_ctx, server, tcp, start, destroy);
- goto done;
-
-/* Error path: cleanup and return */
-error:
- GPR_ASSERT(err != GRPC_ERROR_NONE);
- if (resolved) {
- grpc_resolved_addresses_destroy(resolved);
- }
- if (tcp) {
- grpc_tcp_server_unref(&exec_ctx, tcp);
- }
- port_num = 0;
-
- const char *msg = grpc_error_string(err);
- gpr_log(GPR_ERROR, "%s", msg);
- grpc_error_free_string(msg);
- GRPC_ERROR_UNREF(err);
-
-done:
grpc_exec_ctx_finish(&exec_ctx);
- if (errors != NULL) {
- for (i = 0; i < naddrs; i++) {
- GRPC_ERROR_UNREF(errors[i]);
- }
- }
- gpr_free(errors);
return port_num;
}
diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
index 6293e06b69..a08fb2a18e 100644
--- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
+++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
@@ -38,218 +38,62 @@
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
-#include <grpc/support/sync.h>
-#include <grpc/support/useful.h>
+
+#include "src/core/ext/transport/chttp2/server/chttp2_server.h"
+
#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/handshaker.h"
-#include "src/core/lib/channel/http_server_filter.h"
-#include "src/core/lib/iomgr/endpoint.h"
-#include "src/core/lib/iomgr/resolve_address.h"
-#include "src/core/lib/iomgr/tcp_server.h"
#include "src/core/lib/security/context/security_context.h"
#include "src/core/lib/security/credentials/credentials.h"
-#include "src/core/lib/security/transport/auth_filters.h"
-#include "src/core/lib/security/transport/security_connector.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/server.h"
-typedef struct server_secure_state {
- grpc_server *server;
- grpc_tcp_server *tcp;
- grpc_server_security_connector *sc;
- grpc_server_credentials *creds;
- bool is_shutdown;
- gpr_mu mu;
- grpc_closure tcp_server_shutdown_complete;
- grpc_closure *server_destroy_listener_done;
-} server_secure_state;
-
-typedef struct server_secure_connect {
- server_secure_state *server_state;
- grpc_pollset *accepting_pollset;
- grpc_tcp_server_acceptor *acceptor;
- grpc_handshake_manager *handshake_mgr;
- // TODO(roth): Remove the following two fields when we eliminate
- // grpc_server_security_connector_do_handshake().
- gpr_timespec deadline;
- grpc_channel_args *args;
-} server_secure_connect;
-
-static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep,
- grpc_security_status status,
- grpc_endpoint *secure_endpoint,
- grpc_auth_context *auth_context) {
- server_secure_connect *connection_state = statep;
- if (status == GRPC_SECURITY_OK) {
- if (secure_endpoint) {
- gpr_mu_lock(&connection_state->server_state->mu);
- if (!connection_state->server_state->is_shutdown) {
- grpc_transport *transport = grpc_create_chttp2_transport(
- exec_ctx, grpc_server_get_channel_args(
- connection_state->server_state->server),
- secure_endpoint, 0);
- grpc_arg args_to_add[2];
- args_to_add[0] = grpc_server_credentials_to_arg(
- connection_state->server_state->creds);
- args_to_add[1] = grpc_auth_context_to_arg(auth_context);
- grpc_channel_args *args_copy = grpc_channel_args_copy_and_add(
- connection_state->args, args_to_add, GPR_ARRAY_SIZE(args_to_add));
- grpc_server_setup_transport(
- exec_ctx, connection_state->server_state->server, transport,
- connection_state->accepting_pollset, args_copy);
- grpc_channel_args_destroy(exec_ctx, args_copy);
- grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL);
- } else {
- /* We need to consume this here, because the server may already have
- * gone away. */
- grpc_endpoint_destroy(exec_ctx, secure_endpoint);
- }
- gpr_mu_unlock(&connection_state->server_state->mu);
- }
- } else {
- gpr_log(GPR_ERROR, "Secure transport failed with error %d", status);
- }
- grpc_channel_args_destroy(exec_ctx, connection_state->args);
- grpc_tcp_server_unref(exec_ctx, connection_state->server_state->tcp);
- gpr_free(connection_state);
-}
-
-static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
- grpc_channel_args *args,
- grpc_slice_buffer *read_buffer, void *user_data,
- grpc_error *error) {
- server_secure_connect *connection_state = user_data;
- if (error != GRPC_ERROR_NONE) {
- const char *error_str = grpc_error_string(error);
- gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str);
- grpc_error_free_string(error_str);
- GRPC_ERROR_UNREF(error);
- grpc_channel_args_destroy(exec_ctx, args);
- gpr_free(read_buffer);
- grpc_handshake_manager_shutdown(exec_ctx, connection_state->handshake_mgr);
- grpc_handshake_manager_destroy(exec_ctx, connection_state->handshake_mgr);
- grpc_tcp_server_unref(exec_ctx, connection_state->server_state->tcp);
- gpr_free(connection_state);
- return;
- }
- grpc_handshake_manager_destroy(exec_ctx, connection_state->handshake_mgr);
- connection_state->handshake_mgr = NULL;
- // TODO(roth, jboeuf): Convert security connector handshaking to use new
- // handshake API, and then move the code from on_secure_handshake_done()
- // into this function.
- connection_state->args = args;
- grpc_server_security_connector_do_handshake(
- exec_ctx, connection_state->server_state->sc, connection_state->acceptor,
- endpoint, read_buffer, connection_state->deadline,
- on_secure_handshake_done, connection_state);
-}
-
-static void on_accept(grpc_exec_ctx *exec_ctx, void *statep, grpc_endpoint *tcp,
- grpc_pollset *accepting_pollset,
- grpc_tcp_server_acceptor *acceptor) {
- server_secure_state *server_state = statep;
- server_secure_connect *connection_state = NULL;
- gpr_mu_lock(&server_state->mu);
- if (server_state->is_shutdown) {
- gpr_mu_unlock(&server_state->mu);
- grpc_endpoint_destroy(exec_ctx, tcp);
- return;
- }
- gpr_mu_unlock(&server_state->mu);
- grpc_tcp_server_ref(server_state->tcp);
- connection_state = gpr_malloc(sizeof(*connection_state));
- connection_state->server_state = server_state;
- connection_state->accepting_pollset = accepting_pollset;
- connection_state->acceptor = acceptor;
- connection_state->handshake_mgr = grpc_handshake_manager_create();
- // TODO(roth): We should really get this timeout value from channel
- // args instead of hard-coding it.
- connection_state->deadline = gpr_time_add(
- gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(120, GPR_TIMESPAN));
- grpc_handshake_manager_do_handshake(
- exec_ctx, connection_state->handshake_mgr, tcp,
- grpc_server_get_channel_args(connection_state->server_state->server),
- connection_state->deadline, acceptor, on_handshake_done,
- connection_state);
+typedef struct {
+ grpc_chttp2_server_handshaker_factory base;
+ grpc_server_security_connector *security_connector;
+} server_security_handshaker_factory;
+
+static void server_security_handshaker_factory_create_handshakers(
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_server_handshaker_factory *hf,
+ grpc_handshake_manager *handshake_mgr) {
+ server_security_handshaker_factory *handshaker_factory =
+ (server_security_handshaker_factory *)hf;
+ grpc_server_security_connector_create_handshakers(
+ exec_ctx, handshaker_factory->security_connector, handshake_mgr);
}
-/* Server callback: start listening on our ports */
-static void server_start_listener(grpc_exec_ctx *exec_ctx, grpc_server *server,
- void *statep, grpc_pollset **pollsets,
- size_t pollset_count) {
- server_secure_state *server_state = statep;
- gpr_mu_lock(&server_state->mu);
- server_state->is_shutdown = false;
- gpr_mu_unlock(&server_state->mu);
- grpc_tcp_server_start(exec_ctx, server_state->tcp, pollsets, pollset_count,
- on_accept, server_state);
+static void server_security_handshaker_factory_destroy(
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_server_handshaker_factory *hf) {
+ server_security_handshaker_factory *handshaker_factory =
+ (server_security_handshaker_factory *)hf;
+ GRPC_SECURITY_CONNECTOR_UNREF(
+ exec_ctx, &handshaker_factory->security_connector->base, "server");
+ gpr_free(hf);
}
-static void tcp_server_shutdown_complete(grpc_exec_ctx *exec_ctx, void *statep,
- grpc_error *error) {
- server_secure_state *server_state = statep;
- /* ensure all threads have unlocked */
- gpr_mu_lock(&server_state->mu);
- grpc_closure *destroy_done = server_state->server_destroy_listener_done;
- GPR_ASSERT(server_state->is_shutdown);
- gpr_mu_unlock(&server_state->mu);
- /* clean up */
- grpc_server_security_connector_shutdown(exec_ctx, server_state->sc);
-
- /* Flush queued work before a synchronous unref. */
- grpc_exec_ctx_flush(exec_ctx);
- GRPC_SECURITY_CONNECTOR_UNREF(exec_ctx, &server_state->sc->base, "server");
- grpc_server_credentials_unref(exec_ctx, server_state->creds);
-
- if (destroy_done != NULL) {
- destroy_done->cb(exec_ctx, destroy_done->cb_arg, GRPC_ERROR_REF(error));
- grpc_exec_ctx_flush(exec_ctx);
- }
- gpr_free(server_state);
-}
-
-static void server_destroy_listener(grpc_exec_ctx *exec_ctx,
- grpc_server *server, void *statep,
- grpc_closure *callback) {
- server_secure_state *server_state = statep;
- grpc_tcp_server *tcp;
- gpr_mu_lock(&server_state->mu);
- server_state->is_shutdown = true;
- server_state->server_destroy_listener_done = callback;
- tcp = server_state->tcp;
- gpr_mu_unlock(&server_state->mu);
- grpc_tcp_server_shutdown_listeners(exec_ctx, tcp);
- grpc_tcp_server_unref(exec_ctx, server_state->tcp);
-}
+static const grpc_chttp2_server_handshaker_factory_vtable
+ server_security_handshaker_factory_vtable = {
+ server_security_handshaker_factory_create_handshakers,
+ server_security_handshaker_factory_destroy};
int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,
grpc_server_credentials *creds) {
- grpc_resolved_addresses *resolved = NULL;
- grpc_tcp_server *tcp = NULL;
- server_secure_state *server_state = NULL;
- size_t i;
- size_t count = 0;
- int port_num = -1;
- int port_temp;
- grpc_security_status status = GRPC_SECURITY_ERROR;
- grpc_server_security_connector *sc = NULL;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_error *err = GRPC_ERROR_NONE;
- grpc_error **errors = NULL;
-
+ grpc_server_security_connector *sc = NULL;
+ int port_num = 0;
GRPC_API_TRACE(
"grpc_server_add_secure_http2_port("
"server=%p, addr=%s, creds=%p)",
3, (server, addr, creds));
-
- /* create security context */
+ // Create security context.
if (creds == NULL) {
err = GRPC_ERROR_CREATE(
"No credentials specified for secure server port (creds==NULL)");
- goto error;
+ goto done;
}
- status =
+ grpc_security_status status =
grpc_server_credentials_create_security_connector(&exec_ctx, creds, &sc);
if (status != GRPC_SECURITY_OK) {
char *msg;
@@ -259,107 +103,28 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr,
err = grpc_error_set_int(GRPC_ERROR_CREATE(msg),
GRPC_ERROR_INT_SECURITY_STATUS, status);
gpr_free(msg);
- goto error;
+ goto done;
}
- sc->channel_args = grpc_server_get_channel_args(server);
-
- /* resolve address */
- err = grpc_blocking_resolve_address(addr, "https", &resolved);
- if (err != GRPC_ERROR_NONE) {
- goto error;
- }
- server_state = gpr_malloc(sizeof(*server_state));
- memset(server_state, 0, sizeof(*server_state));
- grpc_closure_init(&server_state->tcp_server_shutdown_complete,
- tcp_server_shutdown_complete, server_state);
- err = grpc_tcp_server_create(&exec_ctx,
- &server_state->tcp_server_shutdown_complete,
- grpc_server_get_channel_args(server), &tcp);
+ // Create handshaker factory.
+ server_security_handshaker_factory *handshaker_factory =
+ gpr_malloc(sizeof(*handshaker_factory));
+ memset(handshaker_factory, 0, sizeof(*handshaker_factory));
+ handshaker_factory->base.vtable = &server_security_handshaker_factory_vtable;
+ handshaker_factory->security_connector = sc;
+ // Create channel args.
+ grpc_arg channel_arg = grpc_server_credentials_to_arg(creds);
+ grpc_channel_args *args = grpc_channel_args_copy_and_add(
+ grpc_server_get_channel_args(server), &channel_arg, 1);
+ // Add server port.
+ err = grpc_chttp2_server_add_port(&exec_ctx, server, addr, args,
+ &handshaker_factory->base, &port_num);
+done:
+ grpc_exec_ctx_finish(&exec_ctx);
if (err != GRPC_ERROR_NONE) {
- goto error;
+ const char *msg = grpc_error_string(err);
+ gpr_log(GPR_ERROR, "%s", msg);
+ grpc_error_free_string(msg);
+ GRPC_ERROR_UNREF(err);
}
-
- server_state->server = server;
- server_state->tcp = tcp;
- server_state->sc = sc;
- server_state->creds = grpc_server_credentials_ref(creds);
- server_state->is_shutdown = true;
- gpr_mu_init(&server_state->mu);
-
- errors = gpr_malloc(sizeof(*errors) * resolved->naddrs);
- for (i = 0; i < resolved->naddrs; i++) {
- errors[i] = grpc_tcp_server_add_port(tcp, &resolved->addrs[i], &port_temp);
- if (errors[i] == GRPC_ERROR_NONE) {
- if (port_num == -1) {
- port_num = port_temp;
- } else {
- GPR_ASSERT(port_num == port_temp);
- }
- count++;
- }
- }
- if (count == 0) {
- char *msg;
- gpr_asprintf(&msg, "No address added out of total %" PRIuPTR " resolved",
- resolved->naddrs);
- err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, resolved->naddrs);
- gpr_free(msg);
- goto error;
- } else if (count != resolved->naddrs) {
- char *msg;
- gpr_asprintf(&msg, "Only %" PRIuPTR
- " addresses added out of total %" PRIuPTR " resolved",
- count, resolved->naddrs);
- err = GRPC_ERROR_CREATE_REFERENCING(msg, errors, resolved->naddrs);
- gpr_free(msg);
-
- const char *warning_message = grpc_error_string(err);
- gpr_log(GPR_INFO, "WARNING: %s", warning_message);
- grpc_error_free_string(warning_message);
- /* we managed to bind some addresses: continue */
- } else {
- for (i = 0; i < resolved->naddrs; i++) {
- GRPC_ERROR_UNREF(errors[i]);
- }
- }
- gpr_free(errors);
- errors = NULL;
- grpc_resolved_addresses_destroy(resolved);
-
- /* Register with the server only upon success */
- grpc_server_add_listener(&exec_ctx, server, server_state,
- server_start_listener, server_destroy_listener);
-
- grpc_exec_ctx_finish(&exec_ctx);
return port_num;
-
-/* Error path: cleanup and return */
-error:
- GPR_ASSERT(err != GRPC_ERROR_NONE);
- if (errors != NULL) {
- for (i = 0; i < resolved->naddrs; i++) {
- GRPC_ERROR_UNREF(errors[i]);
- }
- gpr_free(errors);
- }
- if (resolved) {
- grpc_resolved_addresses_destroy(resolved);
- }
- if (tcp) {
- grpc_tcp_server_unref(&exec_ctx, tcp);
- } else {
- if (sc) {
- grpc_exec_ctx_flush(&exec_ctx);
- GRPC_SECURITY_CONNECTOR_UNREF(&exec_ctx, &sc->base, "server");
- }
- if (server_state) {
- gpr_free(server_state);
- }
- }
- grpc_exec_ctx_finish(&exec_ctx);
- const char *msg = grpc_error_string(err);
- GRPC_ERROR_UNREF(err);
- gpr_log(GPR_ERROR, "%s", msg);
- grpc_error_free_string(msg);
- return 0;
}
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index f84e5a610b..d3f43dd571 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -112,9 +112,6 @@ static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx,
static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx,
void *byte_stream,
grpc_error *error_ignored);
-static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_chttp2_stream *s,
- grpc_error *error);
static void benign_reclaimer(grpc_exec_ctx *exec_ctx, void *t,
grpc_error *error);
@@ -605,11 +602,13 @@ static void set_write_state(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
write_state_name(t->write_state),
write_state_name(st), reason));
t->write_state = st;
- if (st == GRPC_CHTTP2_WRITE_STATE_IDLE &&
- t->close_transport_on_writes_finished != NULL) {
- grpc_error *err = t->close_transport_on_writes_finished;
- t->close_transport_on_writes_finished = NULL;
- close_transport_locked(exec_ctx, t, err);
+ if (st == GRPC_CHTTP2_WRITE_STATE_IDLE) {
+ grpc_exec_ctx_enqueue_list(exec_ctx, &t->run_after_write, NULL);
+ if (t->close_transport_on_writes_finished != NULL) {
+ grpc_error *err = t->close_transport_on_writes_finished;
+ t->close_transport_on_writes_finished = NULL;
+ close_transport_locked(exec_ctx, t, err);
+ }
}
}
@@ -707,8 +706,6 @@ static void write_action_end_locked(grpc_exec_ctx *exec_ctx, void *tp,
}
}
- grpc_chttp2_end_write(exec_ctx, t, GRPC_ERROR_REF(error));
-
switch (t->write_state) {
case GRPC_CHTTP2_WRITE_STATE_IDLE:
GPR_UNREACHABLE_CODE(break);
@@ -737,6 +734,8 @@ static void write_action_end_locked(grpc_exec_ctx *exec_ctx, void *tp,
break;
}
+ grpc_chttp2_end_write(exec_ctx, t, GRPC_ERROR_REF(error));
+
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "writing");
GPR_TIMER_END("terminate_writing_with_lock", 0);
}
@@ -826,7 +825,14 @@ static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx,
}
}
+/* Flag that this closure barrier wants stats to be updated before finishing */
#define CLOSURE_BARRIER_STATS_BIT (1 << 0)
+/* Flag that this closure barrier may be covering a write in a pollset, and so
+ we should not complete this closure until we can prove that the write got
+ scheduled */
+#define CLOSURE_BARRIER_MAY_COVER_WRITE (1 << 1)
+/* First bit of the reference count, stored in the high order bits (with the low
+ bits being used for flags defined above) */
#define CLOSURE_BARRIER_FIRST_REF_BIT (1 << 16)
static grpc_closure *add_closure_barrier(grpc_closure *closure) {
@@ -853,6 +859,16 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
return;
}
closure->next_data.scratch -= CLOSURE_BARRIER_FIRST_REF_BIT;
+ if (grpc_http_trace) {
+ const char *errstr = grpc_error_string(error);
+ gpr_log(GPR_DEBUG,
+ "complete_closure_step: %p refs=%d flags=0x%04x desc=%s err=%s",
+ closure,
+ (int)(closure->next_data.scratch / CLOSURE_BARRIER_FIRST_REF_BIT),
+ (int)(closure->next_data.scratch % CLOSURE_BARRIER_FIRST_REF_BIT),
+ desc, errstr);
+ grpc_error_free_string(errstr);
+ }
if (error != GRPC_ERROR_NONE) {
if (closure->error_data.error == GRPC_ERROR_NONE) {
closure->error_data.error =
@@ -869,7 +885,13 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
grpc_transport_move_stats(&s->stats, s->collecting_stats);
s->collecting_stats = NULL;
}
- grpc_closure_run(exec_ctx, closure, closure->error_data.error);
+ if ((t->write_state == GRPC_CHTTP2_WRITE_STATE_IDLE) ||
+ !(closure->next_data.scratch & CLOSURE_BARRIER_MAY_COVER_WRITE)) {
+ grpc_closure_run(exec_ctx, closure, closure->error_data.error);
+ } else {
+ grpc_closure_list_append(&t->run_after_write, closure,
+ closure->error_data.error);
+ }
}
}
@@ -1017,6 +1039,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
if (op->send_initial_metadata != NULL) {
GPR_ASSERT(s->send_initial_metadata_finished == NULL);
+ on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
s->send_initial_metadata_finished = add_closure_barrier(on_complete);
s->send_initial_metadata = op->send_initial_metadata;
const size_t metadata_size =
@@ -1070,6 +1093,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
}
if (op->send_message != NULL) {
+ on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
s->fetching_send_message_finished = add_closure_barrier(op->on_complete);
if (s->write_closed) {
grpc_chttp2_complete_closure_step(
@@ -1107,6 +1131,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
if (op->send_trailing_metadata != NULL) {
GPR_ASSERT(s->send_trailing_metadata_finished == NULL);
+ on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
s->send_trailing_metadata_finished = add_closure_barrier(on_complete);
s->send_trailing_metadata = op->send_trailing_metadata;
const size_t metadata_size =
@@ -1539,9 +1564,9 @@ static grpc_error *removal_error(grpc_error *extra_error, grpc_chttp2_stream *s,
return error;
}
-static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_chttp2_stream *s,
- grpc_error *error) {
+void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, grpc_error *error) {
error =
removal_error(error, s, "Pending writes failed due to stream closure");
s->send_initial_metadata = NULL;
@@ -1595,7 +1620,7 @@ void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx,
if (close_writes && !s->write_closed) {
s->write_closed_error = GRPC_ERROR_REF(error);
s->write_closed = true;
- fail_pending_writes(exec_ctx, t, s, GRPC_ERROR_REF(error));
+ grpc_chttp2_fail_pending_writes(exec_ctx, t, s, GRPC_ERROR_REF(error));
grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s);
}
if (s->read_closed && s->write_closed) {
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index b74233d992..31eb1e01ac 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -327,6 +327,9 @@ struct grpc_chttp2_transport {
*/
grpc_error *close_transport_on_writes_finished;
+ /* a list of closures to run after writes are finished */
+ grpc_closure_list run_after_write;
+
/* buffer pool state */
/** have we scheduled a benign cleanup? */
bool benign_reclaimer_registered;
@@ -689,4 +692,8 @@ void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t,
grpc_chttp2_stream *s);
+void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, grpc_error *error);
+
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index cf9a35194a..a5996b448f 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -418,6 +418,7 @@ static void on_response_headers_received(
cronet_bidirectional_stream *stream,
const cronet_bidirectional_stream_header_array *headers,
const char *negotiated_protocol) {
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
CRONET_LOG(GPR_DEBUG, "R: on_response_headers_received(%p, %p, %s)", stream,
headers, negotiated_protocol);
stream_obj *s = (stream_obj *)stream->annotation;
@@ -429,11 +430,12 @@ static void on_response_headers_received(
grpc_chttp2_incoming_metadata_buffer_add(
&s->state.rs.initial_metadata,
grpc_mdelem_from_metadata_strings(
- grpc_mdstr_from_string(headers->headers[i].key),
+ &exec_ctx, grpc_mdstr_from_string(headers->headers[i].key),
grpc_mdstr_from_string(headers->headers[i].value)));
}
s->state.state_callback_received[OP_RECV_INITIAL_METADATA] = true;
gpr_mu_unlock(&s->mu);
+ grpc_exec_ctx_finish(&exec_ctx);
execute_from_storage(s);
}
@@ -491,6 +493,7 @@ static void on_read_completed(cronet_bidirectional_stream *stream, char *data,
static void on_response_trailers_received(
cronet_bidirectional_stream *stream,
const cronet_bidirectional_stream_header_array *trailers) {
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
CRONET_LOG(GPR_DEBUG, "R: on_response_trailers_received(%p,%p)", stream,
trailers);
stream_obj *s = (stream_obj *)stream->annotation;
@@ -505,12 +508,13 @@ static void on_response_trailers_received(
grpc_chttp2_incoming_metadata_buffer_add(
&s->state.rs.trailing_metadata,
grpc_mdelem_from_metadata_strings(
- grpc_mdstr_from_string(trailers->headers[i].key),
+ &exec_ctx, grpc_mdstr_from_string(trailers->headers[i].key),
grpc_mdstr_from_string(trailers->headers[i].value)));
s->state.rs.trailing_metadata_valid = true;
}
s->state.state_callback_received[OP_RECV_TRAILING_METADATA] = true;
gpr_mu_unlock(&s->mu);
+ grpc_exec_ctx_finish(&exec_ctx);
execute_from_storage(s);
}