test/core/end2end/tests/retry_cancel_during_delay.cc

   1 //
   2 // Copyright 2017 gRPC authors.
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //     http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15 //
  16
  17 #include "test/core/end2end/end2end_tests.h"
  18
  19 #include <stdio.h>
  20 #include <string.h>
  21
  22 #include <string>
  23
  24 #include "absl/strings/str_cat.h"
  25
  26 #include <grpc/byte_buffer.h>
  27 #include <grpc/grpc.h>
  28 #include <grpc/support/alloc.h>
  29 #include <grpc/support/log.h>
  30 #include <grpc/support/time.h>
  31
  32 #include "src/core/lib/channel/channel_args.h"
  33 #include "src/core/lib/gpr/string.h"
  34 #include "src/core/lib/gpr/useful.h"
  35 #include "src/core/lib/iomgr/exec_ctx.h"
  36 #include "src/core/lib/transport/static_metadata.h"
  37
  38 #include "test/core/end2end/cq_verifier.h"
  39 #include "test/core/end2end/tests/cancel_test_helpers.h"
  40
  41 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
  42
  43 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
  44                                             const char* test_name,
  45                                             grpc_channel_args* client_args,
  46                                             grpc_channel_args* server_args) {
  47   grpc_end2end_test_fixture f;
  48   gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
  49   f = config.create_fixture(client_args, server_args);
  50   config.init_server(&f, server_args);
  51   config.init_client(&f, client_args);
  52   return f;
  53 }
  54
  55 static gpr_timespec n_seconds_from_now(int n) {
  56   return grpc_timeout_seconds_to_deadline(n);
  57 }
  58
  59 static gpr_timespec five_seconds_from_now(void) {
  60   return n_seconds_from_now(5);
  61 }
  62
  63 static void drain_cq(grpc_completion_queue* cq) {
  64   grpc_event ev;
  65   do {
  66     ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
  67   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
  68 }
  69
  70 static void shutdown_server(grpc_end2end_test_fixture* f) {
  71   if (!f->server) return;
  72   grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
  73   GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
  74                                          grpc_timeout_seconds_to_deadline(5),
  75                                          nullptr)
  76                  .type == GRPC_OP_COMPLETE);
  77   grpc_server_destroy(f->server);
  78   f->server = nullptr;
  79 }
  80
  81 static void shutdown_client(grpc_end2end_test_fixture* f) {
  82   if (!f->client) return;
  83   grpc_channel_destroy(f->client);
  84   f->client = nullptr;
  85 }
  86
  87 static void end_test(grpc_end2end_test_fixture* f) {
  88   shutdown_server(f);
  89   shutdown_client(f);
  90
  91   grpc_completion_queue_shutdown(f->cq);
  92   drain_cq(f->cq);
  93   grpc_completion_queue_destroy(f->cq);
  94   grpc_completion_queue_destroy(f->shutdown_cq);
  95 }
  96
  97 // Tests retry cancellation during backoff.
  98 static void test_retry_cancel_during_delay(grpc_end2end_test_config config,
  99                                            cancellation_mode mode) {
 100   grpc_call* c;
 101   grpc_call* s;
 102   grpc_op ops[6];
 103   grpc_op* op;
 104   grpc_metadata_array initial_metadata_recv;
 105   grpc_metadata_array trailing_metadata_recv;
 106   grpc_metadata_array request_metadata_recv;
 107   grpc_call_details call_details;
 108   grpc_slice request_payload_slice = grpc_slice_from_static_string("foo");
 109   grpc_slice response_payload_slice = grpc_slice_from_static_string("bar");
 110   grpc_byte_buffer* request_payload =
 111       grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 112   grpc_byte_buffer* response_payload =
 113       grpc_raw_byte_buffer_create(&response_payload_slice, 1);
 114   grpc_byte_buffer* request_payload_recv = nullptr;
 115   grpc_byte_buffer* response_payload_recv = nullptr;
 116   grpc_status_code status;
 117   grpc_call_error error;
 118   grpc_slice details;
 119   int was_cancelled = 2;
 120   char* peer;
 121
 122   grpc_arg args[] = {
 123       grpc_channel_arg_string_create(
 124           const_cast<char*>(GRPC_ARG_SERVICE_CONFIG),
 125           const_cast<char*>(
 126               "{\n"
 127               "  \"methodConfig\": [ {\n"
 128               "    \"name\": [\n"
 129               "      { \"service\": \"service\", \"method\": \"method\" }\n"
 130               "    ],\n"
 131               "    \"retryPolicy\": {\n"
 132               "      \"maxAttempts\": 3,\n"
 133               "      \"initialBackoff\": \"10s\",\n"
 134               "      \"maxBackoff\": \"120s\",\n"
 135               "      \"backoffMultiplier\": 1.6,\n"
 136               "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
 137               "    },\n"
 138               "    \"timeout\": \"5s\"\n"
 139               "  } ]\n"
 140               "}")),
 141   };
 142   grpc_channel_args client_args = {GPR_ARRAY_SIZE(args), args};
 143   std::string name = absl::StrCat("retry_cancel_during_delay/", mode.name);
 144   grpc_end2end_test_fixture f =
 145       begin_test(config, name.c_str(), &client_args, nullptr);
 146
 147   cq_verifier* cqv = cq_verifier_create(f.cq);
 148
 149   gpr_timespec expect_finish_before = n_seconds_from_now(10);
 150   gpr_timespec deadline = five_seconds_from_now();
 151   c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
 152                                grpc_slice_from_static_string("/service/method"),
 153                                nullptr, deadline, nullptr);
 154   GPR_ASSERT(c);
 155
 156   peer = grpc_call_get_peer(c);
 157   GPR_ASSERT(peer != nullptr);
 158   gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
 159   gpr_free(peer);
 160
 161   grpc_metadata_array_init(&initial_metadata_recv);
 162   grpc_metadata_array_init(&trailing_metadata_recv);
 163   grpc_metadata_array_init(&request_metadata_recv);
 164   grpc_call_details_init(&call_details);
 165   grpc_slice status_details = grpc_slice_from_static_string("xyz");
 166
 167   // Client starts a batch with all 6 ops.
 168   memset(ops, 0, sizeof(ops));
 169   op = ops;
 170   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 171   op->data.send_initial_metadata.count = 0;
 172   op++;
 173   op->op = GRPC_OP_SEND_MESSAGE;
 174   op->data.send_message.send_message = request_payload;
 175   op++;
 176   op->op = GRPC_OP_RECV_MESSAGE;
 177   op->data.recv_message.recv_message = &response_payload_recv;
 178   op++;
 179   op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 180   op++;
 181   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 182   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 183   op++;
 184   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 185   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 186   op->data.recv_status_on_client.status = &status;
 187   op->data.recv_status_on_client.status_details = &details;
 188   op++;
 189   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
 190                                 nullptr);
 191   GPR_ASSERT(GRPC_CALL_OK == error);
 192
 193   // Server gets a call and fails with retryable status.
 194   error =
 195       grpc_server_request_call(f.server, &s, &call_details,
 196                                &request_metadata_recv, f.cq, f.cq, tag(101));
 197   GPR_ASSERT(GRPC_CALL_OK == error);
 198   CQ_EXPECT_COMPLETION(cqv, tag(101), true);
 199   cq_verify(cqv);
 200
 201   peer = grpc_call_get_peer(s);
 202   GPR_ASSERT(peer != nullptr);
 203   gpr_log(GPR_DEBUG, "server_peer=%s", peer);
 204   gpr_free(peer);
 205   peer = grpc_call_get_peer(c);
 206   GPR_ASSERT(peer != nullptr);
 207   gpr_log(GPR_DEBUG, "client_peer=%s", peer);
 208   gpr_free(peer);
 209
 210   memset(ops, 0, sizeof(ops));
 211   op = ops;
 212   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 213   op->data.send_initial_metadata.count = 0;
 214   op++;
 215   op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
 216   op->data.send_status_from_server.trailing_metadata_count = 0;
 217   op->data.send_status_from_server.status = GRPC_STATUS_ABORTED;
 218   op->data.send_status_from_server.status_details = &status_details;
 219   op++;
 220   op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
 221   op->data.recv_close_on_server.cancelled = &was_cancelled;
 222   op++;
 223   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(102),
 224                                 nullptr);
 225   GPR_ASSERT(GRPC_CALL_OK == error);
 226
 227   CQ_EXPECT_COMPLETION(cqv, tag(102), true);
 228   cq_verify(cqv);
 229
 230   grpc_call_unref(s);
 231   grpc_metadata_array_destroy(&request_metadata_recv);
 232   grpc_metadata_array_init(&request_metadata_recv);
 233   grpc_call_details_destroy(&call_details);
 234   grpc_call_details_init(&call_details);
 235
 236   // Server should never get a second call, because the initial retry
 237   // delay is longer than the call's deadline.
 238   error =
 239       grpc_server_request_call(f.server, &s, &call_details,
 240                                &request_metadata_recv, f.cq, f.cq, tag(201));
 241   GPR_ASSERT(GRPC_CALL_OK == error);
 242   cq_verify_empty(cqv);
 243
 244   // Initiate cancellation.
 245   GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c, nullptr));
 246
 247   CQ_EXPECT_COMPLETION(cqv, tag(1), true);
 248   cq_verify(cqv);
 249
 250   // Make sure we didn't wait the full deadline before failing.
 251   gpr_log(
 252       GPR_INFO, "Expect completion before: %s",
 253       absl::FormatTime(grpc_core::ToAbslTime(expect_finish_before)).c_str());
 254   GPR_ASSERT(gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), expect_finish_before) <
 255              0);
 256
 257   gpr_log(GPR_INFO, "status=%d expected=%d", status, mode.expect_status);
 258   GPR_ASSERT(status == mode.expect_status);
 259   GPR_ASSERT(was_cancelled == 0);
 260
 261   grpc_slice_unref(details);
 262   grpc_metadata_array_destroy(&initial_metadata_recv);
 263   grpc_metadata_array_destroy(&trailing_metadata_recv);
 264   grpc_metadata_array_destroy(&request_metadata_recv);
 265   grpc_call_details_destroy(&call_details);
 266   grpc_byte_buffer_destroy(request_payload);
 267   grpc_byte_buffer_destroy(response_payload);
 268   grpc_byte_buffer_destroy(request_payload_recv);
 269   grpc_byte_buffer_destroy(response_payload_recv);
 270
 271   grpc_call_unref(c);
 272
 273   cq_verifier_destroy(cqv);
 274
 275   end_test(&f);
 276   config.tear_down_data(&f);
 277 }
 278
 279 void retry_cancel_during_delay(grpc_end2end_test_config config) {
 280   GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL);
 281   for (size_t i = 0; i < GPR_ARRAY_SIZE(cancellation_modes); ++i) {
 282     test_retry_cancel_during_delay(config, cancellation_modes[i]);
 283   }
 284 }
 285
 286 void retry_cancel_during_delay_pre_init(void) {}