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