Move deadline handling to handshake manager.

reviewable/pr8782/r1
Mark D. Roth 8 years ago
parent 4b5cdb751e
commit b16c1e32fd
  1. 59
      src/core/ext/client_channel/http_connect_handshaker.c
  2. 145
      src/core/lib/channel/handshaker.c
  3. 1
      src/core/lib/channel/handshaker.h

@ -43,7 +43,6 @@
#include "src/core/ext/client_channel/uri_parser.h"
#include "src/core/lib/http/format_request.h"
#include "src/core/lib/http/parser.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/support/env.h"
typedef struct http_connect_handshaker {
@ -53,6 +52,9 @@ typedef struct http_connect_handshaker {
char* proxy_server;
char* server_name;
gpr_refcount refcount;
gpr_mu mu;
// State saved while performing the handshake.
grpc_handshaker_args* args;
grpc_closure* on_handshake_done;
@ -63,14 +65,12 @@ typedef struct http_connect_handshaker {
grpc_closure response_read_closure;
grpc_http_parser http_parser;
grpc_http_response http_response;
grpc_timer timeout_timer;
gpr_refcount refcount;
} http_connect_handshaker;
// Unref and clean up handshaker.
static void http_connect_handshaker_unref(http_connect_handshaker* handshaker) {
if (gpr_unref(&handshaker->refcount)) {
gpr_mu_destroy(&handshaker->mu);
gpr_free(handshaker->proxy_server);
gpr_free(handshaker->server_name);
grpc_slice_buffer_destroy(&handshaker->write_buffer);
@ -80,28 +80,27 @@ static void http_connect_handshaker_unref(http_connect_handshaker* handshaker) {
}
}
// Callback invoked when deadline is exceeded.
static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
http_connect_handshaker* handshaker = arg;
if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled.
grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint);
}
http_connect_handshaker_unref(handshaker);
}
// Callback invoked when finished writing HTTP CONNECT request.
static void on_write_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
http_connect_handshaker* handshaker = arg;
if (error != GRPC_ERROR_NONE) {
// If the write failed, invoke the callback immediately with the error.
gpr_mu_lock(&handshaker->mu);
grpc_exec_ctx_sched(exec_ctx, handshaker->on_handshake_done,
GRPC_ERROR_REF(error), NULL);
handshaker->args = NULL;
gpr_mu_unlock(&handshaker->mu);
http_connect_handshaker_unref(handshaker);
} else {
// Otherwise, read the response.
// The read callback inherits our ref to the handshaker.
gpr_mu_lock(&handshaker->mu);
GPR_ASSERT(handshaker->args != NULL);
grpc_endpoint_read(exec_ctx, handshaker->args->endpoint,
handshaker->args->read_buffer,
&handshaker->response_read_closure);
gpr_mu_unlock(&handshaker->mu);
}
}
@ -109,6 +108,8 @@ static void on_write_done(grpc_exec_ctx* exec_ctx, void* arg,
static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
http_connect_handshaker* handshaker = arg;
gpr_mu_lock(&handshaker->mu);
GPR_ASSERT(handshaker->args != NULL);
if (error != GRPC_ERROR_NONE) {
GRPC_ERROR_REF(error); // Take ref to pass to the handshake-done callback.
goto done;
@ -122,8 +123,6 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
&body_start_offset);
if (error != GRPC_ERROR_NONE) goto done;
if (handshaker->http_parser.state == GRPC_HTTP_BODY) {
// We've gotten back a successul response, so stop the timeout timer.
grpc_timer_cancel(exec_ctx, &handshaker->timeout_timer);
// Remove the data we've already read from the read buffer,
// leaving only the leftover bytes (if any).
grpc_slice_buffer tmp_buffer;
@ -160,6 +159,7 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_endpoint_read(exec_ctx, handshaker->args->endpoint,
handshaker->args->read_buffer,
&handshaker->response_read_closure);
gpr_mu_unlock(&handshaker->mu);
return;
}
// Make sure we got a 2xx response.
@ -173,7 +173,10 @@ static void on_read_done(grpc_exec_ctx* exec_ctx, void* arg,
}
done:
// Invoke handshake-done callback.
handshaker->args = NULL;
grpc_exec_ctx_sched(exec_ctx, handshaker->on_handshake_done, error, NULL);
gpr_mu_unlock(&handshaker->mu);
http_connect_handshaker_unref(handshaker);
}
//
@ -187,13 +190,22 @@ static void http_connect_handshaker_destroy(grpc_exec_ctx* exec_ctx,
}
static void http_connect_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker) {}
grpc_handshaker* handshaker_in) {
http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in;
gpr_mu_lock(&handshaker->mu);
if (handshaker->args != NULL) {
grpc_endpoint_shutdown(exec_ctx, handshaker->args->endpoint);
handshaker->args = NULL;
}
gpr_mu_unlock(&handshaker->mu);
}
static void http_connect_handshaker_do_handshake(
grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker_in,
gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
grpc_closure* on_handshake_done, grpc_handshaker_args* args) {
grpc_tcp_server_acceptor* acceptor, grpc_closure* on_handshake_done,
grpc_handshaker_args* args) {
http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in;
gpr_mu_lock(&handshaker->mu);
// Save state in the handshaker object.
handshaker->args = args;
handshaker->on_handshake_done = on_handshake_done;
@ -208,13 +220,11 @@ static void http_connect_handshaker_do_handshake(
request.handshaker = &grpc_httpcli_plaintext;
grpc_slice request_slice = grpc_httpcli_format_connect_request(&request);
grpc_slice_buffer_add(&handshaker->write_buffer, request_slice);
// Take a new ref to be held by the write callback.
gpr_ref(&handshaker->refcount);
grpc_endpoint_write(exec_ctx, args->endpoint, &handshaker->write_buffer,
&handshaker->request_done_closure);
// Set timeout timer. The timer gets a reference to the handshaker.
gpr_ref(&handshaker->refcount);
grpc_timer_init(exec_ctx, &handshaker->timeout_timer,
gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
on_timeout, handshaker, gpr_now(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&handshaker->mu);
}
static const grpc_handshaker_vtable http_connect_handshaker_vtable = {
@ -228,6 +238,8 @@ grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
http_connect_handshaker* handshaker = gpr_malloc(sizeof(*handshaker));
memset(handshaker, 0, sizeof(*handshaker));
grpc_handshaker_init(&http_connect_handshaker_vtable, &handshaker->base);
gpr_mu_init(&handshaker->mu);
gpr_ref_init(&handshaker->refcount, 1);
handshaker->proxy_server = gpr_strdup(proxy_server);
handshaker->server_name = gpr_strdup(server_name);
grpc_slice_buffer_init(&handshaker->write_buffer);
@ -237,7 +249,6 @@ grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
handshaker);
grpc_http_parser_init(&handshaker->http_parser, GRPC_HTTP_RESPONSE,
&handshaker->http_response);
gpr_ref_init(&handshaker->refcount, 1);
return &handshaker->base;
}

@ -38,6 +38,7 @@
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/handshaker.h"
#include "src/core/lib/iomgr/timer.h"
//
// grpc_handshaker
@ -60,11 +61,10 @@ static void grpc_handshaker_shutdown(grpc_exec_ctx* exec_ctx,
static void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
grpc_handshaker* handshaker,
gpr_timespec deadline,
grpc_tcp_server_acceptor* acceptor,
grpc_closure* on_handshake_done,
grpc_handshaker_args* args) {
handshaker->vtable->do_handshake(exec_ctx, handshaker, deadline, acceptor,
handshaker->vtable->do_handshake(exec_ctx, handshaker, acceptor,
on_handshake_done, args);
}
@ -72,31 +72,29 @@ static void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx,
// grpc_handshake_manager
//
// State used while chaining handshakers.
struct grpc_handshaker_state {
// The index of the handshaker to invoke next and the closure to invoke it.
struct grpc_handshake_manager {
gpr_mu mu;
gpr_refcount refs;
// An array of handshakers added via grpc_handshake_manager_add().
size_t count;
grpc_handshaker** handshakers;
// The index of the handshaker to invoke next and closure to invoke it.
size_t index;
grpc_closure call_next_handshaker;
// The deadline for all handshakers.
gpr_timespec deadline;
// The acceptor to call the handshakers with.
grpc_tcp_server_acceptor* acceptor;
// Deadline timer across all handshakers.
grpc_timer deadline_timer;
// The final callback and user_data to invoke after the last handshaker.
grpc_closure on_handshake_done;
void* user_data;
};
struct grpc_handshake_manager {
// An array of handshakers added via grpc_handshake_manager_add().
size_t count;
grpc_handshaker** handshakers;
// State used while chaining handshakers.
struct grpc_handshaker_state* state;
};
grpc_handshake_manager* grpc_handshake_manager_create() {
grpc_handshake_manager* mgr = gpr_malloc(sizeof(grpc_handshake_manager));
memset(mgr, 0, sizeof(*mgr));
gpr_mu_init(&mgr->mu);
gpr_ref_init(&mgr->refs, 1);
return mgr;
}
@ -104,6 +102,7 @@ static bool is_power_of_2(size_t n) { return (n & (n - 1)) == 0; }
void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
grpc_handshaker* handshaker) {
gpr_mu_lock(&mgr->mu);
// To avoid allocating memory for each handshaker we add, we double
// the number of elements every time we need more.
size_t realloc_count = 0;
@ -117,57 +116,86 @@ void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
gpr_realloc(mgr->handshakers, realloc_count * sizeof(grpc_handshaker*));
}
mgr->handshakers[mgr->count++] = handshaker;
gpr_mu_unlock(&mgr->mu);
}
static void grpc_handshake_manager_unref(grpc_exec_ctx* exec_ctx,
grpc_handshake_manager* mgr) {
if (gpr_unref(&mgr->refs)) {
for (size_t i = 0; i < mgr->count; ++i) {
grpc_handshaker_destroy(exec_ctx, mgr->handshakers[i]);
}
gpr_free(mgr->handshakers);
gpr_mu_destroy(&mgr->mu);
gpr_free(mgr);
}
}
void grpc_handshake_manager_destroy(grpc_exec_ctx* exec_ctx,
grpc_handshake_manager* mgr) {
for (size_t i = 0; i < mgr->count; ++i) {
grpc_handshaker_destroy(exec_ctx, mgr->handshakers[i]);
}
gpr_free(mgr->handshakers);
gpr_free(mgr);
grpc_handshake_manager_unref(exec_ctx, mgr);
}
void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx,
grpc_handshake_manager* mgr) {
gpr_mu_lock(&mgr->mu);
for (size_t i = 0; i < mgr->count; ++i) {
grpc_handshaker_shutdown(exec_ctx, mgr->handshakers[i]);
}
if (mgr->state != NULL) {
gpr_free(mgr->state);
mgr->state = NULL;
}
gpr_mu_unlock(&mgr->mu);
}
// A function used as the handshaker-done callback when chaining
// handshakers together.
static void call_next_handshaker(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
grpc_handshaker_args* args = arg;
grpc_handshake_manager* mgr = args->user_data;
GPR_ASSERT(mgr->state != NULL);
GPR_ASSERT(mgr->state->index <= mgr->count);
grpc_error* error);
// Helper function to call either the next handshaker or the
// on_handshake_done callback.
static void call_next_handshaker_locked(
grpc_exec_ctx* exec_ctx, grpc_handshake_manager* mgr,
grpc_handshaker_args* args, grpc_error* error) {
GPR_ASSERT(mgr->index <= mgr->count);
// If we got an error, skip all remaining handshakers and invoke the
// caller-supplied callback immediately.
// Otherwise, if this is the last handshaker, then call the final
// Otherwise, if this is the last handshaker, then call the on_handshake_done
// callback instead of chaining back to this function again.
if (error != GRPC_ERROR_NONE || mgr->state->index == mgr->count) {
args->user_data = mgr->state->user_data;
grpc_exec_ctx_sched(exec_ctx, &mgr->state->on_handshake_done,
if (error != GRPC_ERROR_NONE || mgr->index == mgr->count) {
// Cancel deadline timer, since we're invoking the on_handshake_done
// callback now.
grpc_timer_cancel(exec_ctx, &mgr->deadline_timer);
args->user_data = mgr->user_data;
grpc_exec_ctx_sched(exec_ctx, &mgr->on_handshake_done,
GRPC_ERROR_REF(error), NULL);
// Since we're invoking the final callback, we won't be coming back
// to this function, so we can release our reference to the
// handshake manager.
grpc_handshake_manager_unref(exec_ctx, mgr);
return;
}
// Call the next handshaker.
grpc_handshaker_do_handshake(
exec_ctx, mgr->handshakers[mgr->state->index], mgr->state->deadline,
mgr->state->acceptor, &mgr->state->call_next_handshaker, args);
// If this is the last handshaker, clean up state.
if (mgr->state->index == mgr->count) {
gpr_free(mgr->state);
mgr->state = NULL;
} else {
++mgr->state->index;
exec_ctx, mgr->handshakers[mgr->index], mgr->acceptor,
&mgr->call_next_handshaker, args);
++mgr->index;
}
// A function used as the handshaker-done callback when chaining
// handshakers together.
static void call_next_handshaker(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
grpc_handshaker_args* args = arg;
grpc_handshake_manager* mgr = args->user_data;
gpr_mu_lock(&mgr->mu);
call_next_handshaker_locked(exec_ctx, mgr, args, error);
gpr_mu_unlock(&mgr->mu);
}
// Callback invoked when deadline is exceeded.
static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
grpc_handshake_manager* mgr = arg;
if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled.
grpc_handshake_manager_shutdown(exec_ctx, mgr);
}
grpc_handshake_manager_unref(exec_ctx, mgr);
}
void grpc_handshake_manager_do_handshake(
@ -176,26 +204,31 @@ void grpc_handshake_manager_do_handshake(
gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor,
grpc_iomgr_cb_func on_handshake_done, void* user_data) {
// Construct handshaker args. These will be passed through all
// handshakers and eventually be freed by the final callback.
// handshakers and eventually be freed by the on_handshake_done callback.
grpc_handshaker_args* args = gpr_malloc(sizeof(*args));
args->endpoint = endpoint;
args->args = grpc_channel_args_copy(channel_args);
args->read_buffer = gpr_malloc(sizeof(*args->read_buffer));
grpc_slice_buffer_init(args->read_buffer);
// Construct state.
GPR_ASSERT(mgr->state == NULL);
mgr->state = gpr_malloc(sizeof(struct grpc_handshaker_state));
memset(mgr->state, 0, sizeof(*mgr->state));
grpc_closure_init(&mgr->state->call_next_handshaker, call_next_handshaker,
args);
mgr->state->deadline = deadline;
mgr->state->acceptor = acceptor;
grpc_closure_init(&mgr->state->on_handshake_done, on_handshake_done, args);
// Initialize state needed for calling handshakers.
gpr_mu_lock(&mgr->mu);
GPR_ASSERT(mgr->index == 0);
mgr->acceptor = acceptor;
grpc_closure_init(&mgr->call_next_handshaker, call_next_handshaker, args);
grpc_closure_init(&mgr->on_handshake_done, on_handshake_done, args);
// While chaining between handshakers, we use args->user_data to
// store a pointer to the handshake manager. This will be
// changed to point to the caller-supplied user_data before calling
// the final callback.
// the on_handshake_done callback.
args->user_data = mgr;
mgr->state->user_data = user_data;
call_next_handshaker(exec_ctx, args, GRPC_ERROR_NONE);
mgr->user_data = user_data;
// Start deadline timer, which owns a ref.
gpr_ref(&mgr->refs);
grpc_timer_init(exec_ctx, &mgr->deadline_timer,
gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
on_timeout, mgr, gpr_now(GPR_CLOCK_MONOTONIC));
// Start first handshaker, which also owns a ref.
gpr_ref(&mgr->refs);
call_next_handshaker_locked(exec_ctx, mgr, args, GRPC_ERROR_NONE);
gpr_mu_unlock(&mgr->mu);
}

@ -80,7 +80,6 @@ typedef struct {
/// When finished, invokes \a on_handshake_done.
/// \a acceptor will be NULL for client-side handshakers.
void (*do_handshake)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker,
gpr_timespec deadline,
grpc_tcp_server_acceptor* acceptor,
grpc_closure* on_handshake_done,
grpc_handshaker_args* args);

Loading…
Cancel
Save