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

Merge pull request #13579 from sreecha/rq-fix

Browse Source 
Properly shutdown endpoints in h2_http_proxy
reviewable/pr10684/r13^2
Sree Kuchibhotla 7 years ago committed by GitHub
parent f7a8eb7b22 9a434371e2
commit f3b73e524f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 76 additions and 31 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. 107
      test/core/end2end/fixtures/http_proxy_fixture.cc

107
test/core/end2end/fixtures/http_proxy_fixture.cc
Unescape View File

@ -68,6 +68,9 @@ struct grpc_end2end_http_proxy {
// Connection handling // Connection handling
// //
// 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.
typedef struct proxy_connection { typedef struct proxy_connection {
grpc_end2end_http_proxy* proxy; grpc_end2end_http_proxy* proxy;
@ -78,6 +81,8 @@ typedef struct proxy_connection {
grpc_pollset_set* pollset_set; grpc_pollset_set* pollset_set;
// NOTE: All the closures execute under proxy->combiner lock. Which means
// there will not be any data-races between the closures
grpc_closure on_read_request_done; grpc_closure on_read_request_done;
grpc_closure on_server_connect_done; grpc_closure on_server_connect_done;
grpc_closure on_write_response_done; grpc_closure on_write_response_done;
@ -86,6 +91,13 @@ typedef struct proxy_connection {
grpc_closure on_server_read_done; grpc_closure on_server_read_done;
grpc_closure on_server_write_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_read_buffer;
grpc_slice_buffer client_deferred_write_buffer; grpc_slice_buffer client_deferred_write_buffer;
bool client_is_writing; bool client_is_writing;
@ -129,21 +141,53 @@ 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. // 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, static void proxy_connection_failed(grpc_exec_ctx* exec_ctx,
proxy_connection* conn, bool is_client, proxy_connection* conn,
const char* prefix, grpc_error* error) { failure_type failure, const char* prefix,
const char* msg = grpc_error_string(error); grpc_error* error) {
gpr_log(GPR_INFO, "%s: %s", prefix, msg); gpr_log(GPR_INFO, "%s: %s", prefix, grpc_error_string(error));
// Decide whether we should shut down the client and server.
grpc_endpoint_shutdown(exec_ctx, conn->client_endpoint, bool shutdown_client = false;
GRPC_ERROR_REF(error)); bool shutdown_server = false;
if (conn->server_endpoint != nullptr) { 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 ((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 &&
(conn->server_endpoint != nullptr)) {
grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint, grpc_endpoint_shutdown(exec_ctx, conn->server_endpoint,
GRPC_ERROR_REF(error)); GRPC_ERROR_REF(error));
conn->server_shutdown = true;
} }
// Unref the connection.
proxy_connection_unref(exec_ctx, conn, "conn_failed"); proxy_connection_unref(exec_ctx, conn, "conn_failed");
GRPC_ERROR_UNREF(error);
} }
// Callback for writing proxy data to the client. // Callback for writing proxy data to the client.
@ -152,8 +196,8 @@ static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
proxy_connection* conn = (proxy_connection*)arg; proxy_connection* conn = (proxy_connection*)arg;
conn->client_is_writing = false; conn->client_is_writing = false;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, CLIENT_WRITE_FAILED,
"HTTP proxy client write", error); "HTTP proxy client write", GRPC_ERROR_REF(error));
return; return;
} }
// Clear write buffer (the data we just wrote). // Clear write buffer (the data we just wrote).
@ -179,8 +223,8 @@ static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
proxy_connection* conn = (proxy_connection*)arg; proxy_connection* conn = (proxy_connection*)arg;
conn->server_is_writing = false; conn->server_is_writing = false;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, false /* is_client */, proxy_connection_failed(exec_ctx, conn, SERVER_WRITE_FAILED,
"HTTP proxy server write", error); "HTTP proxy server write", GRPC_ERROR_REF(error));
return; return;
} }
// Clear write buffer (the data we just wrote). // Clear write buffer (the data we just wrote).
@ -206,8 +250,8 @@ static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) { grpc_error* error) {
proxy_connection* conn = (proxy_connection*)arg; proxy_connection* conn = (proxy_connection*)arg;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, CLIENT_READ_FAILED,
"HTTP proxy client read", error); "HTTP proxy client read", GRPC_ERROR_REF(error));
return; return;
} }
// If there is already a pending write (i.e., server_write_buffer is // If there is already a pending write (i.e., server_write_buffer is
@ -239,8 +283,8 @@ static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) { grpc_error* error) {
proxy_connection* conn = (proxy_connection*)arg; proxy_connection* conn = (proxy_connection*)arg;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, false /* is_client */, proxy_connection_failed(exec_ctx, conn, SERVER_READ_FAILED,
"HTTP proxy server read", error); "HTTP proxy server read", GRPC_ERROR_REF(error));
return; return;
} }
// If there is already a pending write (i.e., client_write_buffer is // If there is already a pending write (i.e., client_write_buffer is
@ -272,8 +316,8 @@ static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg,
proxy_connection* conn = (proxy_connection*)arg; proxy_connection* conn = (proxy_connection*)arg;
conn->client_is_writing = false; conn->client_is_writing = false;
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy write response", error); "HTTP proxy write response", GRPC_ERROR_REF(error));
return; return;
} }
// Clear write buffer. // Clear write buffer.
@ -301,8 +345,8 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
// connection failed. However, for the purposes of this test code, // connection failed. However, for the purposes of this test code,
// it's fine to pretend this is a client-side error, which will // it's fine to pretend this is a client-side error, which will
// cause the client connection to be dropped. // cause the client connection to be dropped.
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy server connect", error); "HTTP proxy server connect", GRPC_ERROR_REF(error));
return; return;
} }
// We've established a connection, so send back a 200 response code to // We've established a connection, so send back a 200 response code to
@ -351,8 +395,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, gpr_log(GPR_DEBUG, "on_read_request_done: %p %s", conn,
grpc_error_string(error)); grpc_error_string(error));
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy read request", error); "HTTP proxy read request", GRPC_ERROR_REF(error));
return; return;
} }
// Read request and feed it to the parser. // Read request and feed it to the parser.
@ -361,8 +405,9 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
error = grpc_http_parser_parse( error = grpc_http_parser_parse(
&conn->http_parser, conn->client_read_buffer.slices[i], nullptr); &conn->http_parser, conn->client_read_buffer.slices[i], nullptr);
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy request parse", error); "HTTP proxy request parse",
GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(error);
return; return;
} }
@ -382,8 +427,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
conn->http_request.method); conn->http_request.method);
error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
gpr_free(msg); gpr_free(msg);
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy read request", error); "HTTP proxy read request", GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(error);
return; return;
} }
@ -403,8 +448,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
if (!client_authenticated) { if (!client_authenticated) {
const char* msg = "HTTP Connect could not verify authentication"; const char* msg = "HTTP Connect could not verify authentication";
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(msg); error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(msg);
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy read request", error); "HTTP proxy read request", GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(error);
return; return;
} }
@ -414,8 +459,8 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
error = grpc_blocking_resolve_address(conn->http_request.path, "80", error = grpc_blocking_resolve_address(conn->http_request.path, "80",
&resolved_addresses); &resolved_addresses);
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
proxy_connection_failed(exec_ctx, conn, true /* is_client */, proxy_connection_failed(exec_ctx, conn, SETUP_FAILED,
"HTTP proxy DNS lookup", error); "HTTP proxy DNS lookup", GRPC_ERROR_REF(error));
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(error);
return; return;
} }

Write Preview
Loading…
Cancel
Save