legacy-libs/grpc/deps/grpc/src/core/lib/channel/handshaker.cc

   1 /*
   2  *
   3  * Copyright 2016 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 <grpc/support/port_platform.h>
  20
  21 #include <string.h>
  22
  23 #include <grpc/impl/codegen/slice.h>
  24 #include <grpc/support/alloc.h>
  25 #include <grpc/support/log.h>
  26 #include <grpc/support/string_util.h>
  27
  28 #include "src/core/lib/channel/channel_args.h"
  29 #include "src/core/lib/channel/handshaker.h"
  30 #include "src/core/lib/debug/trace.h"
  31 #include "src/core/lib/iomgr/timer.h"
  32 #include "src/core/lib/slice/slice_internal.h"
  33
  34 namespace grpc_core {
  35
  36 TraceFlag grpc_handshaker_trace(false, "handshaker");
  37
  38 namespace {
  39
  40 char* HandshakerArgsString(HandshakerArgs* args) {
  41   char* args_str = grpc_channel_args_string(args->args);
  42   size_t num_args = args->args != nullptr ? args->args->num_args : 0;
  43   size_t read_buffer_length =
  44       args->read_buffer != nullptr ? args->read_buffer->length : 0;
  45   char* str;
  46   gpr_asprintf(&str,
  47                "{endpoint=%p, args=%p {size=%" PRIuPTR
  48                ": %s}, read_buffer=%p (length=%" PRIuPTR "), exit_early=%d}",
  49                args->endpoint, args->args, num_args, args_str,
  50                args->read_buffer, read_buffer_length, args->exit_early);
  51   gpr_free(args_str);
  52   return str;
  53 }
  54
  55 }  // namespace
  56
  57 HandshakeManager::HandshakeManager() { gpr_mu_init(&mu_); }
  58
  59 /// Add \a mgr to the server side list of all pending handshake managers, the
  60 /// list starts with \a *head.
  61 // Not thread-safe. Caller needs to synchronize.
  62 void HandshakeManager::AddToPendingMgrList(HandshakeManager** head) {
  63   GPR_ASSERT(prev_ == nullptr);
  64   GPR_ASSERT(next_ == nullptr);
  65   next_ = *head;
  66   if (*head) {
  67     (*head)->prev_ = this;
  68   }
  69   *head = this;
  70 }
  71
  72 /// Remove \a mgr from the server side list of all pending handshake managers.
  73 // Not thread-safe. Caller needs to synchronize.
  74 void HandshakeManager::RemoveFromPendingMgrList(HandshakeManager** head) {
  75   if (next_ != nullptr) {
  76     next_->prev_ = prev_;
  77   }
  78   if (prev_ != nullptr) {
  79     prev_->next_ = next_;
  80   } else {
  81     GPR_ASSERT(*head == this);
  82     *head = next_;
  83   }
  84 }
  85
  86 /// Shutdown all pending handshake managers starting at head on the server
  87 /// side. Not thread-safe. Caller needs to synchronize.
  88 void HandshakeManager::ShutdownAllPending(grpc_error* why) {
  89   auto* head = this;
  90   while (head != nullptr) {
  91     head->Shutdown(GRPC_ERROR_REF(why));
  92     head = head->next_;
  93   }
  94   GRPC_ERROR_UNREF(why);
  95 }
  96
  97 void HandshakeManager::Add(RefCountedPtr<Handshaker> handshaker) {
  98   if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
  99     gpr_log(
 100         GPR_INFO,
 101         "handshake_manager %p: adding handshaker %s [%p] at index %" PRIuPTR,
 102         this, handshaker->name(), handshaker.get(), handshakers_.size());
 103   }
 104   MutexLock lock(&mu_);
 105   handshakers_.push_back(std::move(handshaker));
 106 }
 107
 108 HandshakeManager::~HandshakeManager() {
 109   handshakers_.clear();
 110   gpr_mu_destroy(&mu_);
 111 }
 112
 113 void HandshakeManager::Shutdown(grpc_error* why) {
 114   {
 115     MutexLock lock(&mu_);
 116     // Shutdown the handshaker that's currently in progress, if any.
 117     if (!is_shutdown_ && index_ > 0) {
 118       is_shutdown_ = true;
 119       handshakers_[index_ - 1]->Shutdown(GRPC_ERROR_REF(why));
 120     }
 121   }
 122   GRPC_ERROR_UNREF(why);
 123 }
 124
 125 // Helper function to call either the next handshaker or the
 126 // on_handshake_done callback.
 127 // Returns true if we've scheduled the on_handshake_done callback.
 128 bool HandshakeManager::CallNextHandshakerLocked(grpc_error* error) {
 129   if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
 130     char* args_str = HandshakerArgsString(&args_);
 131     gpr_log(GPR_INFO,
 132             "handshake_manager %p: error=%s shutdown=%d index=%" PRIuPTR
 133             ", args=%s",
 134             this, grpc_error_string(error), is_shutdown_, index_, args_str);
 135     gpr_free(args_str);
 136   }
 137   GPR_ASSERT(index_ <= handshakers_.size());
 138   // If we got an error or we've been shut down or we're exiting early or
 139   // we've finished the last handshaker, invoke the on_handshake_done
 140   // callback.  Otherwise, call the next handshaker.
 141   if (error != GRPC_ERROR_NONE || is_shutdown_ || args_.exit_early ||
 142       index_ == handshakers_.size()) {
 143     if (error == GRPC_ERROR_NONE && is_shutdown_) {
 144       error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("handshaker shutdown");
 145       // It is possible that the endpoint has already been destroyed by
 146       // a shutdown call while this callback was sitting on the ExecCtx
 147       // with no error.
 148       if (args_.endpoint != nullptr) {
 149         // TODO(roth): It is currently necessary to shutdown endpoints
 150         // before destroying then, even when we know that there are no
 151         // pending read/write callbacks.  This should be fixed, at which
 152         // point this can be removed.
 153         grpc_endpoint_shutdown(args_.endpoint, GRPC_ERROR_REF(error));
 154         grpc_endpoint_destroy(args_.endpoint);
 155         args_.endpoint = nullptr;
 156         grpc_channel_args_destroy(args_.args);
 157         args_.args = nullptr;
 158         grpc_slice_buffer_destroy_internal(args_.read_buffer);
 159         gpr_free(args_.read_buffer);
 160         args_.read_buffer = nullptr;
 161       }
 162     }
 163     if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
 164       gpr_log(GPR_INFO,
 165               "handshake_manager %p: handshaking complete -- scheduling "
 166               "on_handshake_done with error=%s",
 167               this, grpc_error_string(error));
 168     }
 169     // Cancel deadline timer, since we're invoking the on_handshake_done
 170     // callback now.
 171     grpc_timer_cancel(&deadline_timer_);
 172     GRPC_CLOSURE_SCHED(&on_handshake_done_, error);
 173     is_shutdown_ = true;
 174   } else {
 175     auto handshaker = handshakers_[index_];
 176     if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
 177       gpr_log(
 178           GPR_INFO,
 179           "handshake_manager %p: calling handshaker %s [%p] at index %" PRIuPTR,
 180           this, handshaker->name(), handshaker.get(), index_);
 181     }
 182     handshaker->DoHandshake(acceptor_, &call_next_handshaker_, &args_);
 183   }
 184   ++index_;
 185   return is_shutdown_;
 186 }
 187
 188 void HandshakeManager::CallNextHandshakerFn(void* arg, grpc_error* error) {
 189   auto* mgr = static_cast<HandshakeManager*>(arg);
 190   bool done;
 191   {
 192     MutexLock lock(&mgr->mu_);
 193     done = mgr->CallNextHandshakerLocked(GRPC_ERROR_REF(error));
 194   }
 195   // If we're invoked the final callback, we won't be coming back
 196   // to this function, so we can release our reference to the
 197   // handshake manager.
 198   if (done) {
 199     mgr->Unref();
 200   }
 201 }
 202
 203 void HandshakeManager::OnTimeoutFn(void* arg, grpc_error* error) {
 204   auto* mgr = static_cast<HandshakeManager*>(arg);
 205   if (error == GRPC_ERROR_NONE) {  // Timer fired, rather than being cancelled
 206     mgr->Shutdown(GRPC_ERROR_CREATE_FROM_STATIC_STRING("Handshake timed out"));
 207   }
 208   mgr->Unref();
 209 }
 210
 211 void HandshakeManager::DoHandshake(grpc_endpoint* endpoint,
 212                                    const grpc_channel_args* channel_args,
 213                                    grpc_millis deadline,
 214                                    grpc_tcp_server_acceptor* acceptor,
 215                                    grpc_iomgr_cb_func on_handshake_done,
 216                                    void* user_data) {
 217   bool done;
 218   {
 219     MutexLock lock(&mu_);
 220     GPR_ASSERT(index_ == 0);
 221     GPR_ASSERT(!is_shutdown_);
 222     // Construct handshaker args.  These will be passed through all
 223     // handshakers and eventually be freed by the on_handshake_done callback.
 224     args_.endpoint = endpoint;
 225     args_.args = grpc_channel_args_copy(channel_args);
 226     args_.user_data = user_data;
 227     args_.read_buffer =
 228         static_cast<grpc_slice_buffer*>(gpr_malloc(sizeof(*args_.read_buffer)));
 229     grpc_slice_buffer_init(args_.read_buffer);
 230     if (acceptor != nullptr && acceptor->external_connection &&
 231         acceptor->pending_data != nullptr) {
 232       grpc_slice_buffer_swap(args_.read_buffer,
 233                              &(acceptor->pending_data->data.raw.slice_buffer));
 234     }
 235     // Initialize state needed for calling handshakers.
 236     acceptor_ = acceptor;
 237     GRPC_CLOSURE_INIT(&call_next_handshaker_,
 238                       &HandshakeManager::CallNextHandshakerFn, this,
 239                       grpc_schedule_on_exec_ctx);
 240     GRPC_CLOSURE_INIT(&on_handshake_done_, on_handshake_done, &args_,
 241                       grpc_schedule_on_exec_ctx);
 242     // Start deadline timer, which owns a ref.
 243     Ref().release();
 244     GRPC_CLOSURE_INIT(&on_timeout_, &HandshakeManager::OnTimeoutFn, this,
 245                       grpc_schedule_on_exec_ctx);
 246     grpc_timer_init(&deadline_timer_, deadline, &on_timeout_);
 247     // Start first handshaker, which also owns a ref.
 248     Ref().release();
 249     done = CallNextHandshakerLocked(GRPC_ERROR_NONE);
 250   }
 251   if (done) {
 252     Unref();
 253   }
 254 }
 255
 256 }  // namespace grpc_core
 257
 258 void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
 259                                 grpc_handshaker* handshaker) {
 260   // This is a transition method to aid the API change for handshakers.
 261   using namespace grpc_core;
 262   RefCountedPtr<Handshaker> refd_hs(static_cast<Handshaker*>(handshaker));
 263   mgr->Add(refd_hs);
 264 }