src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc

   1 //
   2 // Copyright 2016 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 // This is similar to the sockaddr resolver, except that it supports a
  18 // bunch of query args that are useful for dependency injection in tests.
  19
  20 #include <grpc/support/port_platform.h>
  21
  22 #include <limits.h>
  23 #include <stdbool.h>
  24 #include <stdio.h>
  25 #include <stdlib.h>
  26 #include <string.h>
  27
  28 #include <grpc/support/alloc.h>
  29 #include <grpc/support/string_util.h>
  30
  31 #include "src/core/ext/filters/client_channel/parse_address.h"
  32 #include "src/core/ext/filters/client_channel/resolver_registry.h"
  33 #include "src/core/ext/filters/client_channel/server_address.h"
  34 #include "src/core/lib/channel/channel_args.h"
  35 #include "src/core/lib/gpr/string.h"
  36 #include "src/core/lib/gpr/useful.h"
  37 #include "src/core/lib/iomgr/closure.h"
  38 #include "src/core/lib/iomgr/combiner.h"
  39 #include "src/core/lib/iomgr/resolve_address.h"
  40 #include "src/core/lib/iomgr/unix_sockets_posix.h"
  41 #include "src/core/lib/slice/slice_internal.h"
  42 #include "src/core/lib/slice/slice_string_helpers.h"
  43
  44 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
  45
  46 namespace grpc_core {
  47
  48 // This cannot be in an anonymous namespace, because it is a friend of
  49 // FakeResolverResponseGenerator.
  50 class FakeResolver : public Resolver {
  51  public:
  52   explicit FakeResolver(ResolverArgs args);
  53
  54   void StartLocked() override;
  55
  56   void RequestReresolutionLocked() override;
  57
  58  private:
  59   friend class FakeResolverResponseGenerator;
  60
  61   virtual ~FakeResolver();
  62
  63   void ShutdownLocked() override {
  64     shutdown_ = true;
  65     if (response_generator_ != nullptr) {
  66       response_generator_->SetFakeResolver(nullptr);
  67       response_generator_.reset();
  68     }
  69   }
  70
  71   void MaybeSendResultLocked();
  72
  73   static void ReturnReresolutionResult(void* arg, grpc_error* error);
  74
  75   // passed-in parameters
  76   grpc_channel_args* channel_args_ = nullptr;
  77   RefCountedPtr<FakeResolverResponseGenerator> response_generator_;
  78   // If has_next_result_ is true, next_result_ is the next resolution result
  79   // to be returned.
  80   bool has_next_result_ = false;
  81   Result next_result_;
  82   // Result to use for the pretended re-resolution in
  83   // RequestReresolutionLocked().
  84   bool has_reresolution_result_ = false;
  85   Result reresolution_result_;
  86   // True after the call to StartLocked().
  87   bool started_ = false;
  88   // True after the call to ShutdownLocked().
  89   bool shutdown_ = false;
  90   // if true, return failure
  91   bool return_failure_ = false;
  92   // pending re-resolution
  93   grpc_closure reresolution_closure_;
  94   bool reresolution_closure_pending_ = false;
  95 };
  96
  97 FakeResolver::FakeResolver(ResolverArgs args)
  98     : Resolver(args.combiner, std::move(args.result_handler)),
  99       response_generator_(
 100           FakeResolverResponseGenerator::GetFromArgs(args.args)) {
 101   GRPC_CLOSURE_INIT(&reresolution_closure_, ReturnReresolutionResult, this,
 102                     grpc_combiner_scheduler(combiner()));
 103   // Channels sharing the same subchannels may have different resolver response
 104   // generators. If we don't remove this arg, subchannel pool will create new
 105   // subchannels for the same address instead of reusing existing ones because
 106   // of different values of this channel arg.
 107   const char* args_to_remove[] = {GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR};
 108   channel_args_ = grpc_channel_args_copy_and_remove(
 109       args.args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove));
 110   if (response_generator_ != nullptr) {
 111     response_generator_->SetFakeResolver(Ref());
 112   }
 113 }
 114
 115 FakeResolver::~FakeResolver() { grpc_channel_args_destroy(channel_args_); }
 116
 117 void FakeResolver::StartLocked() {
 118   started_ = true;
 119   MaybeSendResultLocked();
 120 }
 121
 122 void FakeResolver::RequestReresolutionLocked() {
 123   if (has_reresolution_result_ || return_failure_) {
 124     next_result_ = reresolution_result_;
 125     has_next_result_ = true;
 126     // Return the result in a different closure, so that we don't call
 127     // back into the LB policy while it's still processing the previous
 128     // update.
 129     if (!reresolution_closure_pending_) {
 130       reresolution_closure_pending_ = true;
 131       Ref().release();  // ref held by closure
 132       GRPC_CLOSURE_SCHED(&reresolution_closure_, GRPC_ERROR_NONE);
 133     }
 134   }
 135 }
 136
 137 void FakeResolver::MaybeSendResultLocked() {
 138   if (!started_ || shutdown_) return;
 139   if (return_failure_) {
 140     // TODO(roth): Change resolver result generator to be able to inject
 141     // the error to be returned.
 142     result_handler()->ReturnError(grpc_error_set_int(
 143         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Resolver transient failure"),
 144         GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE));
 145     return_failure_ = false;
 146   } else if (has_next_result_) {
 147     Result result;
 148     result.addresses = std::move(next_result_.addresses);
 149     result.service_config = std::move(next_result_.service_config);
 150     // TODO(roth): Use std::move() once grpc_error is converted to C++.
 151     result.service_config_error = next_result_.service_config_error;
 152     next_result_.service_config_error = GRPC_ERROR_NONE;
 153     // When both next_results_ and channel_args_ contain an arg with the same
 154     // name, only the one in next_results_ will be kept since next_results_ is
 155     // before channel_args_.
 156     result.args = grpc_channel_args_union(next_result_.args, channel_args_);
 157     result_handler()->ReturnResult(std::move(result));
 158     has_next_result_ = false;
 159   }
 160 }
 161
 162 void FakeResolver::ReturnReresolutionResult(void* arg, grpc_error* error) {
 163   FakeResolver* self = static_cast<FakeResolver*>(arg);
 164   self->reresolution_closure_pending_ = false;
 165   self->MaybeSendResultLocked();
 166   self->Unref();
 167 }
 168
 169 //
 170 // FakeResolverResponseGenerator
 171 //
 172
 173 FakeResolverResponseGenerator::FakeResolverResponseGenerator() {}
 174
 175 FakeResolverResponseGenerator::~FakeResolverResponseGenerator() {}
 176
 177 struct SetResponseClosureArg {
 178   grpc_closure set_response_closure;
 179   RefCountedPtr<FakeResolver> resolver;
 180   Resolver::Result result;
 181   bool has_result = false;
 182   bool immediate = true;
 183 };
 184
 185 void FakeResolverResponseGenerator::SetResponseLocked(void* arg,
 186                                                       grpc_error* error) {
 187   SetResponseClosureArg* closure_arg = static_cast<SetResponseClosureArg*>(arg);
 188   auto& resolver = closure_arg->resolver;
 189   if (!resolver->shutdown_) {
 190     resolver->next_result_ = std::move(closure_arg->result);
 191     resolver->has_next_result_ = true;
 192     resolver->MaybeSendResultLocked();
 193   }
 194   Delete(closure_arg);
 195 }
 196
 197 void FakeResolverResponseGenerator::SetResponse(Resolver::Result result) {
 198   RefCountedPtr<FakeResolver> resolver;
 199   {
 200     MutexLock lock(&mu_);
 201     if (resolver_ == nullptr) {
 202       has_result_ = true;
 203       result_ = std::move(result);
 204       return;
 205     }
 206     resolver = resolver_->Ref();
 207   }
 208   SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>();
 209   closure_arg->resolver = std::move(resolver);
 210   closure_arg->result = std::move(result);
 211   GRPC_CLOSURE_SCHED(
 212       GRPC_CLOSURE_INIT(
 213           &closure_arg->set_response_closure, SetResponseLocked, closure_arg,
 214           grpc_combiner_scheduler(closure_arg->resolver->combiner())),
 215       GRPC_ERROR_NONE);
 216 }
 217
 218 void FakeResolverResponseGenerator::SetReresolutionResponseLocked(
 219     void* arg, grpc_error* error) {
 220   SetResponseClosureArg* closure_arg = static_cast<SetResponseClosureArg*>(arg);
 221   auto& resolver = closure_arg->resolver;
 222   if (!resolver->shutdown_) {
 223     resolver->reresolution_result_ = std::move(closure_arg->result);
 224     resolver->has_reresolution_result_ = closure_arg->has_result;
 225   }
 226   Delete(closure_arg);
 227 }
 228
 229 void FakeResolverResponseGenerator::SetReresolutionResponse(
 230     Resolver::Result result) {
 231   RefCountedPtr<FakeResolver> resolver;
 232   {
 233     MutexLock lock(&mu_);
 234     GPR_ASSERT(resolver_ != nullptr);
 235     resolver = resolver_->Ref();
 236   }
 237   SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>();
 238   closure_arg->resolver = std::move(resolver);
 239   closure_arg->result = std::move(result);
 240   closure_arg->has_result = true;
 241   GRPC_CLOSURE_SCHED(
 242       GRPC_CLOSURE_INIT(
 243           &closure_arg->set_response_closure, SetReresolutionResponseLocked,
 244           closure_arg,
 245           grpc_combiner_scheduler(closure_arg->resolver->combiner())),
 246       GRPC_ERROR_NONE);
 247 }
 248
 249 void FakeResolverResponseGenerator::UnsetReresolutionResponse() {
 250   RefCountedPtr<FakeResolver> resolver;
 251   {
 252     MutexLock lock(&mu_);
 253     GPR_ASSERT(resolver_ != nullptr);
 254     resolver = resolver_->Ref();
 255   }
 256   SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>();
 257   closure_arg->resolver = std::move(resolver);
 258   GRPC_CLOSURE_SCHED(
 259       GRPC_CLOSURE_INIT(
 260           &closure_arg->set_response_closure, SetReresolutionResponseLocked,
 261           closure_arg,
 262           grpc_combiner_scheduler(closure_arg->resolver->combiner())),
 263       GRPC_ERROR_NONE);
 264 }
 265
 266 void FakeResolverResponseGenerator::SetFailureLocked(void* arg,
 267                                                      grpc_error* error) {
 268   SetResponseClosureArg* closure_arg = static_cast<SetResponseClosureArg*>(arg);
 269   auto& resolver = closure_arg->resolver;
 270   if (!resolver->shutdown_) {
 271     resolver->return_failure_ = true;
 272     if (closure_arg->immediate) resolver->MaybeSendResultLocked();
 273   }
 274   Delete(closure_arg);
 275 }
 276
 277 void FakeResolverResponseGenerator::SetFailure() {
 278   RefCountedPtr<FakeResolver> resolver;
 279   {
 280     MutexLock lock(&mu_);
 281     GPR_ASSERT(resolver_ != nullptr);
 282     resolver = resolver_->Ref();
 283   }
 284   SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>();
 285   closure_arg->resolver = std::move(resolver);
 286   GRPC_CLOSURE_SCHED(
 287       GRPC_CLOSURE_INIT(
 288           &closure_arg->set_response_closure, SetFailureLocked, closure_arg,
 289           grpc_combiner_scheduler(closure_arg->resolver->combiner())),
 290       GRPC_ERROR_NONE);
 291 }
 292
 293 void FakeResolverResponseGenerator::SetFailureOnReresolution() {
 294   RefCountedPtr<FakeResolver> resolver;
 295   {
 296     MutexLock lock(&mu_);
 297     GPR_ASSERT(resolver_ != nullptr);
 298     resolver = resolver_->Ref();
 299   }
 300   SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>();
 301   closure_arg->resolver = std::move(resolver);
 302   closure_arg->immediate = false;
 303   GRPC_CLOSURE_SCHED(
 304       GRPC_CLOSURE_INIT(
 305           &closure_arg->set_response_closure, SetFailureLocked, closure_arg,
 306           grpc_combiner_scheduler(closure_arg->resolver->combiner())),
 307       GRPC_ERROR_NONE);
 308 }
 309
 310 void FakeResolverResponseGenerator::SetFakeResolver(
 311     RefCountedPtr<FakeResolver> resolver) {
 312   MutexLock lock(&mu_);
 313   resolver_ = std::move(resolver);
 314   if (resolver_ == nullptr) return;
 315   if (has_result_) {
 316     SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>();
 317     closure_arg->resolver = resolver_->Ref();
 318     closure_arg->result = std::move(result_);
 319     GRPC_CLOSURE_SCHED(
 320         GRPC_CLOSURE_INIT(&closure_arg->set_response_closure, SetResponseLocked,
 321                           closure_arg,
 322                           grpc_combiner_scheduler(resolver_->combiner())),
 323         GRPC_ERROR_NONE);
 324     has_result_ = false;
 325   }
 326 }
 327
 328 namespace {
 329
 330 static void* response_generator_arg_copy(void* p) {
 331   FakeResolverResponseGenerator* generator =
 332       static_cast<FakeResolverResponseGenerator*>(p);
 333   // TODO(roth): We currently deal with this ref manually.  Once the
 334   // new channel args code is converted to C++, find a way to track this ref
 335   // in a cleaner way.
 336   RefCountedPtr<FakeResolverResponseGenerator> copy = generator->Ref();
 337   copy.release();
 338   return p;
 339 }
 340
 341 static void response_generator_arg_destroy(void* p) {
 342   FakeResolverResponseGenerator* generator =
 343       static_cast<FakeResolverResponseGenerator*>(p);
 344   generator->Unref();
 345 }
 346
 347 static int response_generator_cmp(void* a, void* b) { return GPR_ICMP(a, b); }
 348
 349 static const grpc_arg_pointer_vtable response_generator_arg_vtable = {
 350     response_generator_arg_copy, response_generator_arg_destroy,
 351     response_generator_cmp};
 352
 353 }  // namespace
 354
 355 grpc_arg FakeResolverResponseGenerator::MakeChannelArg(
 356     FakeResolverResponseGenerator* generator) {
 357   grpc_arg arg;
 358   arg.type = GRPC_ARG_POINTER;
 359   arg.key = (char*)GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR;
 360   arg.value.pointer.p = generator;
 361   arg.value.pointer.vtable = &response_generator_arg_vtable;
 362   return arg;
 363 }
 364
 365 RefCountedPtr<FakeResolverResponseGenerator>
 366 FakeResolverResponseGenerator::GetFromArgs(const grpc_channel_args* args) {
 367   const grpc_arg* arg =
 368       grpc_channel_args_find(args, GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR);
 369   if (arg == nullptr || arg->type != GRPC_ARG_POINTER) return nullptr;
 370   return static_cast<FakeResolverResponseGenerator*>(arg->value.pointer.p)
 371       ->Ref();
 372 }
 373
 374 //
 375 // Factory
 376 //
 377
 378 namespace {
 379
 380 class FakeResolverFactory : public ResolverFactory {
 381  public:
 382   bool IsValidUri(const grpc_uri* uri) const override { return true; }
 383
 384   OrphanablePtr<Resolver> CreateResolver(ResolverArgs args) const override {
 385     return OrphanablePtr<Resolver>(New<FakeResolver>(std::move(args)));
 386   }
 387
 388   const char* scheme() const override { return "fake"; }
 389 };
 390
 391 }  // namespace
 392
 393 }  // namespace grpc_core
 394
 395 void grpc_resolver_fake_init() {
 396   grpc_core::ResolverRegistry::Builder::RegisterResolverFactory(
 397       grpc_core::UniquePtr<grpc_core::ResolverFactory>(
 398           grpc_core::New<grpc_core::FakeResolverFactory>()));
 399 }
 400
 401 void grpc_resolver_fake_shutdown() {}