test/core/end2end/inproc_callback_test.cc

   1 /*
   2  *
   3  * Copyright 2018 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 <string.h>
  20
  21 #include <grpc/support/alloc.h>
  22 #include <grpc/support/log.h>
  23 #include <grpc/support/sync.h>
  24
  25 #include "src/core/ext/transport/inproc/inproc_transport.h"
  26 #include "src/core/lib/surface/channel.h"
  27 #include "src/core/lib/surface/completion_queue.h"
  28 #include "src/core/lib/surface/server.h"
  29 #include "test/core/end2end/end2end_tests.h"
  30 #include "test/core/util/port.h"
  31 #include "test/core/util/test_config.h"
  32
  33 typedef struct inproc_fixture_data {
  34   bool phony;  // reserved for future expansion. Struct can't be empty
  35 } inproc_fixture_data;
  36
  37 namespace {
  38 template <typename F>
  39 class CQDeletingCallback : public grpc_completion_queue_functor {
  40  public:
  41   explicit CQDeletingCallback(F f) : func_(f) {
  42     functor_run = &CQDeletingCallback::Run;
  43     inlineable = false;
  44   }
  45   ~CQDeletingCallback() {}
  46   static void Run(grpc_completion_queue_functor* cb, int ok) {
  47     auto* callback = static_cast<CQDeletingCallback*>(cb);
  48     callback->func_(static_cast<bool>(ok));
  49     delete callback;
  50   }
  51
  52  private:
  53   F func_;
  54 };
  55
  56 template <typename F>
  57 grpc_completion_queue_functor* NewDeletingCallback(F f) {
  58   return new CQDeletingCallback<F>(f);
  59 }
  60
  61 class ShutdownCallback : public grpc_completion_queue_functor {
  62  public:
  63   ShutdownCallback() : done_(false) {
  64     functor_run = &ShutdownCallback::StaticRun;
  65     inlineable = false;
  66     gpr_mu_init(&mu_);
  67     gpr_cv_init(&cv_);
  68   }
  69   ~ShutdownCallback() {
  70     gpr_mu_destroy(&mu_);
  71     gpr_cv_destroy(&cv_);
  72   }
  73   static void StaticRun(grpc_completion_queue_functor* cb, int ok) {
  74     auto* callback = static_cast<ShutdownCallback*>(cb);
  75     callback->Run(static_cast<bool>(ok));
  76   }
  77   void Run(bool /*ok*/) {
  78     gpr_log(GPR_DEBUG, "CQ shutdown notification invoked");
  79     gpr_mu_lock(&mu_);
  80     done_ = true;
  81     gpr_cv_broadcast(&cv_);
  82     gpr_mu_unlock(&mu_);
  83   }
  84   // The Wait function waits for a specified amount of
  85   // time for the completion of the shutdown and returns
  86   // whether it was successfully shut down
  87   bool Wait(gpr_timespec deadline) {
  88     gpr_mu_lock(&mu_);
  89     while (!done_ && !gpr_cv_wait(&cv_, &mu_, deadline)) {
  90     }
  91     bool ret = done_;
  92     gpr_mu_unlock(&mu_);
  93     return ret;
  94   }
  95
  96  private:
  97   bool done_;
  98   gpr_mu mu_;
  99   gpr_cv cv_;
 100 };
 101
 102 ShutdownCallback* g_shutdown_callback;
 103 }  // namespace
 104
 105 // The following global structure is the tag collection. It holds
 106 // all information related to tags expected and tags received
 107 // during the execution, with each callback setting a tag.
 108 // The tag sets are implemented and checked using arrays and
 109 // linear lookups (rather than maps) so that this test doesn't
 110 // need the C++ standard library.
 111 static gpr_mu tags_mu;
 112 static gpr_cv tags_cv;
 113 const size_t kAvailableTags = 4;
 114 bool tags[kAvailableTags];
 115 bool tags_valid[kAvailableTags];
 116 bool tags_expected[kAvailableTags];
 117 bool tags_needed[kAvailableTags];
 118
 119 // Mark that a tag is expected; this function must be executed in the
 120 // main thread only while there are no other threads altering the
 121 // expectation set (e.g., by calling expect_tag or verify_tags)
 122 static void expect_tag(intptr_t tag, bool ok) {
 123   size_t idx = static_cast<size_t>(tag);
 124   GPR_ASSERT(idx < kAvailableTags);
 125   tags_needed[idx] = true;
 126   tags_expected[idx] = ok;
 127 }
 128
 129 // Check that the expected tags have reached, within a certain
 130 // deadline. This must also be executed only on the main thread while
 131 // there are no other threads altering the expectation set (e.g., by
 132 // calling expect_tag or verify_tags). The tag verifier doesn't have
 133 // to drive the CQ at all (unlike the next-based end2end tests)
 134 // because the tags will get set when the callbacks are executed,
 135 // which happens when a particular batch related to a callback is
 136 // complete.
 137 static void verify_tags(gpr_timespec deadline) {
 138   bool done = false;
 139
 140   gpr_mu_lock(&tags_mu);
 141   while (!done) {
 142     done = gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), deadline) > 0;
 143     for (size_t i = 0; i < kAvailableTags; i++) {
 144       if (tags_needed[i]) {
 145         if (tags_valid[i]) {
 146           gpr_log(GPR_DEBUG, "Verifying tag %d", static_cast<int>(i));
 147           if (tags[i] != tags_expected[i]) {
 148             gpr_log(GPR_ERROR, "Got wrong result (%d instead of %d) for tag %d",
 149                     tags[i], tags_expected[i], static_cast<int>(i));
 150             GPR_ASSERT(false);
 151           }
 152           tags_valid[i] = false;
 153           tags_needed[i] = false;
 154         } else if (done) {
 155           gpr_log(GPR_ERROR, "Didn't get tag %d", static_cast<int>(i));
 156           GPR_ASSERT(false);
 157         }
 158       }
 159     }
 160     bool empty = true;
 161     for (size_t i = 0; i < kAvailableTags; i++) {
 162       if (tags_needed[i]) {
 163         empty = false;
 164       }
 165     }
 166     done = done || empty;
 167     if (done) {
 168       for (size_t i = 0; i < kAvailableTags; i++) {
 169         if (tags_valid[i]) {
 170           gpr_log(GPR_ERROR, "Got unexpected tag %d and result %d",
 171                   static_cast<int>(i), tags[i]);
 172           GPR_ASSERT(false);
 173         }
 174         tags_valid[i] = false;
 175       }
 176     } else {
 177       gpr_cv_wait(&tags_cv, &tags_mu, deadline);
 178     }
 179   }
 180   gpr_mu_unlock(&tags_mu);
 181 }
 182
 183 // This function creates a callback functor that emits the
 184 // desired tag into the global tag set
 185 static grpc_completion_queue_functor* tag(intptr_t t) {
 186   auto func = [t](bool ok) {
 187     gpr_mu_lock(&tags_mu);
 188     gpr_log(GPR_DEBUG, "Completing operation %" PRIdPTR, t);
 189     bool was_empty = true;
 190     for (size_t i = 0; i < kAvailableTags; i++) {
 191       if (tags_valid[i]) {
 192         was_empty = false;
 193       }
 194     }
 195     size_t idx = static_cast<size_t>(t);
 196     tags[idx] = ok;
 197     tags_valid[idx] = true;
 198     if (was_empty) {
 199       gpr_cv_signal(&tags_cv);
 200     }
 201     gpr_mu_unlock(&tags_mu);
 202   };
 203   auto cb = NewDeletingCallback(func);
 204   return cb;
 205 }
 206
 207 static grpc_end2end_test_fixture inproc_create_fixture(
 208     grpc_channel_args* /*client_args*/, grpc_channel_args* /*server_args*/) {
 209   grpc_end2end_test_fixture f;
 210   inproc_fixture_data* ffd = static_cast<inproc_fixture_data*>(
 211       gpr_malloc(sizeof(inproc_fixture_data)));
 212   memset(&f, 0, sizeof(f));
 213
 214   f.fixture_data = ffd;
 215   g_shutdown_callback = new ShutdownCallback();
 216   f.cq =
 217       grpc_completion_queue_create_for_callback(g_shutdown_callback, nullptr);
 218   f.shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
 219
 220   return f;
 221 }
 222
 223 void inproc_init_client(grpc_end2end_test_fixture* f,
 224                         grpc_channel_args* client_args) {
 225   f->client = grpc_inproc_channel_create(f->server, client_args, nullptr);
 226   GPR_ASSERT(f->client);
 227 }
 228
 229 void inproc_init_server(grpc_end2end_test_fixture* f,
 230                         grpc_channel_args* server_args) {
 231   if (f->server) {
 232     grpc_server_destroy(f->server);
 233   }
 234   f->server = grpc_server_create(server_args, nullptr);
 235   grpc_server_register_completion_queue(f->server, f->cq, nullptr);
 236   grpc_server_start(f->server);
 237 }
 238
 239 void inproc_tear_down(grpc_end2end_test_fixture* f) {
 240   inproc_fixture_data* ffd = static_cast<inproc_fixture_data*>(f->fixture_data);
 241   gpr_free(ffd);
 242 }
 243
 244 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
 245                                             const char* test_name,
 246                                             grpc_channel_args* client_args,
 247                                             grpc_channel_args* server_args) {
 248   grpc_end2end_test_fixture f;
 249   gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
 250   f = config.create_fixture(client_args, server_args);
 251   config.init_server(&f, server_args);
 252   config.init_client(&f, client_args);
 253   return f;
 254 }
 255
 256 static gpr_timespec n_seconds_from_now(int n) {
 257   return grpc_timeout_seconds_to_deadline(n);
 258 }
 259
 260 static gpr_timespec five_seconds_from_now() { return n_seconds_from_now(5); }
 261
 262 static void drain_cq(grpc_completion_queue* /*cq*/) {
 263   // Wait for the shutdown callback to arrive, or fail the test
 264   GPR_ASSERT(g_shutdown_callback->Wait(five_seconds_from_now()));
 265   gpr_log(GPR_DEBUG, "CQ shutdown wait complete");
 266   delete g_shutdown_callback;
 267 }
 268
 269 static void shutdown_server(grpc_end2end_test_fixture* f) {
 270   if (!f->server) return;
 271   grpc_server_shutdown_and_notify(
 272       f->server, f->shutdown_cq,
 273       reinterpret_cast<void*>(static_cast<intptr_t>(1000)));
 274   GPR_ASSERT(
 275       grpc_completion_queue_pluck(f->shutdown_cq, (void*)((intptr_t)1000),
 276                                   grpc_timeout_seconds_to_deadline(5), nullptr)
 277           .type == GRPC_OP_COMPLETE);
 278   grpc_server_destroy(f->server);
 279   f->server = nullptr;
 280 }
 281
 282 static void shutdown_client(grpc_end2end_test_fixture* f) {
 283   if (!f->client) return;
 284   grpc_channel_destroy(f->client);
 285   f->client = nullptr;
 286 }
 287
 288 static void end_test(grpc_end2end_test_fixture* f) {
 289   shutdown_server(f);
 290   shutdown_client(f);
 291
 292   grpc_completion_queue_shutdown(f->cq);
 293   drain_cq(f->cq);
 294   grpc_completion_queue_destroy(f->cq);
 295   grpc_completion_queue_destroy(f->shutdown_cq);
 296 }
 297
 298 static void simple_request_body(grpc_end2end_test_config config,
 299                                 grpc_end2end_test_fixture f) {
 300   grpc_call* c;
 301   grpc_call* s;
 302   grpc_op ops[6];
 303   grpc_op* op;
 304   grpc_metadata_array initial_metadata_recv;
 305   grpc_metadata_array trailing_metadata_recv;
 306   grpc_metadata_array request_metadata_recv;
 307   grpc_call_details call_details;
 308   grpc_status_code status;
 309   const char* error_string;
 310   grpc_call_error error;
 311   grpc_slice details;
 312   int was_cancelled = 2;
 313   char* peer;
 314   gpr_timespec deadline = five_seconds_from_now();
 315
 316   c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
 317                                grpc_slice_from_static_string("/foo"), nullptr,
 318                                deadline, nullptr);
 319   GPR_ASSERT(c);
 320
 321   peer = grpc_call_get_peer(c);
 322   GPR_ASSERT(peer != nullptr);
 323   gpr_log(GPR_DEBUG, "client_peer_before_call=%s", peer);
 324   gpr_free(peer);
 325
 326   grpc_metadata_array_init(&initial_metadata_recv);
 327   grpc_metadata_array_init(&trailing_metadata_recv);
 328   grpc_metadata_array_init(&request_metadata_recv);
 329   grpc_call_details_init(&call_details);
 330
 331   // Create a basic client unary request batch (no payload)
 332   memset(ops, 0, sizeof(ops));
 333   op = ops;
 334   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 335   op->data.send_initial_metadata.count = 0;
 336   op->flags = 0;
 337   op->reserved = nullptr;
 338   op++;
 339   op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 340   op->flags = 0;
 341   op->reserved = nullptr;
 342   op++;
 343   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 344   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 345   op->flags = 0;
 346   op->reserved = nullptr;
 347   op++;
 348   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 349   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 350   op->data.recv_status_on_client.status = &status;
 351   op->data.recv_status_on_client.status_details = &details;
 352   op->data.recv_status_on_client.error_string = &error_string;
 353   op->flags = 0;
 354   op->reserved = nullptr;
 355   op++;
 356   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
 357                                 nullptr);
 358   GPR_ASSERT(GRPC_CALL_OK == error);
 359
 360   // Register a call at the server-side to match the incoming client call
 361   error = grpc_server_request_call(f.server, &s, &call_details,
 362                                    &request_metadata_recv, f.cq, f.cq, tag(2));
 363   GPR_ASSERT(GRPC_CALL_OK == error);
 364
 365   // We expect that the server call creation callback (and no others) will
 366   // execute now since no other batch should be complete.
 367   expect_tag(2, true);
 368   verify_tags(deadline);
 369
 370   peer = grpc_call_get_peer(s);
 371   GPR_ASSERT(peer != nullptr);
 372   gpr_log(GPR_DEBUG, "server_peer=%s", peer);
 373   gpr_free(peer);
 374   peer = grpc_call_get_peer(c);
 375   GPR_ASSERT(peer != nullptr);
 376   gpr_log(GPR_DEBUG, "client_peer=%s", peer);
 377   gpr_free(peer);
 378
 379   // Create the server response batch (no payload)
 380   memset(ops, 0, sizeof(ops));
 381   op = ops;
 382   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 383   op->data.send_initial_metadata.count = 0;
 384   op->flags = 0;
 385   op->reserved = nullptr;
 386   op++;
 387   op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
 388   op->data.send_status_from_server.trailing_metadata_count = 0;
 389   op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED;
 390   grpc_slice status_details = grpc_slice_from_static_string("xyz");
 391   op->data.send_status_from_server.status_details = &status_details;
 392   op->flags = 0;
 393   op->reserved = nullptr;
 394   op++;
 395   op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
 396   op->data.recv_close_on_server.cancelled = &was_cancelled;
 397   op->flags = 0;
 398   op->reserved = nullptr;
 399   op++;
 400   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(3),
 401                                 nullptr);
 402   GPR_ASSERT(GRPC_CALL_OK == error);
 403
 404   // Both the client request and server response batches should get complete
 405   // now and we should see that their callbacks have been executed
 406   expect_tag(3, true);
 407   expect_tag(1, true);
 408   verify_tags(deadline);
 409
 410   GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
 411   GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
 412   // the following sanity check makes sure that the requested error string is
 413   // correctly populated by the core. It looks for certain substrings that are
 414   // not likely to change much. Some parts of the error, like time created,
 415   // obviously are not checked.
 416   GPR_ASSERT(nullptr != strstr(error_string, "xyz"));
 417   GPR_ASSERT(nullptr != strstr(error_string, "description"));
 418   GPR_ASSERT(nullptr != strstr(error_string, "Error received from peer"));
 419   GPR_ASSERT(nullptr != strstr(error_string, "grpc_message"));
 420   GPR_ASSERT(nullptr != strstr(error_string, "grpc_status"));
 421   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
 422   GPR_ASSERT(0 == call_details.flags);
 423   GPR_ASSERT(was_cancelled == 0);
 424
 425   grpc_slice_unref(details);
 426   gpr_free(static_cast<void*>(const_cast<char*>(error_string)));
 427   grpc_metadata_array_destroy(&initial_metadata_recv);
 428   grpc_metadata_array_destroy(&trailing_metadata_recv);
 429   grpc_metadata_array_destroy(&request_metadata_recv);
 430   grpc_call_details_destroy(&call_details);
 431
 432   grpc_call_unref(c);
 433   grpc_call_unref(s);
 434
 435   int expected_calls = 1;
 436   if (config.feature_mask & FEATURE_MASK_SUPPORTS_REQUEST_PROXYING) {
 437     expected_calls *= 2;
 438   }
 439 }
 440
 441 static void test_invoke_simple_request(grpc_end2end_test_config config) {
 442   grpc_end2end_test_fixture f;
 443
 444   f = begin_test(config, "test_invoke_simple_request", nullptr, nullptr);
 445   simple_request_body(config, f);
 446   end_test(&f);
 447   config.tear_down_data(&f);
 448 }
 449
 450 static void test_invoke_10_simple_requests(grpc_end2end_test_config config) {
 451   int i;
 452   grpc_end2end_test_fixture f =
 453       begin_test(config, "test_invoke_10_simple_requests", nullptr, nullptr);
 454   for (i = 0; i < 10; i++) {
 455     simple_request_body(config, f);
 456     gpr_log(GPR_INFO, "Running test: Passed simple request %d", i);
 457   }
 458   end_test(&f);
 459   config.tear_down_data(&f);
 460 }
 461
 462 static void test_invoke_many_simple_requests(grpc_end2end_test_config config) {
 463   int i;
 464   const int many = 1000;
 465   grpc_end2end_test_fixture f =
 466       begin_test(config, "test_invoke_many_simple_requests", nullptr, nullptr);
 467   gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
 468   for (i = 0; i < many; i++) {
 469     simple_request_body(config, f);
 470   }
 471   double us =
 472       gpr_timespec_to_micros(gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), t1)) /
 473       many;
 474   gpr_log(GPR_INFO, "Time per ping %f us", us);
 475   end_test(&f);
 476   config.tear_down_data(&f);
 477 }
 478
 479 static void simple_request(grpc_end2end_test_config config) {
 480   int i;
 481   for (i = 0; i < 10; i++) {
 482     test_invoke_simple_request(config);
 483   }
 484   test_invoke_10_simple_requests(config);
 485   test_invoke_many_simple_requests(config);
 486 }
 487
 488 static void simple_request_pre_init() {
 489   gpr_mu_init(&tags_mu);
 490   gpr_cv_init(&tags_cv);
 491 }
 492
 493 /* All test configurations */
 494 static grpc_end2end_test_config configs[] = {
 495     {"inproc-callback", FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, nullptr,
 496      inproc_create_fixture, inproc_init_client, inproc_init_server,
 497      inproc_tear_down},
 498 };
 499
 500 int main(int argc, char** argv) {
 501   grpc::testing::TestEnvironment env(argc, argv);
 502   grpc_init();
 503
 504   simple_request_pre_init();
 505   simple_request(configs[0]);
 506
 507   grpc_shutdown();
 508
 509   return 0;
 510 }