Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Change handshake manager to invoke callbacks via an exec_ctx.

Browse Source
reviewable/pr8782/r1
Mark D. Roth 8 years ago
parent aaefa958d7
commit 27588bbfb9
7 changed files with 144 additions and 164 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 58
      src/core/ext/client_channel/http_connect_handshaker.c
  2. 18
      src/core/ext/transport/chttp2/client/insecure/channel_create.c
  3. 14
      src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
  4. 16
      src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
  5. 20
      src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
  6. 117
      src/core/lib/channel/handshaker.c
  7. 65
      src/core/lib/channel/handshaker.h

58
src/core/ext/client_channel/http_connect_handshaker.c
Unescape View File

@ -54,14 +54,11 @@ typedef struct http_connect_handshaker {
char* server_name; char* server_name;
// State saved while performing the handshake. // State saved while performing the handshake.
grpc_endpoint* endpoint; grpc_handshaker_args* args;
grpc_channel_args* args; grpc_closure cb;
grpc_handshaker_done_cb cb;
void* user_data;
// Objects for processing the HTTP CONNECT request and response. // Objects for processing the HTTP CONNECT request and response.
grpc_slice_buffer write_buffer; grpc_slice_buffer write_buffer;
grpc_slice_buffer* read_buffer; // Ownership passes through this object.
grpc_closure request_done_closure; grpc_closure request_done_closure;
grpc_closure response_read_closure; grpc_closure response_read_closure;
grpc_http_parser http_parser; grpc_http_parser http_parser;
@ -87,7 +84,7 @@ static void http_connect_handshaker_unref(http_connect_handshaker* handshaker) {
static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
http_connect_handshaker* handshaker = arg; http_connect_handshaker* handshaker = arg;
if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled. if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled.
grpc_endpoint_shutdown(exec_ctx, handshaker->endpoint); grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint);
} }
http_connect_handshaker_unref(handshaker); http_connect_handshaker_unref(handshaker);
} }
@ -98,12 +95,11 @@ static void on_write_done(grpc_exec_ctx* exec_ctx, void* arg,
http_connect_handshaker* handshaker = arg; http_connect_handshaker* handshaker = arg;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
// If the write failed, invoke the callback immediately with the error. // If the write failed, invoke the callback immediately with the error.
handshaker->cb(exec_ctx, handshaker->endpoint, handshaker->args, grpc_exec_ctx_sched(exec_ctx, &handshaker->cb, GRPC_ERROR_REF(error), NULL);
handshaker->read_buffer, handshaker->user_data,
GRPC_ERROR_REF(error));
} else { } else {
// Otherwise, read the response. // Otherwise, read the response.
grpc_endpoint_read(exec_ctx, handshaker->endpoint, handshaker->read_buffer, grpc_endpoint_read(exec_ctx, handshaker->args->endpoint,
handshaker->args->read_buffer,
&handshaker->response_read_closure); &handshaker->response_read_closure);
} }
} }
@ -117,11 +113,11 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
goto done; goto done;
} }
// Add buffer to parser. // Add buffer to parser.
for (size_t i = 0; i < handshaker->read_buffer->count; ++i) { for (size_t i = 0; i < handshaker->args->read_buffer->count; ++i) {
if (GRPC_SLICE_LENGTH(handshaker->read_buffer->slices[i]) > 0) { if (GRPC_SLICE_LENGTH(handshaker->args->read_buffer->slices[i]) > 0) {
size_t body_start_offset = 0; size_t body_start_offset = 0;
error = grpc_http_parser_parse(&handshaker->http_parser, error = grpc_http_parser_parse(&handshaker->http_parser,
handshaker->read_buffer->slices[i], handshaker->args->read_buffer->slices[i],
&body_start_offset); &body_start_offset);
if (error != GRPC_ERROR_NONE) goto done; if (error != GRPC_ERROR_NONE) goto done;
if (handshaker->http_parser.state == GRPC_HTTP_BODY) { if (handshaker->http_parser.state == GRPC_HTTP_BODY) {
@ -132,16 +128,16 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_slice_buffer tmp_buffer; grpc_slice_buffer tmp_buffer;
grpc_slice_buffer_init(&tmp_buffer); grpc_slice_buffer_init(&tmp_buffer);
if (body_start_offset < if (body_start_offset <
GRPC_SLICE_LENGTH(handshaker->read_buffer->slices[i])) { GRPC_SLICE_LENGTH(handshaker->args->read_buffer->slices[i])) {
grpc_slice_buffer_add( grpc_slice_buffer_add(
&tmp_buffer, &tmp_buffer,
grpc_slice_split_tail(&handshaker->read_buffer->slices[i], grpc_slice_split_tail(&handshaker->args->read_buffer->slices[i],
body_start_offset)); body_start_offset));
} }
grpc_slice_buffer_addn(&tmp_buffer, grpc_slice_buffer_addn(&tmp_buffer,
&handshaker->read_buffer->slices[i + 1], &handshaker->args->read_buffer->slices[i + 1],
handshaker->read_buffer->count - i - 1); handshaker->args->read_buffer->count - i - 1);
grpc_slice_buffer_swap(handshaker->read_buffer, &tmp_buffer); grpc_slice_buffer_swap(handshaker->args->read_buffer, &tmp_buffer);
grpc_slice_buffer_destroy(&tmp_buffer); grpc_slice_buffer_destroy(&tmp_buffer);
break; break;
} }
@ -159,8 +155,9 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
// complete (e.g., handling chunked transfer encoding or looking // complete (e.g., handling chunked transfer encoding or looking
// at the Content-Length: header). // at the Content-Length: header).
if (handshaker->http_parser.state != GRPC_HTTP_BODY) { if (handshaker->http_parser.state != GRPC_HTTP_BODY) {
grpc_slice_buffer_reset_and_unref(handshaker->read_buffer); grpc_slice_buffer_reset_and_unref(handshaker->args->read_buffer);
grpc_endpoint_read(exec_ctx, handshaker->endpoint, handshaker->read_buffer, grpc_endpoint_read(exec_ctx, handshaker->args->endpoint,
handshaker->args->read_buffer,
&handshaker->response_read_closure); &handshaker->response_read_closure);
return; return;
} }
@ -175,8 +172,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
} }
done: done:
// Invoke handshake-done callback. // Invoke handshake-done callback.
handshaker->cb(exec_ctx, handshaker->endpoint, handshaker->args, grpc_exec_ctx_sched(exec_ctx, &handshaker->cb, error, NULL);
handshaker->read_buffer, handshaker->user_data, error);
} }
// //
@ -194,17 +190,12 @@ static void http_connect_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
static void http_connect_handshaker_do_handshake( static void http_connect_handshaker_do_handshake(
grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker_in, grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker_in,
grpc_endpoint* endpoint, grpc_channel_args* args, gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
grpc_slice_buffer* read_buffer, gpr_timespec deadline, grpc_iomgr_cb_func cb, grpc_handshaker_args* args) {
grpc_tcp_server_acceptor* acceptor, grpc_handshaker_done_cb cb,
void* user_data) {
http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in; http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in;
// Save state in the handshaker object. // Save state in the handshaker object.
handshaker->endpoint = endpoint;
handshaker->args = args; handshaker->args = args;
handshaker->cb = cb; grpc_closure_init(&handshaker->cb, cb, args);
handshaker->user_data = user_data;
handshaker->read_buffer = read_buffer;
// Send HTTP CONNECT request. // Send HTTP CONNECT request.
gpr_log(GPR_INFO, "Connecting to server %s via HTTP proxy %s", gpr_log(GPR_INFO, "Connecting to server %s via HTTP proxy %s",
handshaker->server_name, handshaker->proxy_server); handshaker->server_name, handshaker->proxy_server);
@ -216,7 +207,7 @@ static void http_connect_handshaker_do_handshake(
request.handshaker = &grpc_httpcli_plaintext; request.handshaker = &grpc_httpcli_plaintext;
grpc_slice request_slice = grpc_httpcli_format_connect_request(&request); grpc_slice request_slice = grpc_httpcli_format_connect_request(&request);
grpc_slice_buffer_add(&handshaker->write_buffer, request_slice); grpc_slice_buffer_add(&handshaker->write_buffer, request_slice);
grpc_endpoint_write(exec_ctx, endpoint, &handshaker->write_buffer, grpc_endpoint_write(exec_ctx, args->endpoint, &handshaker->write_buffer,
&handshaker->request_done_closure); &handshaker->request_done_closure);
// Set timeout timer. The timer gets a reference to the handshaker. // Set timeout timer. The timer gets a reference to the handshaker.
gpr_ref(&handshaker->refcount); gpr_ref(&handshaker->refcount);
@ -225,7 +216,7 @@ static void http_connect_handshaker_do_handshake(
on_timeout, handshaker, gpr_now(GPR_CLOCK_MONOTONIC)); on_timeout, handshaker, gpr_now(GPR_CLOCK_MONOTONIC));
} }
static const struct grpc_handshaker_vtable http_connect_handshaker_vtable = { static const grpc_handshaker_vtable http_connect_handshaker_vtable = {
http_connect_handshaker_destroy, http_connect_handshaker_shutdown, http_connect_handshaker_destroy, http_connect_handshaker_shutdown,
http_connect_handshaker_do_handshake}; http_connect_handshaker_do_handshake};
@ -233,8 +224,7 @@ grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
const char* server_name) { const char* server_name) {
GPR_ASSERT(proxy_server != NULL); GPR_ASSERT(proxy_server != NULL);
GPR_ASSERT(server_name != NULL); GPR_ASSERT(server_name != NULL);
http_connect_handshaker* handshaker = http_connect_handshaker* handshaker = gpr_malloc(sizeof(*handshaker));
gpr_malloc(sizeof(http_connect_handshaker));
memset(handshaker, 0, sizeof(*handshaker)); memset(handshaker, 0, sizeof(*handshaker));
grpc_handshaker_init(&http_connect_handshaker_vtable, &handshaker->base); grpc_handshaker_init(&http_connect_handshaker_vtable, &handshaker->base);
handshaker->proxy_server = gpr_strdup(proxy_server); handshaker->proxy_server = gpr_strdup(proxy_server);

18
src/core/ext/transport/chttp2/client/insecure/channel_create.c
Unescape View File

@ -92,22 +92,22 @@ static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
connector_unref(exec_ctx, arg); connector_unref(exec_ctx, arg);
} }
static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint, static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
grpc_channel_args *args,
grpc_slice_buffer *read_buffer, void *user_data,
grpc_error *error) { grpc_error *error) {
connector *c = user_data; grpc_handshaker_args *args = arg;
connector *c = args->user_data;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
grpc_channel_args_destroy(args); grpc_channel_args_destroy(args->args);
gpr_free(read_buffer); gpr_free(args->read_buffer);
} else { } else {
c->result->transport = c->result->transport =
grpc_create_chttp2_transport(exec_ctx, args, endpoint, 1); grpc_create_chttp2_transport(exec_ctx, args->args, args->endpoint, 1);
GPR_ASSERT(c->result->transport); GPR_ASSERT(c->result->transport);
grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport,
read_buffer); args->read_buffer);
c->result->channel_args = args; c->result->channel_args = args->args;
} }
gpr_free(args);
grpc_closure *notify = c->notify; grpc_closure *notify = c->notify;
c->notify = NULL; c->notify = NULL;
grpc_exec_ctx_sched(exec_ctx, notify, error, NULL); grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);

14
src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
Unescape View File

@ -129,14 +129,13 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
grpc_exec_ctx_sched(exec_ctx, notify, error, NULL); grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
} }
static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint, static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
grpc_channel_args *args,
grpc_slice_buffer *read_buffer, void *user_data,
grpc_error *error) { grpc_error *error) {
connector *c = user_data; grpc_handshaker_args *args = arg;
c->tmp_args = args; connector *c = args->user_data;
c->tmp_args = args->args;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
gpr_free(read_buffer); gpr_free(args->read_buffer);
grpc_closure *notify = c->notify; grpc_closure *notify = c->notify;
c->notify = NULL; c->notify = NULL;
grpc_exec_ctx_sched(exec_ctx, notify, error, NULL); grpc_exec_ctx_sched(exec_ctx, notify, error, NULL);
@ -145,9 +144,10 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
// handshake API, and then move the code from on_secure_handshake_done() // handshake API, and then move the code from on_secure_handshake_done()
// into this function. // into this function.
grpc_channel_security_connector_do_handshake( grpc_channel_security_connector_do_handshake(
exec_ctx, c->security_connector, endpoint, read_buffer, exec_ctx, c->security_connector, args->endpoint, args->read_buffer,
c->args.deadline, on_secure_handshake_done, c); c->args.deadline, on_secure_handshake_done, c);
} }
gpr_free(args);
} }
static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg, static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,

16
src/core/ext/transport/chttp2/server/insecure/server_chttp2.c
Unescape View File

@ -54,18 +54,17 @@ typedef struct server_connect_state {
grpc_handshake_manager *handshake_mgr; grpc_handshake_manager *handshake_mgr;
} server_connect_state; } server_connect_state;
static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint, static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
grpc_channel_args *args,
grpc_slice_buffer *read_buffer, void *user_data,
grpc_error *error) { grpc_error *error) {
server_connect_state *state = user_data; grpc_handshaker_args *args = arg;
server_connect_state *state = args->user_data;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
const char *error_str = grpc_error_string(error); const char *error_str = grpc_error_string(error);
gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str); gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str);
grpc_error_free_string(error_str); grpc_error_free_string(error_str);
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(error);
grpc_handshake_manager_shutdown(exec_ctx, state->handshake_mgr); grpc_handshake_manager_shutdown(exec_ctx, state->handshake_mgr);
gpr_free(read_buffer); gpr_free(args->read_buffer);
} else { } else {
// Beware that the call to grpc_create_chttp2_transport() has to happen // Beware that the call to grpc_create_chttp2_transport() has to happen
// before grpc_tcp_server_destroy(). This is fine here, but similar code // before grpc_tcp_server_destroy(). This is fine here, but similar code
@ -73,14 +72,15 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
// grpc_tcp_server_start() (as in server_secure_chttp2.c) needs to add // grpc_tcp_server_start() (as in server_secure_chttp2.c) needs to add
// synchronization to avoid this case. // synchronization to avoid this case.
grpc_transport *transport = grpc_transport *transport =
grpc_create_chttp2_transport(exec_ctx, args, endpoint, 0); grpc_create_chttp2_transport(exec_ctx, args->args, args->endpoint, 0);
grpc_server_setup_transport(exec_ctx, state->server, transport, grpc_server_setup_transport(exec_ctx, state->server, transport,
state->accepting_pollset, state->accepting_pollset,
grpc_server_get_channel_args(state->server)); grpc_server_get_channel_args(state->server));
grpc_chttp2_transport_start_reading(exec_ctx, transport, read_buffer); grpc_chttp2_transport_start_reading(exec_ctx, transport, args->read_buffer);
} }
// Clean up. // Clean up.
grpc_channel_args_destroy(args); grpc_channel_args_destroy(args->args);
gpr_free(args);
grpc_handshake_manager_destroy(exec_ctx, state->handshake_mgr); grpc_handshake_manager_destroy(exec_ctx, state->handshake_mgr);
gpr_free(state); gpr_free(state);
} }

20
src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c
Unescape View File

@ -115,18 +115,19 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *statep,
gpr_free(connection_state); gpr_free(connection_state);
} }
static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint, static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
grpc_channel_args *args,
grpc_slice_buffer *read_buffer, void *user_data,
grpc_error *error) { grpc_error *error) {
server_secure_connect *connection_state = user_data; grpc_handshaker_args* args = arg;
server_secure_connect *connection_state = args->user_data;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
const char *error_str = grpc_error_string(error); const char *error_str = grpc_error_string(error);
gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str); gpr_log(GPR_ERROR, "Handshaking failed: %s", error_str);
grpc_error_free_string(error_str); grpc_error_free_string(error_str);
GRPC_ERROR_UNREF(error); // FIXME: remove?
grpc_channel_args_destroy(args); // GRPC_ERROR_UNREF(error);
gpr_free(read_buffer); grpc_channel_args_destroy(args->args);
gpr_free(args->read_buffer);
gpr_free(args);
grpc_handshake_manager_shutdown(exec_ctx, connection_state->handshake_mgr); grpc_handshake_manager_shutdown(exec_ctx, connection_state->handshake_mgr);
grpc_handshake_manager_destroy(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); grpc_tcp_server_unref(exec_ctx, connection_state->server_state->tcp);
@ -138,11 +139,12 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, grpc_endpoint *endpoint,
// TODO(roth, jboeuf): Convert security connector handshaking to use new // TODO(roth, jboeuf): Convert security connector handshaking to use new
// handshake API, and then move the code from on_secure_handshake_done() // handshake API, and then move the code from on_secure_handshake_done()
// into this function. // into this function.
connection_state->args = args; connection_state->args = args->args;
grpc_server_security_connector_do_handshake( grpc_server_security_connector_do_handshake(
exec_ctx, connection_state->server_state->sc, connection_state->acceptor, exec_ctx, connection_state->server_state->sc, connection_state->acceptor,
endpoint, read_buffer, connection_state->deadline, args->endpoint, args->read_buffer, connection_state->deadline,
on_secure_handshake_done, connection_state); on_secure_handshake_done, connection_state);
gpr_free(args);
} }
static void on_accept(grpc_exec_ctx *exec_ctx, void *statep, grpc_endpoint *tcp, static void on_accept(grpc_exec_ctx *exec_ctx, void *statep, grpc_endpoint *tcp,

117
src/core/lib/channel/handshaker.c
Unescape View File

@ -43,32 +43,29 @@
// grpc_handshaker // grpc_handshaker
// //
void grpc_handshaker_init(const struct grpc_handshaker_vtable* vtable, void grpc_handshaker_init(const grpc_handshaker_vtable* vtable,
grpc_handshaker* handshaker) { grpc_handshaker* handshaker) {
handshaker->vtable = vtable; handshaker->vtable = vtable;
} }
void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx, static void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker) { grpc_handshaker* handshaker) {
handshaker->vtable->destroy(exec_ctx, handshaker); handshaker->vtable->destroy(exec_ctx, handshaker);
} }
void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx, static void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker) { grpc_handshaker* handshaker) {
handshaker->vtable->shutdown(exec_ctx, handshaker); handshaker->vtable->shutdown(exec_ctx, handshaker);
} }
void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx, static void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker, grpc_handshaker* handshaker,
grpc_endpoint* endpoint, gpr_timespec deadline,
grpc_channel_args* args, grpc_tcp_server_acceptor* acceptor,
grpc_slice_buffer* read_buffer, grpc_iomgr_cb_func on_handshake_done,
gpr_timespec deadline, grpc_handshaker_args* args) {
grpc_tcp_server_acceptor* acceptor, handshaker->vtable->do_handshake(exec_ctx, handshaker, deadline, acceptor,
grpc_handshaker_done_cb cb, void* user_data) { on_handshake_done, args);
handshaker->vtable->do_handshake(exec_ctx, handshaker, endpoint, args,
read_buffer, deadline, acceptor, cb,
user_data);
} }
// //
@ -84,8 +81,10 @@ struct grpc_handshaker_state {
// The acceptor to call the handshakers with. // The acceptor to call the handshakers with.
grpc_tcp_server_acceptor* acceptor; grpc_tcp_server_acceptor* acceptor;
// The final callback and user_data to invoke after the last handshaker. // The final callback and user_data to invoke after the last handshaker.
grpc_handshaker_done_cb final_cb; grpc_iomgr_cb_func on_handshake_done;
void* final_user_data; void* user_data;
// Next closure to call.
grpc_closure next_cb;
}; };
struct grpc_handshake_manager { struct grpc_handshake_manager {
@ -143,61 +142,61 @@ void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx,
// A function used as the handshaker-done callback when chaining // A function used as the handshaker-done callback when chaining
// handshakers together. // handshakers together.
static void call_next_handshaker(grpc_exec_ctx* exec_ctx, static void call_next_handshaker(grpc_exec_ctx* exec_ctx, void* arg,
grpc_endpoint* endpoint, grpc_error* error) {
grpc_channel_args* args, grpc_handshaker_args* args = arg;
grpc_slice_buffer* read_buffer, grpc_handshake_manager* mgr = args->user_data;
void* user_data, grpc_error* error) {
grpc_handshake_manager* mgr = user_data;
GPR_ASSERT(mgr->state != NULL); GPR_ASSERT(mgr->state != NULL);
GPR_ASSERT(mgr->state->index < mgr->count); GPR_ASSERT(mgr->state->index <= mgr->count);
// If we got an error, skip all remaining handshakers and invoke the // If we got an error, skip all remaining handshakers and invoke the
// caller-supplied callback immediately. // caller-supplied callback immediately.
if (error != GRPC_ERROR_NONE) { // Otherwise, if this is the last handshaker, then call the final
mgr->state->final_cb(exec_ctx, endpoint, args, read_buffer, // callback instead of chaining back to this function again.
mgr->state->final_user_data, error); if (error != GRPC_ERROR_NONE || mgr->state->index == mgr->count) {
args->user_data = mgr->state->user_data;
grpc_closure_init(&mgr->state->next_cb, mgr->state->on_handshake_done,
args);
grpc_exec_ctx_sched(exec_ctx, &mgr->state->next_cb, GRPC_ERROR_REF(error),
NULL);
return; return;
} }
grpc_handshaker_done_cb cb = call_next_handshaker; // Call the next handshaker.
// If this is the last handshaker, use the caller-supplied callback
// and user_data instead of chaining back to this function again.
if (mgr->state->index == mgr->count - 1) {
cb = mgr->state->final_cb;
user_data = mgr->state->final_user_data;
}
// Invoke handshaker.
grpc_handshaker_do_handshake( grpc_handshaker_do_handshake(
exec_ctx, mgr->handshakers[mgr->state->index], endpoint, args, exec_ctx, mgr->handshakers[mgr->state->index], mgr->state->deadline,
read_buffer, mgr->state->deadline, mgr->state->acceptor, cb, user_data); mgr->state->acceptor, call_next_handshaker, args);
++mgr->state->index;
// If this is the last handshaker, clean up state. // If this is the last handshaker, clean up state.
if (mgr->state->index == mgr->count) { if (mgr->state->index == mgr->count) {
gpr_free(mgr->state); gpr_free(mgr->state);
mgr->state = NULL; mgr->state = NULL;
} else {
++mgr->state->index;
} }
} }
void grpc_handshake_manager_do_handshake( void grpc_handshake_manager_do_handshake(
grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr, grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr,
grpc_endpoint* endpoint, const grpc_channel_args* args, grpc_endpoint* endpoint, const grpc_channel_args* channel_args,
gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor, gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
grpc_handshaker_done_cb cb, void* user_data) { grpc_iomgr_cb_func on_handshake_done, void* user_data) {
grpc_channel_args* args_copy = grpc_channel_args_copy(args); // Construct state.
grpc_slice_buffer* read_buffer = gpr_malloc(sizeof(*read_buffer)); GPR_ASSERT(mgr->state == NULL);
grpc_slice_buffer_init(read_buffer); mgr->state = gpr_malloc(sizeof(struct grpc_handshaker_state));
if (mgr->count == 0) { memset(mgr->state, 0, sizeof(*mgr->state));
// No handshakers registered, so we just immediately call the done mgr->state->deadline = deadline;
// callback with the passed-in endpoint. mgr->state->acceptor = acceptor;
cb(exec_ctx, endpoint, args_copy, read_buffer, user_data, GRPC_ERROR_NONE); mgr->state->on_handshake_done = on_handshake_done;
} else { mgr->state->user_data = user_data;
GPR_ASSERT(mgr->state == NULL); // Construct handshaker args. These will be passed through all
mgr->state = gpr_malloc(sizeof(struct grpc_handshaker_state)); // handshakers and eventually be freed by the final callback.
memset(mgr->state, 0, sizeof(*mgr->state)); grpc_handshaker_args* args = gpr_malloc(sizeof(*args));
mgr->state->deadline = deadline; args->endpoint = endpoint;
mgr->state->acceptor = acceptor; args->args = grpc_channel_args_copy(channel_args);
mgr->state->final_cb = cb; args->read_buffer = gpr_malloc(sizeof(*args->read_buffer));
mgr->state->final_user_data = user_data; grpc_slice_buffer_init(args->read_buffer);
call_next_handshaker(exec_ctx, endpoint, args_copy, read_buffer, mgr, // While chaining between handshakers, we use args->user_data to
GRPC_ERROR_NONE); // store a pointer to the handshake manager. This will be
} // changed to point to the caller-supplied user_data before calling
// the final callback.
args->user_data = mgr;
call_next_handshaker(exec_ctx, args, GRPC_ERROR_NONE);
} }

65
src/core/lib/channel/handshaker.h
Unescape View File

@ -48,21 +48,26 @@
/// ///
/// In general, handshakers should be used via a handshake manager. /// In general, handshakers should be used via a handshake manager.
///
/// grpc_handshaker_state
///
/// Arguments passed through handshakers and to the on_handshake_done callback.
/// All data members are owned by the struct.
typedef struct {
grpc_endpoint* endpoint;
grpc_channel_args* args;
void* user_data;
grpc_slice_buffer* read_buffer;
} grpc_handshaker_args;
/// ///
/// grpc_handshaker /// grpc_handshaker
/// ///
typedef struct grpc_handshaker grpc_handshaker; typedef struct grpc_handshaker grpc_handshaker;
/// Callback type invoked when a handshaker is done. typedef struct {
/// Takes ownership of \a args and \a read_buffer.
typedef void (*grpc_handshaker_done_cb)(grpc_exec_ctx* exec_ctx,
grpc_endpoint* endpoint,
grpc_channel_args* args,
grpc_slice_buffer* read_buffer,
void* user_data, grpc_error* error);
struct grpc_handshaker_vtable {
/// Destroys the handshaker. /// Destroys the handshaker.
void (*destroy)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker); void (*destroy)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker);
@ -70,44 +75,28 @@ struct grpc_handshaker_vtable {
/// aborted in the middle). /// aborted in the middle).
void (*shutdown)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker); void (*shutdown)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker);
/// Performs handshaking. When finished, calls \a cb with \a user_data. /// Performs handshaking.
/// Takes ownership of \a args. /// Takes ownership of \a args.
/// Takes ownership of \a read_buffer, which contains leftover bytes read /// When finished, calls \a on_handshake_done with \a args, which the
/// from the endpoint by the previous handshaker. /// callback takes ownership of.
/// \a acceptor will be NULL for client-side handshakers. /// \a acceptor will be NULL for client-side handshakers.
void (*do_handshake)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker, void (*do_handshake)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker,
grpc_endpoint* endpoint, grpc_channel_args* args, gpr_timespec deadline,
grpc_slice_buffer* read_buffer, gpr_timespec deadline,
grpc_tcp_server_acceptor* acceptor, grpc_tcp_server_acceptor* acceptor,
grpc_handshaker_done_cb cb, void* user_data); grpc_iomgr_cb_func on_handshake_done,
}; grpc_handshaker_args* args);
} grpc_handshaker_vtable;
/// Base struct. To subclass, make this the first member of the /// Base struct. To subclass, make this the first member of the
/// implementation struct. /// implementation struct.
struct grpc_handshaker { struct grpc_handshaker {
const struct grpc_handshaker_vtable* vtable; const grpc_handshaker_vtable* vtable;
}; };
/// Called by concrete implementations to initialize the base struct. /// Called by concrete implementations to initialize the base struct.
void grpc_handshaker_init(const struct grpc_handshaker_vtable* vtable, void grpc_handshaker_init(const grpc_handshaker_vtable* vtable,
grpc_handshaker* handshaker); grpc_handshaker* handshaker);
/// Convenient wrappers for invoking methods via the vtable.
/// These probably do not need to be called from anywhere but
/// grpc_handshake_manager.
void grpc_handshaker_destroy(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker);
void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker);
void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker,
grpc_endpoint* endpoint,
grpc_channel_args* args,
grpc_slice_buffer* read_buffer,
gpr_timespec deadline,
grpc_tcp_server_acceptor* acceptor,
grpc_handshaker_done_cb cb, void* user_data);
/// ///
/// grpc_handshake_manager /// grpc_handshake_manager
/// ///
@ -137,12 +126,12 @@ void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx,
/// Does NOT take ownership of \a args. Instead, makes a copy before /// Does NOT take ownership of \a args. Instead, makes a copy before
/// invoking the first handshaker. /// invoking the first handshaker.
/// \a acceptor will be NULL for client-side handshakers. /// \a acceptor will be NULL for client-side handshakers.
/// Invokes \a cb with \a user_data after either a handshaker fails or /// When done, invokes \a on_handshake_done with an argument of a
/// all handshakers have completed successfully. /// grpc_handshaker_args object, which the callback takes ownership of.
void grpc_handshake_manager_do_handshake( void grpc_handshake_manager_do_handshake(
grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr, grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr,
grpc_endpoint* endpoint, const grpc_channel_args* args, grpc_endpoint* endpoint, const grpc_channel_args* channel_args,
gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor, gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
grpc_handshaker_done_cb cb, void* user_data); grpc_iomgr_cb_func on_handshake_done, void* user_data);
#endif /* GRPC_CORE_LIB_CHANNEL_HANDSHAKER_H */ #endif /* GRPC_CORE_LIB_CHANNEL_HANDSHAKER_H */

Write Preview
Loading…
Cancel
Save