Skip to content

Commit

Permalink
Merge pull request #10 from markdroth/http_proxy_patch
Browse files Browse the repository at this point in the history 
Provide a cleaner API for proxy_connection_failed().
  • Loading branch information
@sreecha
sreecha authored Dec 1, 2017
2 parents 906adfd + 1af674a commit 45abfe1
Showing 1 changed file with 70 additions and 68 deletions.
138 changes: 70 additions & 68 deletions test/core/end2end/fixtures/http_proxy_fixture.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,15 +68,6 @@ struct grpc_end2end_http_proxy {
// Connection handling
//

#define SERVER_EP_READ_FAIL 1
#define SERVER_EP_WRITE_FAIL 2
#define CLIENT_EP_READ_FAIL 4
#define CLIENT_EP_WRITE_FAIL 8

#define SERVER_EP_FAIL (SERVER_EP_READ_FAIL | SERVER_EP_WRITE_FAIL)
#define CLIENT_EP_FAIL (CLIENT_EP_READ_FAIL | CLIENT_EP_WRITE_FAIL)
#define EP_FAIL (SERVER_EP_FAIL | CLIENT_EP_FAIL)

// proxy_connection structure is only accessed in the closures which are all
// scheduled under the same combiner lock. So there is is no need for a mutex to
// protect this structure.
Expand All @@ -88,13 +79,6 @@ typedef struct proxy_connection {

gpr_refcount refcount;

// The lower four bits store the endpoint failure information
// bit-0 Server endpoint read failure
// bit-1 Server endpoint write failure
// bit-2 Client endpoint read failure
// bit-3 Client endpoint write failure
int ep_state;

grpc_pollset_set* pollset_set;

// NOTE: All the closures execute under proxy->combiner lock. Which means
Expand All @@ -107,6 +91,13 @@ typedef struct proxy_connection {
grpc_closure on_server_read_done;
grpc_closure on_server_write_done;

bool client_read_failed : 1;
bool client_write_failed : 1;
bool client_shutdown : 1;
bool server_read_failed : 1;
bool server_write_failed : 1;
bool server_shutdown : 1;

grpc_slice_buffer client_read_buffer;
grpc_slice_buffer client_deferred_write_buffer;
bool client_is_writing;
Expand Down Expand Up @@ -150,36 +141,52 @@ static void proxy_connection_unref(grpc_exec_ctx* exec_ctx,
}
}

enum failure_type {
SETUP_FAILED, // To be used before we start proxying.
CLIENT_READ_FAILED,
CLIENT_WRITE_FAILED,
SERVER_READ_FAILED,
SERVER_WRITE_FAILED,
};

// Helper function to shut down the proxy connection.
// Does NOT take ownership of a reference to error.
static void proxy_connection_failed(grpc_exec_ctx* exec_ctx,
proxy_connection* conn, int failure_type,
const char* prefix, grpc_error* error) {
const char* msg = grpc_error_string(error);
gpr_log(GPR_INFO, "%s: %s", prefix, msg);
int ep_state = conn->ep_state;
int new_ep_state = ep_state | failure_type;
if (ep_state != new_ep_state) {
conn->ep_state = new_ep_state;
// Shutdown the endpoint (client and/or server) if both read and write
// failures are observed after setting the failure_type.
// To prevent calling endpoint shutdown multiple times, It is important to
// ensure that ep_state i.e the old state, did not already have both
// failures set.
if (((ep_state & SERVER_EP_FAIL) != SERVER_EP_FAIL) &&
((new_ep_state & SERVER_EP_FAIL) == SERVER_EP_FAIL)) {
if (conn->server_endpoint != nullptr) {
grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint,
GRPC_ERROR_REF(error));
}
proxy_connection* conn,
failure_type failure, const char* prefix,
grpc_error* error) {
gpr_log(GPR_INFO, "%s: %s", prefix, grpc_error_string(error));
// Decide whether we should shut down the client and server.
bool shutdown_client = false;
bool shutdown_server = false;
if (failure == SETUP_FAILED) {
shutdown_client = true;
shutdown_server = true;
} else {
if ((failure == CLIENT_READ_FAILED && conn->client_write_failed) ||
(failure == CLIENT_WRITE_FAILED && conn->client_read_failed) ||
(failure == SERVER_READ_FAILED && !conn->client_is_writing)) {
shutdown_client = true;
}
if (((ep_state & CLIENT_EP_FAIL) != CLIENT_EP_FAIL) &&
((new_ep_state & CLIENT_EP_FAIL) == CLIENT_EP_FAIL)) {
grpc_endpoint_shutdown(exec_ctx, conn->client_endpoint,
GRPC_ERROR_REF(error));
if ((failure == SERVER_READ_FAILED && conn->server_write_failed) ||
(failure == SERVER_WRITE_FAILED && conn->server_read_failed) ||
(failure == CLIENT_READ_FAILED && !conn->server_is_writing)) {
shutdown_server = true;
}
}
// If we decided to shut down either one and have not yet done so, do so.
if (shutdown_client && !conn->client_shutdown) {
grpc_endpoint_shutdown(exec_ctx, conn->client_endpoint,
GRPC_ERROR_REF(error));
conn->client_shutdown = true;
}
if (shutdown_server && !conn->server_shutdown) {
grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint,
GRPC_ERROR_REF(error));
conn->server_shutdown = true;
}
// Unref the connection.
proxy_connection_unref(exec_ctx, conn, "conn_failed");
GRPC_ERROR_UNREF(error);
}

// Callback for writing proxy data to the client.
Expand All @@ -188,8 +195,8 @@ static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
proxy_connection* conn = (proxy_connection*)arg;
conn->client_is_writing = false;
if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, CLIENT_EP_WRITE_FAIL,
"HTTP proxy client write", error);
proxy_connection_failed(exec_ctx, conn, CLIENT_WRITE_FAILED,
"HTTP proxy client write", GRPC_ERROR_REF(error));
return;
}
// Clear write buffer (the data we just wrote).
Expand All @@ -215,8 +222,8 @@ static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
proxy_connection* conn = (proxy_connection*)arg;
conn->server_is_writing = false;
if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, SERVER_EP_WRITE_FAIL,
"HTTP proxy server write", error);
proxy_connection_failed(exec_ctx, conn, SERVER_WRITE_FAILED,
"HTTP proxy server write", GRPC_ERROR_REF(error));
return;
}
// Clear write buffer (the data we just wrote).
Expand Down Expand Up @@ -244,11 +251,8 @@ static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
if (error != GRPC_ERROR_NONE) {
// Report a read failure on the client endpoint. If there is no pending
// server write, then shutdown the server endpoint as well.
proxy_connection_failed(
exec_ctx, conn,
(conn->server_is_writing ? CLIENT_EP_READ_FAIL
: (CLIENT_EP_READ_FAIL | SERVER_EP_FAIL)),
"HTTP proxy client read", error);
proxy_connection_failed(exec_ctx, conn, CLIENT_READ_FAILED,
"HTTP proxy client read", GRPC_ERROR_REF(error));
return;
}
// If there is already a pending write (i.e., server_write_buffer is
Expand Down Expand Up @@ -282,11 +286,8 @@ static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
if (error != GRPC_ERROR_NONE) {
// Report a read failure on the server end point. If there is no pending
// write to the client, then shutdown the client endpoint as well.
proxy_connection_failed(
exec_ctx, conn,
(conn->client_is_writing ? SERVER_EP_READ_FAIL
: (SERVER_EP_READ_FAIL | CLIENT_EP_FAIL)),
"HTTP proxy server read", error);
proxy_connection_failed(exec_ctx, conn, SERVER_READ_FAILED,
"HTTP proxy server read", GRPC_ERROR_REF(error));
return;
}
// If there is already a pending write (i.e., client_write_buffer is
Expand Down Expand Up @@ -318,8 +319,8 @@ static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg,
proxy_connection* conn = (proxy_connection*)arg;
conn->client_is_writing = false;
if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, EP_FAIL,
"HTTP proxy write response", error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy write response", GRPC_ERROR_REF(error));
return;
}
// Clear write buffer.
Expand Down Expand Up @@ -347,8 +348,8 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
// connection failed. However, for the purposes of this test code,
// it's fine to pretend this is a client-side error, which will
// cause the client connection to be dropped.
proxy_connection_failed(exec_ctx, conn, EP_FAIL,
"HTTP proxy server connect", error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy server connect", GRPC_ERROR_REF(error));
return;
}
// We've established a connection, so send back a 200 response code to
Expand Down Expand Up @@ -397,8 +398,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
gpr_log(GPR_DEBUG, "on_read_request_done: %p %s", conn,
grpc_error_string(error));
if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, EP_FAIL, "HTTP proxy read request",
error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy read request", GRPC_ERROR_REF(error));
return;
}
// Read request and feed it to the parser.
Expand All @@ -407,8 +408,9 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
error = grpc_http_parser_parse(
&conn->http_parser, conn->client_read_buffer.slices[i], nullptr);
if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, EP_FAIL,
"HTTP proxy request parse", error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy request parse",
GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error);
return;
}
Expand All @@ -428,8 +430,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
conn->http_request.method);
error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
gpr_free(msg);
proxy_connection_failed(exec_ctx, conn, EP_FAIL, "HTTP proxy read request",
error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy read request", GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error);
return;
}
Expand All @@ -449,8 +451,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
if (!client_authenticated) {
const char* msg = "HTTP Connect could not verify authentication";
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(msg);
proxy_connection_failed(exec_ctx, conn, EP_FAIL,
"HTTP proxy read request", error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy read request", GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error);
return;
}
Expand All @@ -460,8 +462,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
error = grpc_blocking_resolve_address(conn->http_request.path, "80",
&resolved_addresses);
if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, EP_FAIL, "HTTP proxy DNS lookup",
error);
proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy DNS lookup", GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error);
return;
}
Expand Down

0 comments on commit 45abfe1

Please sign in to comment.