diff --git a/test/core/end2end/fixtures/h2_sockpair+trace.c b/test/core/end2end/fixtures/h2_sockpair+trace.c index b8a5257ab2a..fe16d056513 100644 --- a/test/core/end2end/fixtures/h2_sockpair+trace.c +++ b/test/core/end2end/fixtures/h2_sockpair+trace.c @@ -91,7 +91,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( f.fixture_data = sfd; f.cq = grpc_completion_queue_create(NULL); - *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + *sfd = grpc_iomgr_create_endpoint_pair("fixture", buffer_pool, 65536); + grpc_buffer_pool_unref(buffer_pool); return f; } diff --git a/test/core/end2end/fixtures/h2_sockpair.c b/test/core/end2end/fixtures/h2_sockpair.c index a57990d6e73..77847adb9c7 100644 --- a/test/core/end2end/fixtures/h2_sockpair.c +++ b/test/core/end2end/fixtures/h2_sockpair.c @@ -90,7 +90,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( f.fixture_data = sfd; f.cq = grpc_completion_queue_create(NULL); - *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + *sfd = grpc_iomgr_create_endpoint_pair("fixture", buffer_pool, 65536); + grpc_buffer_pool_unref(buffer_pool); return f; } diff --git a/test/core/end2end/fixtures/h2_sockpair_1byte.c b/test/core/end2end/fixtures/h2_sockpair_1byte.c index 50aac8045a9..04678f168a7 100644 --- a/test/core/end2end/fixtures/h2_sockpair_1byte.c +++ b/test/core/end2end/fixtures/h2_sockpair_1byte.c @@ -90,7 +90,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair( f.fixture_data = sfd; f.cq = grpc_completion_queue_create(NULL); - *sfd = grpc_iomgr_create_endpoint_pair("fixture", 1); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + *sfd = grpc_iomgr_create_endpoint_pair("fixture", buffer_pool, 1); + grpc_buffer_pool_unref(buffer_pool); return f; } diff --git a/test/core/end2end/fixtures/http_proxy.c b/test/core/end2end/fixtures/http_proxy.c index 22533b9694c..bc24cb33cf6 100644 --- a/test/core/end2end/fixtures/http_proxy.c +++ b/test/core/end2end/fixtures/http_proxy.c @@ -357,7 +357,7 @@ static void on_read_request_done(grpc_exec_ctx* exec_ctx, void* arg, const gpr_timespec deadline = gpr_time_add( gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(10, GPR_TIMESPAN)); grpc_tcp_client_connect(exec_ctx, &conn->on_server_connect_done, - &conn->server_endpoint, conn->pollset_set, + &conn->server_endpoint, conn->pollset_set, NULL, (struct sockaddr*)&resolved_addresses->addrs[0].addr, resolved_addresses->addrs[0].len, deadline); grpc_resolved_addresses_destroy(resolved_addresses); @@ -417,7 +417,8 @@ static void thread_main(void* arg) { grpc_exec_ctx_finish(&exec_ctx); } -grpc_end2end_http_proxy* grpc_end2end_http_proxy_create() { +grpc_end2end_http_proxy* grpc_end2end_http_proxy_create(void) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_end2end_http_proxy* proxy = gpr_malloc(sizeof(*proxy)); memset(proxy, 0, sizeof(*proxy)); // Construct proxy address. @@ -426,8 +427,8 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create() { gpr_log(GPR_INFO, "Proxy address: %s", proxy->proxy_name); // Create TCP server. proxy->channel_args = grpc_channel_args_copy(NULL); - grpc_error* error = - grpc_tcp_server_create(NULL, proxy->channel_args, &proxy->server); + grpc_error* error = grpc_tcp_server_create( + &exec_ctx, NULL, proxy->channel_args, &proxy->server); GPR_ASSERT(error == GRPC_ERROR_NONE); // Bind to port. struct sockaddr_in addr; @@ -442,7 +443,6 @@ grpc_end2end_http_proxy* grpc_end2end_http_proxy_create() { // Start server. proxy->pollset = gpr_malloc(grpc_pollset_size()); 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, proxy); grpc_exec_ctx_finish(&exec_ctx); diff --git a/test/core/end2end/fuzzers/api_fuzzer.c b/test/core/end2end/fuzzers/api_fuzzer.c index 96ea82d95ea..6d676dd02e6 100644 --- a/test/core/end2end/fuzzers/api_fuzzer.c +++ b/test/core/end2end/fuzzers/api_fuzzer.c @@ -173,6 +173,7 @@ static bool is_eof(input_stream *inp) { return inp->cur == inp->end; } static gpr_timespec g_now; static grpc_server *g_server; static grpc_channel *g_channel; +static grpc_buffer_pool *g_buffer_pool; extern gpr_timespec (*gpr_now_impl)(gpr_clock_type clock_type); @@ -252,7 +253,7 @@ static void do_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { } else if (g_server != NULL) { grpc_endpoint *client; grpc_endpoint *server; - grpc_passthru_endpoint_create(&client, &server); + grpc_passthru_endpoint_create(&client, &server, g_buffer_pool); *fc->ep = client; grpc_transport *transport = @@ -520,6 +521,7 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { int pending_pings = 0; g_active_call = new_call(NULL, ROOT); + g_buffer_pool = grpc_buffer_pool_create(); grpc_completion_queue *cq = grpc_completion_queue_create(NULL); @@ -939,6 +941,11 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { } break; } + // resize the buffer pool + case 21: { + grpc_buffer_pool_resize(g_buffer_pool, read_uint22(&inp)); + break; + } } } @@ -954,6 +961,8 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { .type == GRPC_QUEUE_SHUTDOWN); grpc_completion_queue_destroy(cq); + grpc_buffer_pool_unref(g_buffer_pool); + grpc_shutdown(); return 0; } diff --git a/test/core/end2end/fuzzers/client_fuzzer.c b/test/core/end2end/fuzzers/client_fuzzer.c index 00e650a30b9..90c35e6ad90 100644 --- a/test/core/end2end/fuzzers/client_fuzzer.c +++ b/test/core/end2end/fuzzers/client_fuzzer.c @@ -58,7 +58,10 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { grpc_init(); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_endpoint *mock_endpoint = grpc_mock_endpoint_create(discard_write); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + grpc_endpoint *mock_endpoint = + grpc_mock_endpoint_create(discard_write, buffer_pool); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_transport *transport = diff --git a/test/core/end2end/fuzzers/server_fuzzer.c b/test/core/end2end/fuzzers/server_fuzzer.c index 79eaad70c5d..cb34927c966 100644 --- a/test/core/end2end/fuzzers/server_fuzzer.c +++ b/test/core/end2end/fuzzers/server_fuzzer.c @@ -56,7 +56,10 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) { grpc_init(); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - grpc_endpoint *mock_endpoint = grpc_mock_endpoint_create(discard_write); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + grpc_endpoint *mock_endpoint = + grpc_mock_endpoint_create(discard_write, buffer_pool); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_mock_endpoint_put_read( &exec_ctx, mock_endpoint, gpr_slice_from_copied_buffer((const char *)data, size)); diff --git a/test/core/iomgr/tcp_client_posix_test.c b/test/core/iomgr/tcp_client_posix_test.c index d0c10474239..a8ba0a17e48 100644 --- a/test/core/iomgr/tcp_client_posix_test.c +++ b/test/core/iomgr/tcp_client_posix_test.c @@ -111,7 +111,7 @@ void test_succeeds(void) { /* connect to it */ GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0); grpc_closure_init(&done, must_succeed, NULL); - grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, g_pollset_set, + grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, g_pollset_set, NULL, (struct sockaddr *)&addr, addr_len, gpr_inf_future(GPR_CLOCK_REALTIME)); @@ -160,7 +160,7 @@ void test_fails(void) { /* connect to a broken address */ grpc_closure_init(&done, must_fail, NULL); - grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, g_pollset_set, + grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, g_pollset_set, NULL, (struct sockaddr *)&addr, addr_len, gpr_inf_future(GPR_CLOCK_REALTIME)); diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c index 42614567ca9..c2f857020aa 100644 --- a/test/core/iomgr/tcp_posix_test.c +++ b/test/core/iomgr/tcp_posix_test.c @@ -176,7 +176,10 @@ static void read_test(size_t num_bytes, size_t slice_size) { create_sockets(sv); - ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), slice_size, "test"); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), buffer_pool, + slice_size, "test"); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); written_bytes = fill_socket_partial(sv[0], num_bytes); @@ -223,8 +226,10 @@ static void large_read_test(size_t slice_size) { create_sockets(sv); - ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), slice_size, - "test"); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), buffer_pool, + slice_size, "test"); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); written_bytes = fill_socket(sv[0]); @@ -359,8 +364,10 @@ static void write_test(size_t num_bytes, size_t slice_size) { create_sockets(sv); - ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test"), + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test"), buffer_pool, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, "test"); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); state.ep = ep; @@ -423,8 +430,11 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) { create_sockets(sv); - ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), slice_size, "test"); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), buffer_pool, + slice_size, "test"); GPR_ASSERT(grpc_tcp_fd(ep) == sv[1] && sv[1] >= 0); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset); written_bytes = fill_socket_partial(sv[0], num_bytes); @@ -506,10 +516,12 @@ static grpc_endpoint_test_fixture create_fixture_tcp_socketpair( grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; create_sockets(sv); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); f.client_ep = grpc_tcp_create(grpc_fd_create(sv[0], "fixture:client"), - slice_size, "test"); + buffer_pool, slice_size, "test"); f.server_ep = grpc_tcp_create(grpc_fd_create(sv[1], "fixture:server"), - slice_size, "test"); + buffer_pool, slice_size, "test"); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_endpoint_add_to_pollset(&exec_ctx, f.client_ep, g_pollset); grpc_endpoint_add_to_pollset(&exec_ctx, f.server_ep, g_pollset); diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c index 6e2d1d0fc9e..f747dbc6e39 100644 --- a/test/core/iomgr/tcp_server_posix_test.c +++ b/test/core/iomgr/tcp_server_posix_test.c @@ -129,7 +129,8 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp, static void test_no_op(void) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_tcp_server *s; - GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s)); + GPR_ASSERT(GRPC_ERROR_NONE == + grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); grpc_tcp_server_unref(&exec_ctx, s); grpc_exec_ctx_finish(&exec_ctx); } @@ -137,7 +138,8 @@ static void test_no_op(void) { static void test_no_op_with_start(void) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_tcp_server *s; - GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s)); + GPR_ASSERT(GRPC_ERROR_NONE == + grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); LOG_TEST("test_no_op_with_start"); grpc_tcp_server_start(&exec_ctx, s, NULL, 0, on_connect, NULL); grpc_tcp_server_unref(&exec_ctx, s); @@ -148,7 +150,8 @@ static void test_no_op_with_port(void) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; struct sockaddr_in addr; grpc_tcp_server *s; - GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s)); + GPR_ASSERT(GRPC_ERROR_NONE == + grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); LOG_TEST("test_no_op_with_port"); memset(&addr, 0, sizeof(addr)); @@ -166,7 +169,8 @@ static void test_no_op_with_port_and_start(void) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; struct sockaddr_in addr; grpc_tcp_server *s; - GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s)); + GPR_ASSERT(GRPC_ERROR_NONE == + grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); LOG_TEST("test_no_op_with_port_and_start"); int port; @@ -226,7 +230,8 @@ static void test_connect(unsigned n) { unsigned svr1_fd_count; int svr1_port; grpc_tcp_server *s; - GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s)); + GPR_ASSERT(GRPC_ERROR_NONE == + grpc_tcp_server_create(&exec_ctx, NULL, NULL, &s)); unsigned i; server_weak_ref weak_ref; server_weak_ref_init(&weak_ref); diff --git a/test/core/security/secure_endpoint_test.c b/test/core/security/secure_endpoint_test.c index 1d2bf73bb1d..abc59f0b3d6 100644 --- a/test/core/security/secure_endpoint_test.c +++ b/test/core/security/secure_endpoint_test.c @@ -56,7 +56,9 @@ static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair( grpc_endpoint_test_fixture f; grpc_endpoint_pair tcp; - tcp = grpc_iomgr_create_endpoint_pair("fixture", slice_size); + grpc_buffer_pool *buffer_pool = grpc_buffer_pool_create(); + tcp = grpc_iomgr_create_endpoint_pair("fixture", buffer_pool, slice_size); + grpc_buffer_pool_internal_unref(&exec_ctx, buffer_pool); grpc_endpoint_add_to_pollset(&exec_ctx, tcp.client, g_pollset); grpc_endpoint_add_to_pollset(&exec_ctx, tcp.server, g_pollset);