aboutsummaryrefslogtreecommitdiffhomepage
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/core/end2end/fixtures/h2_http_proxy.c10
-rw-r--r--test/core/end2end/fixtures/http_proxy.c138
-rw-r--r--test/core/end2end/fixtures/http_proxy.h2
3 files changed, 119 insertions, 31 deletions
diff --git a/test/core/end2end/fixtures/h2_http_proxy.c b/test/core/end2end/fixtures/h2_http_proxy.c
index d84f0b8cb9..002e2cb16b 100644
--- a/test/core/end2end/fixtures/h2_http_proxy.c
+++ b/test/core/end2end/fixtures/h2_http_proxy.c
@@ -52,6 +52,7 @@
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
+#if 0
typedef struct fullstack_fixture_data {
char *server_addr;
grpc_end2end_http_proxy *proxy;
@@ -110,17 +111,24 @@ static grpc_end2end_test_config configs[] = {
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
};
+#endif
int main(int argc, char **argv) {
- size_t i;
+// size_t i;
grpc_test_init(argc, argv);
grpc_end2end_tests_pre_init();
grpc_init();
+ grpc_end2end_http_proxy* proxy = grpc_end2end_http_proxy_create();
+ grpc_end2end_http_proxy_start_thread(proxy);
+ grpc_end2end_http_proxy_destroy(proxy);
+
+#if 0
for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) {
grpc_end2end_tests(argc, argv, configs[i]);
}
+#endif
grpc_shutdown();
diff --git a/test/core/end2end/fixtures/http_proxy.c b/test/core/end2end/fixtures/http_proxy.c
index a8d68f0249..975deaeb30 100644
--- a/test/core/end2end/fixtures/http_proxy.c
+++ b/test/core/end2end/fixtures/http_proxy.c
@@ -41,7 +41,7 @@
#include <grpc/support/slice_buffer.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
-//#include <grpc/support/thd.h>
+#include <grpc/support/thd.h>
#include <grpc/support/useful.h>
#include "src/core/lib/channel/channel_args.h"
@@ -58,6 +58,16 @@
#include "src/core/lib/iomgr/tcp_server.h"
#include "test/core/util/port.h"
+struct grpc_end2end_http_proxy {
+ char* proxy_name;
+ gpr_thd_id thd;
+ grpc_tcp_server* server;
+ grpc_channel_args* channel_args;
+ gpr_mu* mu;
+ grpc_pollset* pollset;
+ bool shutdown;
+};
+
//
// Connection handling
//
@@ -83,10 +93,16 @@ typedef struct connection_data {
grpc_http_parser http_parser;
grpc_http_request http_request;
+
+ grpc_end2end_http_proxy* proxy;
+
+ gpr_refcount refcount;
} connection_data;
static void connection_data_destroy(grpc_exec_ctx* exec_ctx,
connection_data* cd) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
+ cd->proxy->shutdown = true;
grpc_endpoint_destroy(exec_ctx, cd->client_endpoint);
if (cd->server_endpoint != NULL)
grpc_endpoint_destroy(exec_ctx, cd->server_endpoint);
@@ -103,62 +119,96 @@ static void connection_data_destroy(grpc_exec_ctx* exec_ctx,
static void connection_data_failed(grpc_exec_ctx* exec_ctx,
connection_data* cd, const char* prefix,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
const char* msg = grpc_error_string(error);
gpr_log(GPR_ERROR, "%s: %s", prefix, msg);
grpc_error_free_string(msg);
GRPC_ERROR_UNREF(error);
+gpr_log(GPR_ERROR, "HERE 0");
grpc_endpoint_shutdown(exec_ctx, cd->client_endpoint);
+gpr_log(GPR_ERROR, "HERE 1");
if (cd->server_endpoint != NULL)
grpc_endpoint_shutdown(exec_ctx, cd->server_endpoint);
- connection_data_destroy(exec_ctx, cd);
+gpr_log(GPR_ERROR, "HERE 2");
+ if (gpr_unref(&cd->refcount)) {
+gpr_log(GPR_ERROR, "HERE 2.5");
+ connection_data_destroy(exec_ctx, cd);
+ }
+gpr_log(GPR_ERROR, "HERE 3");
}
static void on_client_write_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
- if (error != GRPC_ERROR_NONE)
+ if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy client write", error);
+ return;
+ }
+ // Clear write buffer.
+ gpr_slice_buffer_reset_and_unref(&cd->client_write_buffer);
}
static void on_server_write_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
- if (error != GRPC_ERROR_NONE)
+ if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy server write", error);
+ return;
+ }
+ // Clear write buffer.
+ gpr_slice_buffer_reset_and_unref(&cd->server_write_buffer);
}
static void on_client_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy client read", error);
return;
}
+ // Move read data into write buffer and write it.
gpr_slice_buffer_move_into(&cd->client_read_buffer, &cd->server_write_buffer);
grpc_endpoint_write(exec_ctx, cd->server_endpoint, &cd->server_write_buffer,
&cd->on_server_write_done);
+ // Read more data.
+ grpc_endpoint_read(exec_ctx, cd->client_endpoint, &cd->client_read_buffer,
+ &cd->on_client_read_done);
}
static void on_server_read_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy server read", error);
return;
}
+ // Move read data into write buffer and write it.
gpr_slice_buffer_move_into(&cd->server_read_buffer, &cd->client_write_buffer);
grpc_endpoint_write(exec_ctx, cd->client_endpoint, &cd->client_write_buffer,
&cd->on_client_write_done);
+ // Read more data.
+ grpc_endpoint_read(exec_ctx, cd->server_endpoint, &cd->server_read_buffer,
+ &cd->on_server_read_done);
}
static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy write response", error);
return;
}
- // Set up proxying.
+ // Clear write buffer.
+ gpr_slice_buffer_reset_and_unref(&cd->client_write_buffer);
+ // Start reading from both client and server.
+ // We increase the refcount by one, since we already held one reference
+ // for ourselves, and there will now be two pending callbacks.
+ gpr_ref(&cd->refcount);
grpc_endpoint_read(exec_ctx, cd->client_endpoint, &cd->client_read_buffer,
&cd->on_client_read_done);
grpc_endpoint_read(exec_ctx, cd->server_endpoint, &cd->server_read_buffer,
@@ -167,6 +217,7 @@ static void on_write_response_done(grpc_exec_ctx* exec_ctx, void* arg,
static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy server connect", error);
@@ -174,8 +225,7 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
}
// We've established a connection, so send back a 200 response code to
// the client.
- gpr_slice slice = gpr_slice_from_copied_string("200 connected\r\n");
- gpr_slice_buffer_reset_and_unref(&cd->client_write_buffer);
+ gpr_slice slice = gpr_slice_from_copied_string("200 connected\r\n\r\n");
gpr_slice_buffer_add(&cd->client_write_buffer, slice);
grpc_endpoint_write(exec_ctx, cd->client_endpoint, &cd->client_write_buffer,
&cd->on_write_response_done);
@@ -183,6 +233,7 @@ static void on_server_connect_done(grpc_exec_ctx* exec_ctx, void* arg,
static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg,
grpc_error* error) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
connection_data* cd = arg;
if (error != GRPC_ERROR_NONE) {
connection_data_failed(exec_ctx, cd, "HTTP proxy read request", error);
@@ -240,12 +291,14 @@ static void on_accept(grpc_exec_ctx* exec_ctx, void* arg,
grpc_endpoint* ep, grpc_pollset* accepting_pollset,
grpc_tcp_server_acceptor* acceptor) {
// FIXME: remove
-gpr_log(GPR_ERROR, "==> on_accept()");
+gpr_log(GPR_ERROR, "==> %s()", __func__);
+ grpc_end2end_http_proxy* proxy = arg;
// Instantiate connection_data.
connection_data* cd = gpr_malloc(sizeof(*cd));
memset(cd, 0, sizeof(*cd));
cd->client_endpoint = ep;
cd->pollset_set = grpc_pollset_set_create();
+ grpc_pollset_set_add_pollset(exec_ctx, cd->pollset_set, proxy->pollset);
grpc_closure_init(&cd->on_read_request_done, on_read_request_done, cd);
grpc_closure_init(&cd->on_server_connect_done, on_server_connect_done, cd);
grpc_closure_init(&cd->on_write_response_done, on_write_response_done, cd);
@@ -259,6 +312,8 @@ gpr_log(GPR_ERROR, "==> on_accept()");
gpr_slice_buffer_init(&cd->server_write_buffer);
grpc_http_parser_init(&cd->http_parser, GRPC_HTTP_REQUEST,
&cd->http_request);
+ cd->proxy = proxy;
+ gpr_ref_init(&cd->refcount, 1);
grpc_endpoint_read(exec_ctx, cd->client_endpoint, &cd->client_read_buffer,
&cd->on_read_request_done);
}
@@ -267,25 +322,6 @@ gpr_log(GPR_ERROR, "==> on_accept()");
// Proxy class
//
-struct grpc_end2end_http_proxy {
- char* proxy_name;
-// gpr_thd_id thd;
- grpc_tcp_server* server;
- grpc_channel_args* channel_args;
- gpr_mu* mu;
- grpc_pollset* pollset;
-};
-
-#if 0
-static void thread_main(void *arg) {
- //grpc_end2end_http_proxy *proxy = arg;
- grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- while (true) {
- grpc_exec_ctx_flush(&exec_ctx);
- }
-}
-#endif
-
grpc_end2end_http_proxy* grpc_end2end_http_proxy_create() {
grpc_end2end_http_proxy* proxy = gpr_malloc(sizeof(*proxy));
memset(proxy, 0, sizeof(*proxy));
@@ -315,7 +351,7 @@ gpr_log(GPR_ERROR, "Proxy address: %s", proxy->proxy_name);
grpc_pollset_init(proxy->pollset, &proxy->mu);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_tcp_server_start(&exec_ctx, proxy->server, &proxy->pollset, 1,
- on_accept, NULL);
+ on_accept, proxy);
grpc_exec_ctx_finish(&exec_ctx);
#if 0
// Start proxy thread.
@@ -331,15 +367,21 @@ static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p,
grpc_pollset_destroy(p);
}
+// FIXME: remove (including all references below)
+//#define USE_THREAD 1
+
void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_tcp_server_shutdown_listeners(&exec_ctx, proxy->server);
grpc_tcp_server_unref(&exec_ctx, proxy->server);
-// gpr_thd_join(proxy->thd);
+#ifdef USE_THREAD
+ gpr_thd_join(proxy->thd);
+#endif
gpr_free(proxy->proxy_name);
grpc_channel_args_destroy(proxy->channel_args);
grpc_closure destroyed;
- grpc_closure_init(&destroyed, destroy_pollset, &proxy->pollset);
+ grpc_closure_init(&destroyed, destroy_pollset, proxy->pollset);
grpc_pollset_shutdown(&exec_ctx, proxy->pollset, &destroyed);
gpr_free(proxy);
grpc_exec_ctx_finish(&exec_ctx);
@@ -349,3 +391,39 @@ const char *grpc_end2end_http_proxy_get_proxy_name(
grpc_end2end_http_proxy *proxy) {
return proxy->proxy_name;
}
+
+static void thread_main(void* arg) {
+gpr_log(GPR_ERROR, "==> %s()", __func__);
+ grpc_end2end_http_proxy *proxy = arg;
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ do {
+gpr_log(GPR_ERROR, "HERE a");
+ const gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
+ const gpr_timespec deadline =
+ gpr_time_add(now, gpr_time_from_seconds(5, GPR_TIMESPAN));
+ grpc_pollset_worker *worker = NULL;
+gpr_log(GPR_ERROR, "HERE b");
+ gpr_mu_lock(proxy->mu);
+gpr_log(GPR_ERROR, "HERE c");
+ GRPC_LOG_IF_ERROR("grpc_pollset_work",
+ grpc_pollset_work(&exec_ctx, proxy->pollset, &worker,
+ now, deadline));
+gpr_log(GPR_ERROR, "HERE d");
+ gpr_mu_unlock(proxy->mu);
+gpr_log(GPR_ERROR, "HERE e");
+ grpc_exec_ctx_flush(&exec_ctx);
+gpr_log(GPR_ERROR, "HERE f");
+ } while (!proxy->shutdown);
+gpr_log(GPR_ERROR, "HERE g");
+ grpc_exec_ctx_finish(&exec_ctx);
+}
+
+void grpc_end2end_http_proxy_start_thread(grpc_end2end_http_proxy *proxy) {
+#ifdef USE_THREAD
+ gpr_thd_options opt = gpr_thd_options_default();
+ gpr_thd_options_set_joinable(&opt);
+ GPR_ASSERT(gpr_thd_new(&proxy->thd, thread_main, proxy, &opt));
+#else
+ thread_main(proxy);
+#endif
+}
diff --git a/test/core/end2end/fixtures/http_proxy.h b/test/core/end2end/fixtures/http_proxy.h
index 7af2ea92d0..444324e7e0 100644
--- a/test/core/end2end/fixtures/http_proxy.h
+++ b/test/core/end2end/fixtures/http_proxy.h
@@ -39,3 +39,5 @@ void grpc_end2end_http_proxy_destroy(grpc_end2end_http_proxy* proxy);
const char *grpc_end2end_http_proxy_get_proxy_name(
grpc_end2end_http_proxy *proxy);
+
+void grpc_end2end_http_proxy_start_thread(grpc_end2end_http_proxy *proxy);
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-21 04:27:44 +0000