1 /*
2 *
3 * Copyright 2019 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 * close_fd_test tests the behavior of grpc core when the transport gets
18 * disconnected.
19 * The test creates an http2 transport over a socket pair and closes the
20 * client or server file descriptor to simulate connection breakage while
21 * an RPC call is in progress.
22 *
23 */
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 "test/core/util/test_config.h"
30
31 #include <stdio.h>
32 #include <string.h>
33 #include <unistd.h>
34
35 #include <grpc/byte_buffer.h>
36 #include <grpc/byte_buffer_reader.h>
37 #include <grpc/grpc.h>
38 #include <grpc/support/alloc.h>
39 #include <grpc/support/log.h>
40 #include <grpc/support/time.h>
41 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
42 #include "src/core/lib/iomgr/endpoint_pair.h"
43 #include "src/core/lib/surface/channel.h"
44 #include "src/core/lib/surface/completion_queue.h"
45 #include "src/core/lib/surface/server.h"
46
tag(intptr_t t)47 static void* tag(intptr_t t) { return (void*)t; }
48
49 typedef struct test_ctx test_ctx;
50
51 struct test_ctx {
52 /* completion queue for call notifications on the server */
53 grpc_completion_queue* cq;
54 /* completion queue registered to server for shutdown events */
55 grpc_completion_queue* shutdown_cq;
56 /* client's completion queue */
57 grpc_completion_queue* client_cq;
58 /* completion queue bound to call on the server */
59 grpc_completion_queue* bound_cq;
60 /* Server responds to client calls */
61 grpc_server* server;
62 /* Client calls are sent over the channel */
63 grpc_channel* client;
64 /* encapsulates client, server endpoints */
65 grpc_endpoint_pair* ep;
66 };
67
68 static test_ctx g_ctx;
69
70 /* chttp2 transport that is immediately available (used for testing
71 connected_channel without a client_channel */
72
server_setup_transport(grpc_transport * transport)73 static void server_setup_transport(grpc_transport* transport) {
74 grpc_core::ExecCtx exec_ctx;
75 grpc_endpoint_add_to_pollset(g_ctx.ep->server, grpc_cq_pollset(g_ctx.cq));
76 g_ctx.server->core_server->SetupTransport(
77 transport, nullptr, g_ctx.server->core_server->channel_args(), nullptr);
78 }
79
client_setup_transport(grpc_transport * transport)80 static void client_setup_transport(grpc_transport* transport) {
81 grpc_core::ExecCtx exec_ctx;
82 grpc_endpoint_add_to_pollset(g_ctx.ep->client,
83 grpc_cq_pollset(g_ctx.client_cq));
84 grpc_arg authority_arg = grpc_channel_arg_string_create(
85 const_cast<char*>(GRPC_ARG_DEFAULT_AUTHORITY),
86 const_cast<char*>("test-authority"));
87 grpc_channel_args* args =
88 grpc_channel_args_copy_and_add(nullptr, &authority_arg, 1);
89 /* TODO (pjaikumar): use GRPC_CLIENT_CHANNEL instead of
90 * GRPC_CLIENT_DIRECT_CHANNEL */
91 g_ctx.client = grpc_channel_create("socketpair-target", args,
92 GRPC_CLIENT_DIRECT_CHANNEL, transport);
93 grpc_channel_args_destroy(args);
94 }
95
init_client()96 static void init_client() {
97 grpc_core::ExecCtx exec_ctx;
98 grpc_transport* transport;
99 transport = grpc_create_chttp2_transport(nullptr, g_ctx.ep->client, true);
100 client_setup_transport(transport);
101 GPR_ASSERT(g_ctx.client);
102 grpc_chttp2_transport_start_reading(transport, nullptr, nullptr);
103 }
104
init_server()105 static void init_server() {
106 grpc_core::ExecCtx exec_ctx;
107 grpc_transport* transport;
108 GPR_ASSERT(!g_ctx.server);
109 g_ctx.server = grpc_server_create(nullptr, nullptr);
110 grpc_server_register_completion_queue(g_ctx.server, g_ctx.cq, nullptr);
111 grpc_server_start(g_ctx.server);
112 transport = grpc_create_chttp2_transport(nullptr, g_ctx.ep->server, false);
113 server_setup_transport(transport);
114 grpc_chttp2_transport_start_reading(transport, nullptr, nullptr);
115 }
116
test_init()117 static void test_init() {
118 grpc_endpoint_pair* sfd =
119 static_cast<grpc_endpoint_pair*>(gpr_malloc(sizeof(grpc_endpoint_pair)));
120 memset(&g_ctx, 0, sizeof(g_ctx));
121 g_ctx.ep = sfd;
122 g_ctx.cq = grpc_completion_queue_create_for_next(nullptr);
123 g_ctx.shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
124 g_ctx.bound_cq = grpc_completion_queue_create_for_next(nullptr);
125 g_ctx.client_cq = grpc_completion_queue_create_for_next(nullptr);
126
127 /* Create endpoints */
128 *sfd = grpc_iomgr_create_endpoint_pair("fixture", nullptr);
129 /* Create client, server and setup transport over endpoint pair */
130 init_server();
131 init_client();
132 }
133
drain_cq(grpc_completion_queue * cq)134 static void drain_cq(grpc_completion_queue* cq) {
135 grpc_event event;
136 do {
137 event = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(1),
138 nullptr);
139 } while (event.type != GRPC_QUEUE_SHUTDOWN);
140 }
141
drain_and_destroy_cq(grpc_completion_queue * cq)142 static void drain_and_destroy_cq(grpc_completion_queue* cq) {
143 grpc_completion_queue_shutdown(cq);
144 drain_cq(cq);
145 grpc_completion_queue_destroy(cq);
146 }
147
shutdown_server()148 static void shutdown_server() {
149 if (!g_ctx.server) return;
150 grpc_server_shutdown_and_notify(g_ctx.server, g_ctx.shutdown_cq, tag(1000));
151 GPR_ASSERT(grpc_completion_queue_pluck(g_ctx.shutdown_cq, tag(1000),
152 grpc_timeout_seconds_to_deadline(1),
153 nullptr)
154 .type == GRPC_OP_COMPLETE);
155 grpc_server_destroy(g_ctx.server);
156 g_ctx.server = nullptr;
157 }
158
shutdown_client()159 static void shutdown_client() {
160 if (!g_ctx.client) return;
161 grpc_channel_destroy(g_ctx.client);
162 g_ctx.client = nullptr;
163 }
164
end_test()165 static void end_test() {
166 shutdown_server();
167 shutdown_client();
168
169 drain_and_destroy_cq(g_ctx.cq);
170 drain_and_destroy_cq(g_ctx.client_cq);
171 drain_and_destroy_cq(g_ctx.bound_cq);
172 grpc_completion_queue_destroy(g_ctx.shutdown_cq);
173 gpr_free(g_ctx.ep);
174 }
175
176 typedef enum fd_type { CLIENT_FD, SERVER_FD } fd_type;
177
fd_type_str(fd_type fdtype)178 static const char* fd_type_str(fd_type fdtype) {
179 if (fdtype == CLIENT_FD) {
180 return "client";
181 } else if (fdtype == SERVER_FD) {
182 return "server";
183 } else {
184 gpr_log(GPR_ERROR, "Unexpected fd_type %d", fdtype);
185 abort();
186 }
187 }
188
_test_close_before_server_recv(fd_type fdtype)189 static void _test_close_before_server_recv(fd_type fdtype) {
190 grpc_core::ExecCtx exec_ctx;
191 grpc_call* call;
192 grpc_call* server_call;
193 grpc_event event;
194 grpc_slice request_payload_slice =
195 grpc_slice_from_copied_string("hello world");
196 grpc_slice response_payload_slice =
197 grpc_slice_from_copied_string("hello you");
198 grpc_byte_buffer* request_payload =
199 grpc_raw_byte_buffer_create(&request_payload_slice, 1);
200 grpc_byte_buffer* response_payload =
201 grpc_raw_byte_buffer_create(&response_payload_slice, 1);
202 gpr_log(GPR_INFO, "Running test: test_close_%s_before_server_recv",
203 fd_type_str(fdtype));
204 test_init();
205
206 grpc_op ops[6];
207 grpc_op* op;
208 grpc_metadata_array initial_metadata_recv;
209 grpc_metadata_array trailing_metadata_recv;
210 grpc_metadata_array request_metadata_recv;
211 grpc_byte_buffer* request_payload_recv = nullptr;
212 grpc_byte_buffer* response_payload_recv = nullptr;
213 grpc_call_details call_details;
214 grpc_status_code status = GRPC_STATUS__DO_NOT_USE;
215 grpc_call_error error;
216 grpc_slice details;
217
218 gpr_timespec deadline = grpc_timeout_seconds_to_deadline(1);
219 call = grpc_channel_create_call(
220 g_ctx.client, nullptr, GRPC_PROPAGATE_DEFAULTS, g_ctx.client_cq,
221 grpc_slice_from_static_string("/foo"), nullptr, deadline, nullptr);
222 GPR_ASSERT(call);
223
224 grpc_metadata_array_init(&initial_metadata_recv);
225 grpc_metadata_array_init(&trailing_metadata_recv);
226 grpc_metadata_array_init(&request_metadata_recv);
227 grpc_call_details_init(&call_details);
228
229 memset(ops, 0, sizeof(ops));
230 op = ops;
231 op->op = GRPC_OP_SEND_INITIAL_METADATA;
232 op->data.send_initial_metadata.count = 0;
233 op->flags = 0;
234 op->reserved = nullptr;
235 op++;
236 op->op = GRPC_OP_SEND_MESSAGE;
237 op->data.send_message.send_message = request_payload;
238 op->flags = 0;
239 op->reserved = nullptr;
240 op++;
241 op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
242 op->flags = 0;
243 op->reserved = nullptr;
244 op++;
245 op->op = GRPC_OP_RECV_INITIAL_METADATA;
246 op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
247 op->flags = 0;
248 op->reserved = nullptr;
249 op++;
250 op->op = GRPC_OP_RECV_MESSAGE;
251 op->data.recv_message.recv_message = &response_payload_recv;
252 op->flags = 0;
253 op->reserved = nullptr;
254 op++;
255 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
256 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
257 op->data.recv_status_on_client.status = &status;
258 op->data.recv_status_on_client.status_details = &details;
259 op->flags = 0;
260 op->reserved = nullptr;
261 op++;
262 error = grpc_call_start_batch(call, ops, static_cast<size_t>(op - ops),
263 tag(1), nullptr);
264 GPR_ASSERT(GRPC_CALL_OK == error);
265
266 error = grpc_server_request_call(g_ctx.server, &server_call, &call_details,
267 &request_metadata_recv, g_ctx.bound_cq,
268 g_ctx.cq, tag(101));
269 GPR_ASSERT(GRPC_CALL_OK == error);
270 event = grpc_completion_queue_next(
271 g_ctx.cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
272 GPR_ASSERT(event.success == 1);
273 GPR_ASSERT(event.tag == tag(101));
274 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
275
276 memset(ops, 0, sizeof(ops));
277 op = ops;
278 op->op = GRPC_OP_SEND_INITIAL_METADATA;
279 op->data.send_initial_metadata.count = 0;
280 op->flags = 0;
281 op->reserved = nullptr;
282 op++;
283 op->op = GRPC_OP_RECV_MESSAGE;
284 op->data.recv_message.recv_message = &request_payload_recv;
285 op->flags = 0;
286 op->reserved = nullptr;
287 op++;
288
289 grpc_endpoint_pair* sfd = g_ctx.ep;
290 int fd;
291 if (fdtype == SERVER_FD) {
292 fd = sfd->server->vtable->get_fd(sfd->server);
293 } else {
294 GPR_ASSERT(fdtype == CLIENT_FD);
295 fd = sfd->client->vtable->get_fd(sfd->client);
296 }
297 /* Connection is closed before the server receives the client's message. */
298 close(fd);
299
300 error = grpc_call_start_batch(server_call, ops, static_cast<size_t>(op - ops),
301 tag(102), nullptr);
302 GPR_ASSERT(GRPC_CALL_OK == error);
303
304 event = grpc_completion_queue_next(
305 g_ctx.bound_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
306
307 /* Batch operation completes on the server side.
308 * event.success will be true if the op completes successfully.
309 * event.success will be false if the op completes with an error. This can
310 * happen due to a race with closing the fd resulting in pending writes
311 * failing due to stream closure.
312 * */
313 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
314 GPR_ASSERT(event.tag == tag(102));
315
316 event = grpc_completion_queue_next(
317 g_ctx.client_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
318 /* When the client fd is closed, the server gets EPIPE.
319 * When server fd is closed, server gets EBADF.
320 * In both cases server sends GRPC_STATUS_UNAVALABLE to the client. However,
321 * the client may not receive this grpc_status as it's socket is being closed.
322 * If the client didn't get grpc_status from the server it will time out
323 * waiting on the completion queue. So there 2 2 possibilities:
324 * 1. client times out waiting for server's response
325 * 2. client receives GRPC_STATUS_UNAVAILABLE from server
326 */
327 if (event.type == GRPC_QUEUE_TIMEOUT) {
328 GPR_ASSERT(event.success == 0);
329 /* status is not initialized */
330 GPR_ASSERT(status == GRPC_STATUS__DO_NOT_USE);
331 } else {
332 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
333 GPR_ASSERT(event.success == 1);
334 GPR_ASSERT(event.tag == tag(1));
335 GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
336 }
337
338 grpc_metadata_array_destroy(&initial_metadata_recv);
339 grpc_metadata_array_destroy(&trailing_metadata_recv);
340 grpc_metadata_array_destroy(&request_metadata_recv);
341 grpc_call_details_destroy(&call_details);
342
343 grpc_call_unref(call);
344 grpc_call_unref(server_call);
345
346 grpc_byte_buffer_destroy(request_payload);
347 grpc_byte_buffer_destroy(response_payload);
348 grpc_byte_buffer_destroy(request_payload_recv);
349 grpc_byte_buffer_destroy(response_payload_recv);
350
351 end_test();
352 }
353
test_close_before_server_recv()354 static void test_close_before_server_recv() {
355 /* Close client side of the connection before server receives message from
356 * client */
357 _test_close_before_server_recv(CLIENT_FD);
358 /* Close server side of the connection before server receives message from
359 * client */
360 _test_close_before_server_recv(SERVER_FD);
361 }
362
_test_close_before_server_send(fd_type fdtype)363 static void _test_close_before_server_send(fd_type fdtype) {
364 grpc_core::ExecCtx exec_ctx;
365 grpc_call* call;
366 grpc_call* server_call;
367 grpc_event event;
368 grpc_slice request_payload_slice =
369 grpc_slice_from_copied_string("hello world");
370 grpc_slice response_payload_slice =
371 grpc_slice_from_copied_string("hello you");
372 grpc_byte_buffer* request_payload =
373 grpc_raw_byte_buffer_create(&request_payload_slice, 1);
374 grpc_byte_buffer* response_payload =
375 grpc_raw_byte_buffer_create(&response_payload_slice, 1);
376 gpr_log(GPR_INFO, "Running test: test_close_%s_before_server_send",
377 fd_type_str(fdtype));
378 test_init();
379
380 grpc_op ops[6];
381 grpc_op* op;
382 grpc_metadata_array initial_metadata_recv;
383 grpc_metadata_array trailing_metadata_recv;
384 grpc_metadata_array request_metadata_recv;
385 grpc_byte_buffer* request_payload_recv = nullptr;
386 grpc_byte_buffer* response_payload_recv = nullptr;
387 grpc_call_details call_details;
388 grpc_status_code status = GRPC_STATUS__DO_NOT_USE;
389 grpc_call_error error;
390 grpc_slice details;
391 int was_cancelled = 2;
392
393 gpr_timespec deadline = grpc_timeout_seconds_to_deadline(1);
394 call = grpc_channel_create_call(
395 g_ctx.client, nullptr, GRPC_PROPAGATE_DEFAULTS, g_ctx.client_cq,
396 grpc_slice_from_static_string("/foo"), nullptr, deadline, nullptr);
397 GPR_ASSERT(call);
398
399 grpc_metadata_array_init(&initial_metadata_recv);
400 grpc_metadata_array_init(&trailing_metadata_recv);
401 grpc_metadata_array_init(&request_metadata_recv);
402 grpc_call_details_init(&call_details);
403
404 memset(ops, 0, sizeof(ops));
405 op = ops;
406 op->op = GRPC_OP_SEND_INITIAL_METADATA;
407 op->data.send_initial_metadata.count = 0;
408 op->flags = 0;
409 op->reserved = nullptr;
410 op++;
411 op->op = GRPC_OP_SEND_MESSAGE;
412 op->data.send_message.send_message = request_payload;
413 op->flags = 0;
414 op->reserved = nullptr;
415 op++;
416 op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
417 op->flags = 0;
418 op->reserved = nullptr;
419 op++;
420 op->op = GRPC_OP_RECV_INITIAL_METADATA;
421 op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
422 op->flags = 0;
423 op->reserved = nullptr;
424 op++;
425 op->op = GRPC_OP_RECV_MESSAGE;
426 op->data.recv_message.recv_message = &response_payload_recv;
427 op->flags = 0;
428 op->reserved = nullptr;
429 op++;
430 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
431 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
432 op->data.recv_status_on_client.status = &status;
433 op->data.recv_status_on_client.status_details = &details;
434 op->flags = 0;
435 op->reserved = nullptr;
436 op++;
437 error = grpc_call_start_batch(call, ops, static_cast<size_t>(op - ops),
438 tag(1), nullptr);
439 GPR_ASSERT(GRPC_CALL_OK == error);
440
441 error = grpc_server_request_call(g_ctx.server, &server_call, &call_details,
442 &request_metadata_recv, g_ctx.bound_cq,
443 g_ctx.cq, tag(101));
444 GPR_ASSERT(GRPC_CALL_OK == error);
445 event = grpc_completion_queue_next(
446 g_ctx.cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
447 GPR_ASSERT(event.success == 1);
448 GPR_ASSERT(event.tag == tag(101));
449 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
450
451 memset(ops, 0, sizeof(ops));
452 op = ops;
453 op->op = GRPC_OP_SEND_INITIAL_METADATA;
454 op->data.send_initial_metadata.count = 0;
455 op->flags = 0;
456 op->reserved = nullptr;
457 op++;
458 op->op = GRPC_OP_RECV_MESSAGE;
459 op->data.recv_message.recv_message = &request_payload_recv;
460 op->flags = 0;
461 op->reserved = nullptr;
462 op++;
463 error = grpc_call_start_batch(server_call, ops, static_cast<size_t>(op - ops),
464 tag(102), nullptr);
465 GPR_ASSERT(GRPC_CALL_OK == error);
466
467 event = grpc_completion_queue_next(
468 g_ctx.bound_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
469 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
470 GPR_ASSERT(event.success == 1);
471 GPR_ASSERT(event.tag == tag(102));
472
473 memset(ops, 0, sizeof(ops));
474 op = ops;
475 op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
476 op->data.recv_close_on_server.cancelled = &was_cancelled;
477 op->flags = 0;
478 op->reserved = nullptr;
479 op++;
480 op->op = GRPC_OP_SEND_MESSAGE;
481 op->data.send_message.send_message = response_payload;
482 op->flags = 0;
483 op->reserved = nullptr;
484 op++;
485 op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
486 op->data.send_status_from_server.trailing_metadata_count = 0;
487 op->data.send_status_from_server.status = GRPC_STATUS_OK;
488 grpc_slice status_details = grpc_slice_from_static_string("xyz");
489 op->data.send_status_from_server.status_details = &status_details;
490 op->flags = 0;
491 op->reserved = nullptr;
492 op++;
493
494 grpc_endpoint_pair* sfd = g_ctx.ep;
495 int fd;
496 if (fdtype == SERVER_FD) {
497 fd = sfd->server->vtable->get_fd(sfd->server);
498 } else {
499 GPR_ASSERT(fdtype == CLIENT_FD);
500 fd = sfd->client->vtable->get_fd(sfd->client);
501 }
502
503 /* Connection is closed before the server sends message and status to the
504 * client. */
505 close(fd);
506 error = grpc_call_start_batch(server_call, ops, static_cast<size_t>(op - ops),
507 tag(103), nullptr);
508 GPR_ASSERT(GRPC_CALL_OK == error);
509
510 /* Batch operation succeeds on the server side */
511 event = grpc_completion_queue_next(
512 g_ctx.bound_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
513 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
514 GPR_ASSERT(event.success == 1);
515 GPR_ASSERT(event.tag == tag(103));
516
517 event = grpc_completion_queue_next(
518 g_ctx.client_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
519 /* In both cases server sends GRPC_STATUS_UNAVALABLE to the client. However,
520 * the client may not receive this grpc_status as it's socket is being closed.
521 * If the client didn't get grpc_status from the server it will time out
522 * waiting on the completion queue
523 */
524 if (event.type == GRPC_OP_COMPLETE) {
525 GPR_ASSERT(event.success == 1);
526 GPR_ASSERT(event.tag == tag(1));
527 GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
528 } else {
529 GPR_ASSERT(event.type == GRPC_QUEUE_TIMEOUT);
530 GPR_ASSERT(event.success == 0);
531 /* status is not initialized */
532 GPR_ASSERT(status == GRPC_STATUS__DO_NOT_USE);
533 }
534 GPR_ASSERT(was_cancelled == 0);
535
536 grpc_metadata_array_destroy(&initial_metadata_recv);
537 grpc_metadata_array_destroy(&trailing_metadata_recv);
538 grpc_metadata_array_destroy(&request_metadata_recv);
539 grpc_call_details_destroy(&call_details);
540
541 grpc_call_unref(call);
542 grpc_call_unref(server_call);
543
544 grpc_byte_buffer_destroy(request_payload);
545 grpc_byte_buffer_destroy(response_payload);
546 grpc_byte_buffer_destroy(request_payload_recv);
547 grpc_byte_buffer_destroy(response_payload_recv);
548
549 end_test();
550 }
551
test_close_before_server_send()552 static void test_close_before_server_send() {
553 /* Close client side of the connection before server sends message to client
554 * */
555 _test_close_before_server_send(CLIENT_FD);
556 /* Close server side of the connection before server sends message to client
557 * */
558 _test_close_before_server_send(SERVER_FD);
559 }
560
_test_close_before_client_send(fd_type fdtype)561 static void _test_close_before_client_send(fd_type fdtype) {
562 grpc_core::ExecCtx exec_ctx;
563 grpc_call* call;
564 grpc_event event;
565 grpc_slice request_payload_slice =
566 grpc_slice_from_copied_string("hello world");
567 grpc_slice response_payload_slice =
568 grpc_slice_from_copied_string("hello you");
569 grpc_byte_buffer* request_payload =
570 grpc_raw_byte_buffer_create(&request_payload_slice, 1);
571 grpc_byte_buffer* response_payload =
572 grpc_raw_byte_buffer_create(&response_payload_slice, 1);
573 gpr_log(GPR_INFO, "Running test: test_close_%s_before_client_send",
574 fd_type_str(fdtype));
575 test_init();
576
577 grpc_op ops[6];
578 grpc_op* op;
579 grpc_metadata_array initial_metadata_recv;
580 grpc_metadata_array trailing_metadata_recv;
581 grpc_metadata_array request_metadata_recv;
582 grpc_byte_buffer* request_payload_recv = nullptr;
583 grpc_byte_buffer* response_payload_recv = nullptr;
584 grpc_call_details call_details;
585 grpc_status_code status;
586 grpc_call_error error;
587 grpc_slice details;
588
589 gpr_timespec deadline = grpc_timeout_seconds_to_deadline(1);
590 call = grpc_channel_create_call(
591 g_ctx.client, nullptr, GRPC_PROPAGATE_DEFAULTS, g_ctx.client_cq,
592 grpc_slice_from_static_string("/foo"), nullptr, deadline, nullptr);
593 GPR_ASSERT(call);
594
595 grpc_metadata_array_init(&initial_metadata_recv);
596 grpc_metadata_array_init(&trailing_metadata_recv);
597 grpc_metadata_array_init(&request_metadata_recv);
598 grpc_call_details_init(&call_details);
599
600 memset(ops, 0, sizeof(ops));
601 op = ops;
602 op->op = GRPC_OP_SEND_INITIAL_METADATA;
603 op->data.send_initial_metadata.count = 0;
604 op->flags = 0;
605 op->reserved = nullptr;
606 op++;
607 op->op = GRPC_OP_SEND_MESSAGE;
608 op->data.send_message.send_message = request_payload;
609 op->flags = 0;
610 op->reserved = nullptr;
611 op++;
612 op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
613 op->flags = 0;
614 op->reserved = nullptr;
615 op++;
616 op->op = GRPC_OP_RECV_INITIAL_METADATA;
617 op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
618 op->flags = 0;
619 op->reserved = nullptr;
620 op++;
621 op->op = GRPC_OP_RECV_MESSAGE;
622 op->data.recv_message.recv_message = &response_payload_recv;
623 op->flags = 0;
624 op->reserved = nullptr;
625 op++;
626 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
627 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
628 op->data.recv_status_on_client.status = &status;
629 op->data.recv_status_on_client.status_details = &details;
630 op->flags = 0;
631 op->reserved = nullptr;
632 op++;
633
634 grpc_endpoint_pair* sfd = g_ctx.ep;
635 int fd;
636 if (fdtype == SERVER_FD) {
637 fd = sfd->server->vtable->get_fd(sfd->server);
638 } else {
639 GPR_ASSERT(fdtype == CLIENT_FD);
640 fd = sfd->client->vtable->get_fd(sfd->client);
641 }
642 /* Connection is closed before the client sends a batch to the server */
643 close(fd);
644
645 error = grpc_call_start_batch(call, ops, static_cast<size_t>(op - ops),
646 tag(1), nullptr);
647 GPR_ASSERT(GRPC_CALL_OK == error);
648
649 /* Status unavailable is returned to the client when client or server fd is
650 * closed */
651 event = grpc_completion_queue_next(
652 g_ctx.client_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
653 GPR_ASSERT(event.success == 1);
654 GPR_ASSERT(event.type == GRPC_OP_COMPLETE);
655 GPR_ASSERT(event.tag == tag(1));
656 GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
657
658 /* No event is received on the server */
659 event = grpc_completion_queue_next(
660 g_ctx.cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
661 GPR_ASSERT(event.success == 0);
662 GPR_ASSERT(event.type == GRPC_QUEUE_TIMEOUT);
663
664 grpc_slice_unref(details);
665 grpc_metadata_array_destroy(&initial_metadata_recv);
666 grpc_metadata_array_destroy(&trailing_metadata_recv);
667 grpc_metadata_array_destroy(&request_metadata_recv);
668 grpc_call_details_destroy(&call_details);
669
670 grpc_call_unref(call);
671
672 grpc_byte_buffer_destroy(request_payload);
673 grpc_byte_buffer_destroy(response_payload);
674 grpc_byte_buffer_destroy(request_payload_recv);
675 grpc_byte_buffer_destroy(response_payload_recv);
676
677 end_test();
678 }
test_close_before_client_send()679 static void test_close_before_client_send() {
680 /* Close client side of the connection before client sends message to server
681 * */
682 _test_close_before_client_send(CLIENT_FD);
683 /* Close server side of the connection before client sends message to server
684 * */
685 _test_close_before_client_send(SERVER_FD);
686 }
687
_test_close_before_call_create(fd_type fdtype)688 static void _test_close_before_call_create(fd_type fdtype) {
689 grpc_core::ExecCtx exec_ctx;
690 grpc_call* call;
691 grpc_event event;
692 test_init();
693
694 gpr_timespec deadline = grpc_timeout_milliseconds_to_deadline(100);
695
696 grpc_endpoint_pair* sfd = g_ctx.ep;
697 int fd;
698 if (fdtype == SERVER_FD) {
699 fd = sfd->server->vtable->get_fd(sfd->server);
700 } else {
701 GPR_ASSERT(fdtype == CLIENT_FD);
702 fd = sfd->client->vtable->get_fd(sfd->client);
703 }
704 /* Connection is closed before the client creates a call */
705 close(fd);
706
707 call = grpc_channel_create_call(
708 g_ctx.client, nullptr, GRPC_PROPAGATE_DEFAULTS, g_ctx.client_cq,
709 grpc_slice_from_static_string("/foo"), nullptr, deadline, nullptr);
710 GPR_ASSERT(call);
711
712 /* Client and server time out waiting on their completion queues and nothing
713 * is sent or received */
714 event = grpc_completion_queue_next(
715 g_ctx.client_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
716 GPR_ASSERT(event.type == GRPC_QUEUE_TIMEOUT);
717 GPR_ASSERT(event.success == 0);
718
719 event = grpc_completion_queue_next(
720 g_ctx.cq, grpc_timeout_milliseconds_to_deadline(100), nullptr);
721 GPR_ASSERT(event.type == GRPC_QUEUE_TIMEOUT);
722 GPR_ASSERT(event.success == 0);
723
724 grpc_call_unref(call);
725 end_test();
726 }
727
test_close_before_call_create()728 static void test_close_before_call_create() {
729 /* Close client side of the connection before client creates a call */
730 _test_close_before_call_create(CLIENT_FD);
731 /* Close server side of the connection before client creates a call */
732 _test_close_before_call_create(SERVER_FD);
733 }
734
main(int argc,char ** argv)735 int main(int argc, char** argv) {
736 grpc::testing::TestEnvironment env(argc, argv);
737 /* Init grpc */
738 grpc_init();
739 int iterations = 10;
740
741 for (int i = 0; i < iterations; ++i) {
742 test_close_before_call_create();
743 test_close_before_client_send();
744 test_close_before_server_recv();
745 test_close_before_server_send();
746 }
747
748 grpc_shutdown();
749
750 return 0;
751 }
752
753 #else /* GRPC_POSIX_SOCKET_TCP */
754
main(int argc,char ** argv)755 int main(int argc, char** argv) { return 1; }
756
757 #endif /* GRPC_POSIX_SOCKET_TCP */
758