test/core/end2end/tests/payload.cc

   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 <stdio.h>
  20 #include <string.h>
  21
  22 #include <grpc/byte_buffer.h>
  23 #include <grpc/support/alloc.h>
  24 #include <grpc/support/log.h>
  25 #include <grpc/support/time.h>
  26
  27 #include "test/core/end2end/cq_verifier.h"
  28 #include "test/core/end2end/end2end_tests.h"
  29
  30 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
  31
  32 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
  33                                             const char* test_name,
  34                                             grpc_channel_args* client_args,
  35                                             grpc_channel_args* server_args) {
  36   grpc_end2end_test_fixture f;
  37   gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
  38   f = config.create_fixture(client_args, server_args);
  39   config.init_server(&f, server_args);
  40   config.init_client(&f, client_args);
  41   return f;
  42 }
  43
  44 static gpr_timespec n_seconds_from_now(int n) {
  45   return grpc_timeout_seconds_to_deadline(n);
  46 }
  47
  48 static gpr_timespec five_seconds_from_now(void) {
  49   return n_seconds_from_now(5);
  50 }
  51
  52 static void drain_cq(grpc_completion_queue* cq) {
  53   grpc_event ev;
  54   do {
  55     ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
  56   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
  57 }
  58
  59 static void shutdown_server(grpc_end2end_test_fixture* f) {
  60   if (!f->server) return;
  61   grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
  62   GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
  63                                          grpc_timeout_seconds_to_deadline(5),
  64                                          nullptr)
  65                  .type == GRPC_OP_COMPLETE);
  66   grpc_server_destroy(f->server);
  67   f->server = nullptr;
  68 }
  69
  70 static void shutdown_client(grpc_end2end_test_fixture* f) {
  71   if (!f->client) return;
  72   grpc_channel_destroy(f->client);
  73   f->client = nullptr;
  74 }
  75
  76 static void end_test(grpc_end2end_test_fixture* f) {
  77   shutdown_server(f);
  78   shutdown_client(f);
  79
  80   grpc_completion_queue_shutdown(f->cq);
  81   drain_cq(f->cq);
  82   grpc_completion_queue_destroy(f->cq);
  83   grpc_completion_queue_destroy(f->shutdown_cq);
  84 }
  85
  86 /* Creates and returns a grpc_slice containing random alphanumeric characters.
  87  */
  88 static grpc_slice generate_random_slice() {
  89   size_t i;
  90   static const char chars[] = "abcdefghijklmnopqrstuvwxyz1234567890";
  91   char* output;
  92   const size_t output_size = 1024 * 1024;
  93   output = static_cast<char*>(gpr_malloc(output_size));
  94   for (i = 0; i < output_size - 1; ++i) {
  95     output[i] = chars[rand() % static_cast<int>(sizeof(chars) - 1)];
  96   }
  97   output[output_size - 1] = '\0';
  98   grpc_slice out = grpc_slice_from_copied_string(output);
  99   gpr_free(output);
 100   return out;
 101 }
 102
 103 static void request_response_with_payload(grpc_end2end_test_config /*config*/,
 104                                           grpc_end2end_test_fixture f) {
 105   /* Create large request and response bodies. These are big enough to require
 106    * multiple round trips to deliver to the peer, and their exact contents of
 107    * will be verified on completion. */
 108   grpc_slice request_payload_slice = generate_random_slice();
 109   grpc_slice response_payload_slice = generate_random_slice();
 110
 111   grpc_call* c;
 112   grpc_call* s;
 113   grpc_byte_buffer* request_payload =
 114       grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 115   grpc_byte_buffer* response_payload =
 116       grpc_raw_byte_buffer_create(&response_payload_slice, 1);
 117   cq_verifier* cqv = cq_verifier_create(f.cq);
 118   grpc_op ops[6];
 119   grpc_op* op;
 120   grpc_metadata_array initial_metadata_recv;
 121   grpc_metadata_array trailing_metadata_recv;
 122   grpc_metadata_array request_metadata_recv;
 123   grpc_byte_buffer* request_payload_recv = nullptr;
 124   grpc_byte_buffer* response_payload_recv = nullptr;
 125   grpc_call_details call_details;
 126   grpc_status_code status;
 127   grpc_call_error error;
 128   grpc_slice details;
 129   int was_cancelled = 2;
 130
 131   gpr_timespec deadline = n_seconds_from_now(60);
 132   c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
 133                                grpc_slice_from_static_string("/foo"), nullptr,
 134                                deadline, nullptr);
 135   GPR_ASSERT(c);
 136
 137   grpc_metadata_array_init(&initial_metadata_recv);
 138   grpc_metadata_array_init(&trailing_metadata_recv);
 139   grpc_metadata_array_init(&request_metadata_recv);
 140   grpc_call_details_init(&call_details);
 141
 142   memset(ops, 0, sizeof(ops));
 143   op = ops;
 144   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 145   op->data.send_initial_metadata.count = 0;
 146   op->flags = 0;
 147   op->reserved = nullptr;
 148   op++;
 149   op->op = GRPC_OP_SEND_MESSAGE;
 150   op->data.send_message.send_message = request_payload;
 151   op->flags = 0;
 152   op->reserved = nullptr;
 153   op++;
 154   op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 155   op->flags = 0;
 156   op->reserved = nullptr;
 157   op++;
 158   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 159   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 160   op->flags = 0;
 161   op->reserved = nullptr;
 162   op++;
 163   op->op = GRPC_OP_RECV_MESSAGE;
 164   op->data.recv_message.recv_message = &response_payload_recv;
 165   op->flags = 0;
 166   op->reserved = nullptr;
 167   op++;
 168   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 169   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 170   op->data.recv_status_on_client.status = &status;
 171   op->data.recv_status_on_client.status_details = &details;
 172   op->flags = 0;
 173   op->reserved = nullptr;
 174   op++;
 175   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
 176                                 nullptr);
 177   GPR_ASSERT(GRPC_CALL_OK == error);
 178
 179   error =
 180       grpc_server_request_call(f.server, &s, &call_details,
 181                                &request_metadata_recv, f.cq, f.cq, tag(101));
 182   GPR_ASSERT(GRPC_CALL_OK == error);
 183   CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
 184   cq_verify(cqv);
 185
 186   memset(ops, 0, sizeof(ops));
 187   op = ops;
 188   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 189   op->data.send_initial_metadata.count = 0;
 190   op->flags = 0;
 191   op->reserved = nullptr;
 192   op++;
 193   op->op = GRPC_OP_RECV_MESSAGE;
 194   op->data.recv_message.recv_message = &request_payload_recv;
 195   op->flags = 0;
 196   op->reserved = nullptr;
 197   op++;
 198   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(102),
 199                                 nullptr);
 200   GPR_ASSERT(GRPC_CALL_OK == error);
 201
 202   CQ_EXPECT_COMPLETION(cqv, tag(102), 1);
 203   cq_verify(cqv);
 204
 205   memset(ops, 0, sizeof(ops));
 206   op = ops;
 207   op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
 208   op->data.recv_close_on_server.cancelled = &was_cancelled;
 209   op->flags = 0;
 210   op->reserved = nullptr;
 211   op++;
 212   op->op = GRPC_OP_SEND_MESSAGE;
 213   op->data.send_message.send_message = response_payload;
 214   op->flags = 0;
 215   op->reserved = nullptr;
 216   op++;
 217   op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
 218   op->data.send_status_from_server.trailing_metadata_count = 0;
 219   op->data.send_status_from_server.status = GRPC_STATUS_OK;
 220   grpc_slice status_details = grpc_slice_from_static_string("xyz");
 221   op->data.send_status_from_server.status_details = &status_details;
 222   op->flags = 0;
 223   op->reserved = nullptr;
 224   op++;
 225   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(103),
 226                                 nullptr);
 227   GPR_ASSERT(GRPC_CALL_OK == error);
 228
 229   CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
 230   CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
 231   cq_verify(cqv);
 232
 233   GPR_ASSERT(status == GRPC_STATUS_OK);
 234   GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
 235   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
 236   GPR_ASSERT(was_cancelled == 0);
 237   GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
 238   GPR_ASSERT(
 239       byte_buffer_eq_slice(response_payload_recv, response_payload_slice));
 240
 241   grpc_slice_unref(details);
 242   grpc_metadata_array_destroy(&initial_metadata_recv);
 243   grpc_metadata_array_destroy(&trailing_metadata_recv);
 244   grpc_metadata_array_destroy(&request_metadata_recv);
 245   grpc_call_details_destroy(&call_details);
 246
 247   grpc_call_unref(c);
 248   grpc_call_unref(s);
 249
 250   cq_verifier_destroy(cqv);
 251
 252   grpc_byte_buffer_destroy(request_payload);
 253   grpc_byte_buffer_destroy(response_payload);
 254   grpc_byte_buffer_destroy(request_payload_recv);
 255   grpc_byte_buffer_destroy(response_payload_recv);
 256 }
 257
 258 /* Client sends a request with payload, server reads then returns a response
 259    payload and status. */
 260 static void test_invoke_request_response_with_payload(
 261     grpc_end2end_test_config config) {
 262   grpc_end2end_test_fixture f = begin_test(
 263       config, "test_invoke_request_response_with_payload", nullptr, nullptr);
 264   request_response_with_payload(config, f);
 265   end_test(&f);
 266   config.tear_down_data(&f);
 267 }
 268
 269 static void test_invoke_10_request_response_with_payload(
 270     grpc_end2end_test_config config) {
 271   int i;
 272   grpc_end2end_test_fixture f = begin_test(
 273       config, "test_invoke_10_request_response_with_payload", nullptr, nullptr);
 274   for (i = 0; i < 10; i++) {
 275     request_response_with_payload(config, f);
 276   }
 277   end_test(&f);
 278   config.tear_down_data(&f);
 279 }
 280
 281 void payload(grpc_end2end_test_config config) {
 282   test_invoke_request_response_with_payload(config);
 283   test_invoke_10_request_response_with_payload(config);
 284 }
 285
 286 void payload_pre_init(void) {}