1 //
2 //
3 // Copyright 2015 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18
19 #include "absl/time/time.h"
20
21 #include "src/core/lib/channel/channel_args.h"
22 #include "src/core/lib/gprpp/notification.h"
23 #include "src/core/lib/gprpp/time.h"
24 #include "src/core/lib/iomgr/port.h"
25
26 // This test won't work except with posix sockets enabled
27 #ifdef GRPC_POSIX_SOCKET_TCP
28
29 #include <errno.h>
30 #include <fcntl.h>
31 #include <limits.h>
32 #include <string.h>
33 #include <sys/socket.h>
34 #include <sys/types.h>
35 #include <unistd.h>
36
37 #include <gtest/gtest.h>
38
39 #include <grpc/grpc.h>
40 #include <grpc/support/alloc.h>
41 #include <grpc/support/log.h>
42 #include <grpc/support/time.h>
43
44 #include "src/core/lib/event_engine/channel_args_endpoint_config.h"
45 #include "src/core/lib/event_engine/default_event_engine.h"
46 #include "src/core/lib/event_engine/posix.h"
47 #include "src/core/lib/event_engine/shim.h"
48 #include "src/core/lib/gpr/useful.h"
49 #include "src/core/lib/gprpp/crash.h"
50 #include "src/core/lib/iomgr/buffer_list.h"
51 #include "src/core/lib/iomgr/ev_posix.h"
52 #include "src/core/lib/iomgr/event_engine_shims/endpoint.h"
53 #include "src/core/lib/iomgr/sockaddr_posix.h"
54 #include "src/core/lib/iomgr/socket_utils_posix.h"
55 #include "src/core/lib/iomgr/tcp_posix.h"
56 #include "src/core/lib/slice/slice_internal.h"
57 #include "test/core/iomgr/endpoint_tests.h"
58 #include "test/core/util/test_config.h"
59
60 static gpr_mu* g_mu;
61 static grpc_pollset* g_pollset;
62
63 static constexpr int64_t kDeadlineMillis = 20000;
64
65 //
66 // General test notes:
67
68 // All tests which write data into a socket write i%256 into byte i, which is
69 // verified by readers.
70
71 // In general there are a few interesting things to vary which may lead to
72 // exercising different codepaths in an implementation:
73 // 1. Total amount of data written to the socket
74 // 2. Size of slice allocations
75 // 3. Amount of data we read from or write to the socket at once
76
77 // The tests here tend to parameterize these where applicable.
78
79 //
80
create_sockets(int sv[2])81 static void create_sockets(int sv[2]) {
82 int flags;
83 GPR_ASSERT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
84 flags = fcntl(sv[0], F_GETFL, 0);
85 GPR_ASSERT(fcntl(sv[0], F_SETFL, flags | O_NONBLOCK) == 0);
86 flags = fcntl(sv[1], F_GETFL, 0);
87 GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
88 }
89
fill_socket(int fd)90 static ssize_t fill_socket(int fd) {
91 ssize_t write_bytes;
92 ssize_t total_bytes = 0;
93 int i;
94 unsigned char buf[256];
95 for (i = 0; i < 256; ++i) {
96 buf[i] = static_cast<uint8_t>(i);
97 }
98 do {
99 write_bytes = write(fd, buf, 256);
100 if (write_bytes > 0) {
101 total_bytes += write_bytes;
102 }
103 } while (write_bytes >= 0 || errno == EINTR);
104 GPR_ASSERT(errno == EAGAIN);
105 return total_bytes;
106 }
107
fill_socket_partial(int fd,size_t bytes)108 static size_t fill_socket_partial(int fd, size_t bytes) {
109 ssize_t write_bytes;
110 size_t total_bytes = 0;
111 unsigned char* buf = static_cast<unsigned char*>(gpr_malloc(bytes));
112 unsigned i;
113 for (i = 0; i < bytes; ++i) {
114 buf[i] = static_cast<uint8_t>(i % 256);
115 }
116
117 do {
118 write_bytes = write(fd, buf, bytes - total_bytes);
119 if (write_bytes > 0) {
120 total_bytes += static_cast<size_t>(write_bytes);
121 }
122 } while ((write_bytes >= 0 || errno == EINTR) && bytes > total_bytes);
123
124 gpr_free(buf);
125
126 return total_bytes;
127 }
128
129 struct read_socket_state {
130 grpc_endpoint* ep;
131 size_t min_progress_size;
132 size_t read_bytes;
133 size_t target_read_bytes;
134 grpc_slice_buffer incoming;
135 grpc_closure read_cb;
136 };
137
count_slices(grpc_slice * slices,size_t nslices,int * current_data)138 static size_t count_slices(grpc_slice* slices, size_t nslices,
139 int* current_data) {
140 size_t num_bytes = 0;
141 unsigned i, j;
142 unsigned char* buf;
143 for (i = 0; i < nslices; ++i) {
144 buf = GRPC_SLICE_START_PTR(slices[i]);
145 for (j = 0; j < GRPC_SLICE_LENGTH(slices[i]); ++j) {
146 GPR_ASSERT(buf[j] == *current_data);
147 *current_data = (*current_data + 1) % 256;
148 }
149 num_bytes += GRPC_SLICE_LENGTH(slices[i]);
150 }
151 return num_bytes;
152 }
153
read_cb(void * user_data,grpc_error_handle error)154 static void read_cb(void* user_data, grpc_error_handle error) {
155 struct read_socket_state* state =
156 static_cast<struct read_socket_state*>(user_data);
157 size_t read_bytes;
158 int current_data;
159
160 GPR_ASSERT(error.ok());
161
162 gpr_mu_lock(g_mu);
163 current_data = state->read_bytes % 256;
164 // The number of bytes read each time this callback is invoked must be >=
165 // the min_progress_size.
166 if (grpc_core::IsTcpFrameSizeTuningEnabled()) {
167 GPR_ASSERT(state->min_progress_size <= state->incoming.length);
168 }
169 read_bytes = count_slices(state->incoming.slices, state->incoming.count,
170 ¤t_data);
171 state->read_bytes += read_bytes;
172 gpr_log(GPR_INFO, "Read %" PRIuPTR " bytes of %" PRIuPTR, read_bytes,
173 state->target_read_bytes);
174 if (state->read_bytes >= state->target_read_bytes) {
175 GPR_ASSERT(
176 GRPC_LOG_IF_ERROR("kick", grpc_pollset_kick(g_pollset, nullptr)));
177 gpr_mu_unlock(g_mu);
178 } else {
179 gpr_mu_unlock(g_mu);
180 state->min_progress_size = state->target_read_bytes - state->read_bytes;
181 grpc_endpoint_read(state->ep, &state->incoming, &state->read_cb,
182 /*urgent=*/false, state->min_progress_size);
183 }
184 }
185
186 // Write to a socket, then read from it using the grpc_tcp API.
read_test(size_t num_bytes,size_t slice_size,int min_progress_size)187 static void read_test(size_t num_bytes, size_t slice_size,
188 int min_progress_size) {
189 int sv[2];
190 grpc_endpoint* ep;
191 struct read_socket_state state;
192 size_t written_bytes;
193 grpc_core::Timestamp deadline = grpc_core::Timestamp::FromTimespecRoundUp(
194 grpc_timeout_milliseconds_to_deadline(kDeadlineMillis));
195 grpc_core::ExecCtx exec_ctx;
196
197 gpr_log(GPR_INFO, "Read test of size %" PRIuPTR ", slice size %" PRIuPTR,
198 num_bytes, slice_size);
199
200 create_sockets(sv);
201
202 grpc_arg a[2];
203 a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
204 a[0].type = GRPC_ARG_INTEGER,
205 a[0].value.integer = static_cast<int>(slice_size);
206 a[1].key = const_cast<char*>(GRPC_ARG_RESOURCE_QUOTA);
207 a[1].type = GRPC_ARG_POINTER;
208 a[1].value.pointer.p = grpc_resource_quota_create("test");
209 a[1].value.pointer.vtable = grpc_resource_quota_arg_vtable();
210 grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
211 ep = grpc_tcp_create(
212 grpc_fd_create(sv[1], "read_test", false),
213 TcpOptionsFromEndpointConfig(
214 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
215 grpc_core::ChannelArgs::FromC(&args))),
216 "test");
217 grpc_endpoint_add_to_pollset(ep, g_pollset);
218
219 written_bytes = fill_socket_partial(sv[0], num_bytes);
220 gpr_log(GPR_INFO, "Wrote %" PRIuPTR " bytes", written_bytes);
221
222 state.ep = ep;
223 state.read_bytes = 0;
224 state.target_read_bytes = written_bytes;
225 state.min_progress_size =
226 std::min(min_progress_size, static_cast<int>(written_bytes));
227 grpc_slice_buffer_init(&state.incoming);
228 GRPC_CLOSURE_INIT(&state.read_cb, read_cb, &state, grpc_schedule_on_exec_ctx);
229
230 grpc_endpoint_read(ep, &state.incoming, &state.read_cb, /*urgent=*/false,
231 /*min_progress_size=*/state.min_progress_size);
232 grpc_core::ExecCtx::Get()->Flush();
233 gpr_mu_lock(g_mu);
234 while (state.read_bytes < state.target_read_bytes) {
235 grpc_pollset_worker* worker = nullptr;
236 GPR_ASSERT(GRPC_LOG_IF_ERROR(
237 "pollset_work", grpc_pollset_work(g_pollset, &worker, deadline)));
238 gpr_mu_unlock(g_mu);
239 grpc_core::ExecCtx::Get()->Flush();
240 gpr_mu_lock(g_mu);
241 }
242 GPR_ASSERT(state.read_bytes == state.target_read_bytes);
243 gpr_mu_unlock(g_mu);
244
245 grpc_slice_buffer_destroy(&state.incoming);
246 grpc_endpoint_destroy(ep);
247 grpc_resource_quota_unref(
248 static_cast<grpc_resource_quota*>(a[1].value.pointer.p));
249 }
250
251 // Write to a socket until it fills up, then read from it using the grpc_tcp
252 // API.
large_read_test(size_t slice_size,int min_progress_size)253 static void large_read_test(size_t slice_size, int min_progress_size) {
254 int sv[2];
255 grpc_endpoint* ep;
256 struct read_socket_state state;
257 ssize_t written_bytes;
258 grpc_core::Timestamp deadline = grpc_core::Timestamp::FromTimespecRoundUp(
259 grpc_timeout_milliseconds_to_deadline(kDeadlineMillis));
260 grpc_core::ExecCtx exec_ctx;
261
262 gpr_log(GPR_INFO, "Start large read test, slice size %" PRIuPTR, slice_size);
263
264 create_sockets(sv);
265
266 grpc_arg a[2];
267 a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
268 a[0].type = GRPC_ARG_INTEGER;
269 a[0].value.integer = static_cast<int>(slice_size);
270 a[1].key = const_cast<char*>(GRPC_ARG_RESOURCE_QUOTA);
271 a[1].type = GRPC_ARG_POINTER;
272 a[1].value.pointer.p = grpc_resource_quota_create("test");
273 a[1].value.pointer.vtable = grpc_resource_quota_arg_vtable();
274 grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
275 ep = grpc_tcp_create(
276 grpc_fd_create(sv[1], "large_read_test", false),
277 TcpOptionsFromEndpointConfig(
278 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
279 grpc_core::ChannelArgs::FromC(&args))),
280 "test");
281 grpc_endpoint_add_to_pollset(ep, g_pollset);
282
283 written_bytes = fill_socket(sv[0]);
284 gpr_log(GPR_INFO, "Wrote %" PRIuPTR " bytes", written_bytes);
285
286 state.ep = ep;
287 state.read_bytes = 0;
288 state.target_read_bytes = static_cast<size_t>(written_bytes);
289 state.min_progress_size =
290 std::min(min_progress_size, static_cast<int>(written_bytes));
291 grpc_slice_buffer_init(&state.incoming);
292 GRPC_CLOSURE_INIT(&state.read_cb, read_cb, &state, grpc_schedule_on_exec_ctx);
293
294 grpc_endpoint_read(ep, &state.incoming, &state.read_cb, /*urgent=*/false,
295 /*min_progress_size=*/state.min_progress_size);
296 grpc_core::ExecCtx::Get()->Flush();
297 gpr_mu_lock(g_mu);
298 while (state.read_bytes < state.target_read_bytes) {
299 grpc_pollset_worker* worker = nullptr;
300 GPR_ASSERT(GRPC_LOG_IF_ERROR(
301 "pollset_work", grpc_pollset_work(g_pollset, &worker, deadline)));
302 gpr_mu_unlock(g_mu);
303 grpc_core::ExecCtx::Get()->Flush();
304 gpr_mu_lock(g_mu);
305 }
306 GPR_ASSERT(state.read_bytes == state.target_read_bytes);
307 gpr_mu_unlock(g_mu);
308
309 grpc_slice_buffer_destroy(&state.incoming);
310 grpc_endpoint_destroy(ep);
311 grpc_resource_quota_unref(
312 static_cast<grpc_resource_quota*>(a[1].value.pointer.p));
313 }
314
315 struct write_socket_state {
316 grpc_endpoint* ep;
317 int write_done;
318 };
319
allocate_blocks(size_t num_bytes,size_t slice_size,size_t * num_blocks,uint8_t * current_data)320 static grpc_slice* allocate_blocks(size_t num_bytes, size_t slice_size,
321 size_t* num_blocks, uint8_t* current_data) {
322 size_t nslices = num_bytes / slice_size + (num_bytes % slice_size ? 1u : 0u);
323 grpc_slice* slices =
324 static_cast<grpc_slice*>(gpr_malloc(sizeof(grpc_slice) * nslices));
325 size_t num_bytes_left = num_bytes;
326 unsigned i, j;
327 unsigned char* buf;
328 *num_blocks = nslices;
329
330 for (i = 0; i < nslices; ++i) {
331 slices[i] = grpc_slice_malloc(slice_size > num_bytes_left ? num_bytes_left
332 : slice_size);
333 num_bytes_left -= GRPC_SLICE_LENGTH(slices[i]);
334 buf = GRPC_SLICE_START_PTR(slices[i]);
335 for (j = 0; j < GRPC_SLICE_LENGTH(slices[i]); ++j) {
336 buf[j] = *current_data;
337 (*current_data)++;
338 }
339 }
340 GPR_ASSERT(num_bytes_left == 0);
341 return slices;
342 }
343
write_done(void * user_data,grpc_error_handle error)344 static void write_done(void* user_data /* write_socket_state */,
345 grpc_error_handle error) {
346 GPR_ASSERT(error.ok());
347 struct write_socket_state* state =
348 static_cast<struct write_socket_state*>(user_data);
349 gpr_mu_lock(g_mu);
350 state->write_done = 1;
351 GPR_ASSERT(
352 GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, nullptr)));
353 gpr_mu_unlock(g_mu);
354 }
355
drain_socket_blocking(int fd,size_t num_bytes,size_t read_size)356 void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
357 unsigned char* buf = static_cast<unsigned char*>(gpr_malloc(read_size));
358 ssize_t bytes_read;
359 size_t bytes_left = num_bytes;
360 int flags;
361 int current = 0;
362 int i;
363 grpc_core::ExecCtx exec_ctx;
364
365 flags = fcntl(fd, F_GETFL, 0);
366 GPR_ASSERT(fcntl(fd, F_SETFL, flags & ~O_NONBLOCK) == 0);
367
368 for (;;) {
369 grpc_pollset_worker* worker = nullptr;
370 gpr_mu_lock(g_mu);
371 GPR_ASSERT(GRPC_LOG_IF_ERROR(
372 "pollset_work",
373 grpc_pollset_work(g_pollset, &worker,
374 grpc_core::Timestamp::FromTimespecRoundUp(
375 grpc_timeout_milliseconds_to_deadline(10)))));
376 gpr_mu_unlock(g_mu);
377
378 do {
379 bytes_read =
380 read(fd, buf, bytes_left > read_size ? read_size : bytes_left);
381 } while (bytes_read < 0 && errno == EINTR);
382 GPR_ASSERT(bytes_read >= 0);
383 for (i = 0; i < bytes_read; ++i) {
384 GPR_ASSERT(buf[i] == current);
385 current = (current + 1) % 256;
386 }
387 bytes_left -= static_cast<size_t>(bytes_read);
388 if (bytes_left == 0) break;
389 }
390 flags = fcntl(fd, F_GETFL, 0);
391 GPR_ASSERT(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
392
393 gpr_free(buf);
394 }
395
396 // Write to a socket using the grpc_tcp API, then drain it directly.
397 // Note that if the write does not complete immediately we need to drain the
398 // socket in parallel with the read. If collect_timestamps is true, it will
399 // try to get timestamps for the write.
write_test(size_t num_bytes,size_t slice_size)400 static void write_test(size_t num_bytes, size_t slice_size) {
401 int sv[2];
402 grpc_endpoint* ep;
403 struct write_socket_state state;
404 size_t num_blocks;
405 grpc_slice* slices;
406 uint8_t current_data = 0;
407 grpc_slice_buffer outgoing;
408 grpc_closure write_done_closure;
409 grpc_core::Timestamp deadline = grpc_core::Timestamp::FromTimespecRoundUp(
410 grpc_timeout_milliseconds_to_deadline(kDeadlineMillis));
411 grpc_core::ExecCtx exec_ctx;
412
413 gpr_log(GPR_INFO,
414 "Start write test with %" PRIuPTR " bytes, slice size %" PRIuPTR,
415 num_bytes, slice_size);
416
417 create_sockets(sv);
418
419 grpc_arg a[2];
420 a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
421 a[0].type = GRPC_ARG_INTEGER,
422 a[0].value.integer = static_cast<int>(slice_size);
423 a[1].key = const_cast<char*>(GRPC_ARG_RESOURCE_QUOTA);
424 a[1].type = GRPC_ARG_POINTER;
425 a[1].value.pointer.p = grpc_resource_quota_create("test");
426 a[1].value.pointer.vtable = grpc_resource_quota_arg_vtable();
427 grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
428 ep = grpc_tcp_create(
429 grpc_fd_create(sv[1], "write_test", false),
430 TcpOptionsFromEndpointConfig(
431 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
432 grpc_core::ChannelArgs::FromC(&args))),
433 "test");
434 grpc_endpoint_add_to_pollset(ep, g_pollset);
435
436 state.ep = ep;
437 state.write_done = 0;
438
439 slices = allocate_blocks(num_bytes, slice_size, &num_blocks, ¤t_data);
440
441 grpc_slice_buffer_init(&outgoing);
442 grpc_slice_buffer_addn(&outgoing, slices, num_blocks);
443 GRPC_CLOSURE_INIT(&write_done_closure, write_done, &state,
444 grpc_schedule_on_exec_ctx);
445
446 grpc_endpoint_write(ep, &outgoing, &write_done_closure, nullptr,
447 /*max_frame_size=*/INT_MAX);
448 drain_socket_blocking(sv[0], num_bytes, num_bytes);
449 exec_ctx.Flush();
450 gpr_mu_lock(g_mu);
451 for (;;) {
452 grpc_pollset_worker* worker = nullptr;
453 if (state.write_done) {
454 break;
455 }
456 GPR_ASSERT(GRPC_LOG_IF_ERROR(
457 "pollset_work", grpc_pollset_work(g_pollset, &worker, deadline)));
458 gpr_mu_unlock(g_mu);
459 exec_ctx.Flush();
460 gpr_mu_lock(g_mu);
461 }
462 gpr_mu_unlock(g_mu);
463
464 grpc_slice_buffer_destroy(&outgoing);
465 grpc_endpoint_destroy(ep);
466 gpr_free(slices);
467 grpc_resource_quota_unref(
468 static_cast<grpc_resource_quota*>(a[1].value.pointer.p));
469 }
470
471 struct release_fd_arg {
472 std::atomic<int> fd_released_done{0};
473 grpc_core::Notification notify;
474 };
475
on_fd_released(void * arg,grpc_error_handle)476 void on_fd_released(void* arg, grpc_error_handle /*errors*/) {
477 release_fd_arg* rel_fd = static_cast<release_fd_arg*>(arg);
478 rel_fd->fd_released_done = 1;
479 rel_fd->notify.Notify();
480 }
481
482 // Do a read_test, then release fd and try to read/write again. Verify that
483 // grpc_tcp_fd() is available before the fd is released.
release_fd_test(size_t num_bytes,size_t slice_size)484 static void release_fd_test(size_t num_bytes, size_t slice_size) {
485 int sv[2];
486 grpc_endpoint* ep;
487 struct read_socket_state state;
488 size_t written_bytes;
489 int fd;
490 grpc_core::Timestamp deadline = grpc_core::Timestamp::FromTimespecRoundUp(
491 grpc_timeout_milliseconds_to_deadline(kDeadlineMillis));
492 grpc_core::ExecCtx exec_ctx;
493 grpc_closure fd_released_cb;
494 release_fd_arg rel_fd;
495 GRPC_CLOSURE_INIT(&fd_released_cb, &on_fd_released, &rel_fd,
496 grpc_schedule_on_exec_ctx);
497
498 gpr_log(GPR_INFO,
499 "Release fd read_test of size %" PRIuPTR ", slice size %" PRIuPTR,
500 num_bytes, slice_size);
501
502 create_sockets(sv);
503
504 grpc_arg a[2];
505 a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
506 a[0].type = GRPC_ARG_INTEGER;
507 a[0].value.integer = static_cast<int>(slice_size);
508 a[1].key = const_cast<char*>(GRPC_ARG_RESOURCE_QUOTA);
509 a[1].type = GRPC_ARG_POINTER;
510 a[1].value.pointer.p = grpc_resource_quota_create("test");
511 a[1].value.pointer.vtable = grpc_resource_quota_arg_vtable();
512 auto memory_quota = std::make_unique<grpc_core::MemoryQuota>("bar");
513 grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
514 if (grpc_event_engine::experimental::UseEventEngineListener()) {
515 // Create an event engine wrapped endpoint to test release_fd operations.
516 auto eeep =
517 reinterpret_cast<
518 grpc_event_engine::experimental::PosixEventEngineWithFdSupport*>(
519 grpc_event_engine::experimental::GetDefaultEventEngine().get())
520 ->CreatePosixEndpointFromFd(
521 sv[1],
522 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
523 grpc_core::ChannelArgs::FromC(&args)),
524 memory_quota->CreateMemoryAllocator("test"));
525 ep = grpc_event_engine::experimental::grpc_event_engine_endpoint_create(
526 std::move(eeep));
527 } else {
528 ep = grpc_tcp_create(
529 grpc_fd_create(sv[1], "read_test", false),
530 TcpOptionsFromEndpointConfig(
531 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
532 grpc_core::ChannelArgs::FromC(&args))),
533 "test");
534 GPR_ASSERT(grpc_tcp_fd(ep) == sv[1] && sv[1] >= 0);
535 }
536 grpc_endpoint_add_to_pollset(ep, g_pollset);
537
538 written_bytes = fill_socket_partial(sv[0], num_bytes);
539 gpr_log(GPR_INFO, "Wrote %" PRIuPTR " bytes", written_bytes);
540
541 state.ep = ep;
542 state.read_bytes = 0;
543 state.target_read_bytes = written_bytes;
544 state.min_progress_size = 1;
545 grpc_slice_buffer_init(&state.incoming);
546 GRPC_CLOSURE_INIT(&state.read_cb, read_cb, &state, grpc_schedule_on_exec_ctx);
547
548 grpc_endpoint_read(ep, &state.incoming, &state.read_cb, /*urgent=*/false,
549 /*min_progress_size=*/state.min_progress_size);
550 grpc_core::ExecCtx::Get()->Flush();
551 gpr_mu_lock(g_mu);
552 while (state.read_bytes < state.target_read_bytes) {
553 grpc_pollset_worker* worker = nullptr;
554 GPR_ASSERT(GRPC_LOG_IF_ERROR(
555 "pollset_work", grpc_pollset_work(g_pollset, &worker, deadline)));
556 gpr_log(GPR_DEBUG, "wakeup: read=%" PRIdPTR " target=%" PRIdPTR,
557 state.read_bytes, state.target_read_bytes);
558 gpr_mu_unlock(g_mu);
559 grpc_core::ExecCtx::Get()->Flush();
560 gpr_mu_lock(g_mu);
561 }
562 GPR_ASSERT(state.read_bytes == state.target_read_bytes);
563 gpr_mu_unlock(g_mu);
564
565 grpc_slice_buffer_destroy(&state.incoming);
566 grpc_tcp_destroy_and_release_fd(ep, &fd, &fd_released_cb);
567 grpc_core::ExecCtx::Get()->Flush();
568 rel_fd.notify.WaitForNotificationWithTimeout(absl::Seconds(20));
569 GPR_ASSERT(rel_fd.fd_released_done == 1);
570 GPR_ASSERT(fd == sv[1]);
571 written_bytes = fill_socket_partial(sv[0], num_bytes);
572 drain_socket_blocking(fd, written_bytes, written_bytes);
573 written_bytes = fill_socket_partial(fd, num_bytes);
574 drain_socket_blocking(sv[0], written_bytes, written_bytes);
575 close(fd);
576 grpc_resource_quota_unref(
577 static_cast<grpc_resource_quota*>(a[1].value.pointer.p));
578 }
579
run_tests(void)580 void run_tests(void) {
581 size_t i = 0;
582 for (int i = 1; i <= 8192; i = i * 2) {
583 read_test(100, 8192, i);
584 read_test(10000, 8192, i);
585 read_test(10000, 137, i);
586 read_test(10000, 1, i);
587 large_read_test(8192, i);
588 large_read_test(1, i);
589 }
590 write_test(100, 8192);
591 write_test(100, 1);
592 write_test(100000, 8192);
593 write_test(100000, 1);
594 write_test(100000, 137);
595
596 for (i = 1; i < 1000; i = std::max(i + 1, i * 5 / 4)) {
597 write_test(40320, i);
598 }
599
600 release_fd_test(100, 8192);
601 }
602
clean_up(void)603 static void clean_up(void) {}
604
create_fixture_tcp_socketpair(size_t slice_size)605 static grpc_endpoint_test_fixture create_fixture_tcp_socketpair(
606 size_t slice_size) {
607 int sv[2];
608 grpc_endpoint_test_fixture f;
609 grpc_core::ExecCtx exec_ctx;
610
611 create_sockets(sv);
612 grpc_arg a[2];
613 a[0].key = const_cast<char*>(GRPC_ARG_TCP_READ_CHUNK_SIZE);
614 a[0].type = GRPC_ARG_INTEGER;
615 a[0].value.integer = static_cast<int>(slice_size);
616 a[1].key = const_cast<char*>(GRPC_ARG_RESOURCE_QUOTA);
617 a[1].type = GRPC_ARG_POINTER;
618 a[1].value.pointer.p = grpc_resource_quota_create("test");
619 a[1].value.pointer.vtable = grpc_resource_quota_arg_vtable();
620 grpc_channel_args args = {GPR_ARRAY_SIZE(a), a};
621 f.client_ep = grpc_tcp_create(
622 grpc_fd_create(sv[0], "fixture:client", false),
623 TcpOptionsFromEndpointConfig(
624 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
625 grpc_core::ChannelArgs::FromC(&args))),
626 "test");
627 f.server_ep = grpc_tcp_create(
628 grpc_fd_create(sv[1], "fixture:server", false),
629 TcpOptionsFromEndpointConfig(
630 grpc_event_engine::experimental::ChannelArgsEndpointConfig(
631 grpc_core::ChannelArgs::FromC(&args))),
632 "test");
633 grpc_endpoint_add_to_pollset(f.client_ep, g_pollset);
634 grpc_endpoint_add_to_pollset(f.server_ep, g_pollset);
635 grpc_resource_quota_unref(
636 static_cast<grpc_resource_quota*>(a[1].value.pointer.p));
637
638 return f;
639 }
640
641 static grpc_endpoint_test_config configs[] = {
642 {"tcp/tcp_socketpair", create_fixture_tcp_socketpair, clean_up},
643 };
644
destroy_pollset(void * p,grpc_error_handle)645 static void destroy_pollset(void* p, grpc_error_handle /*error*/) {
646 grpc_pollset_destroy(static_cast<grpc_pollset*>(p));
647 }
648
main(int argc,char ** argv)649 int main(int argc, char** argv) {
650 grpc_closure destroyed;
651 ::testing::InitGoogleTest(&argc, argv);
652 grpc::testing::TestEnvironment env(&argc, argv);
653 grpc_init();
654 {
655 grpc_core::ExecCtx exec_ctx;
656 g_pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
657 grpc_pollset_init(g_pollset, &g_mu);
658 grpc_endpoint_tests(configs[0], g_pollset, g_mu);
659 run_tests();
660 GRPC_CLOSURE_INIT(&destroyed, destroy_pollset, g_pollset,
661 grpc_schedule_on_exec_ctx);
662 grpc_pollset_shutdown(g_pollset, &destroyed);
663
664 grpc_core::ExecCtx::Get()->Flush();
665 }
666 grpc_shutdown();
667 gpr_free(g_pollset);
668
669 return 0;
670 }
671
672 #else // GRPC_POSIX_SOCKET_TCP
673
main(int argc,char ** argv)674 int main(int argc, char** argv) { return 1; }
675
676 #endif // GRPC_POSIX_SOCKET_TCP
677