src/core/lib/http/httpcli.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 <grpc/support/port_platform.h>
  20
  21 #include "src/core/lib/http/httpcli.h"
  22
  23 #include <string.h>
  24
  25 #include <string>
  26
  27 #include "absl/strings/str_format.h"
  28
  29 #include <grpc/support/alloc.h>
  30 #include <grpc/support/log.h>
  31 #include <grpc/support/string_util.h>
  32
  33 #include "src/core/lib/address_utils/sockaddr_utils.h"
  34 #include "src/core/lib/gpr/string.h"
  35 #include "src/core/lib/gprpp/memory.h"
  36 #include "src/core/lib/http/format_request.h"
  37 #include "src/core/lib/http/parser.h"
  38 #include "src/core/lib/iomgr/endpoint.h"
  39 #include "src/core/lib/iomgr/iomgr_internal.h"
  40 #include "src/core/lib/iomgr/resolve_address.h"
  41 #include "src/core/lib/iomgr/tcp_client.h"
  42 #include "src/core/lib/slice/slice_internal.h"
  43
  44 struct internal_request {
  45   grpc_slice request_text;
  46   grpc_http_parser parser;
  47   grpc_resolved_addresses* addresses;
  48   size_t next_address;
  49   grpc_endpoint* ep;
  50   grpc_resource_quota* resource_quota;
  51   char* host;
  52   char* ssl_host_override;
  53   grpc_millis deadline;
  54   int have_read_byte;
  55   const grpc_httpcli_handshaker* handshaker;
  56   grpc_closure* on_done;
  57   grpc_httpcli_context* context;
  58   grpc_polling_entity* pollent;
  59   grpc_iomgr_object iomgr_obj;
  60   grpc_slice_buffer incoming;
  61   grpc_slice_buffer outgoing;
  62   grpc_closure on_read;
  63   grpc_closure done_write;
  64   grpc_closure connected;
  65   grpc_error_handle overall_error;
  66 };
  67 static grpc_httpcli_get_override g_get_override = nullptr;
  68 static grpc_httpcli_post_override g_post_override = nullptr;
  69
  70 static void plaintext_handshake(void* arg, grpc_endpoint* endpoint,
  71                                 const char* /*host*/, grpc_millis /*deadline*/,
  72                                 void (*on_done)(void* arg,
  73                                                 grpc_endpoint* endpoint)) {
  74   on_done(arg, endpoint);
  75 }
  76
  77 const grpc_httpcli_handshaker grpc_httpcli_plaintext = {"http",
  78                                                         plaintext_handshake};
  79
  80 void grpc_httpcli_context_init(grpc_httpcli_context* context) {
  81   context->pollset_set = grpc_pollset_set_create();
  82 }
  83
  84 void grpc_httpcli_context_destroy(grpc_httpcli_context* context) {
  85   grpc_pollset_set_destroy(context->pollset_set);
  86 }
  87
  88 static void next_address(internal_request* req, grpc_error_handle due_to_error);
  89
  90 static void finish(internal_request* req, grpc_error_handle error) {
  91   grpc_polling_entity_del_from_pollset_set(req->pollent,
  92                                            req->context->pollset_set);
  93   grpc_core::ExecCtx::Run(DEBUG_LOCATION, req->on_done, error);
  94   grpc_http_parser_destroy(&req->parser);
  95   if (req->addresses != nullptr) {
  96     grpc_resolved_addresses_destroy(req->addresses);
  97   }
  98   if (req->ep != nullptr) {
  99     grpc_endpoint_destroy(req->ep);
 100   }
 101   grpc_slice_unref_internal(req->request_text);
 102   gpr_free(req->host);
 103   gpr_free(req->ssl_host_override);
 104   grpc_iomgr_unregister_object(&req->iomgr_obj);
 105   grpc_slice_buffer_destroy_internal(&req->incoming);
 106   grpc_slice_buffer_destroy_internal(&req->outgoing);
 107   GRPC_ERROR_UNREF(req->overall_error);
 108   grpc_resource_quota_unref_internal(req->resource_quota);
 109   gpr_free(req);
 110 }
 111
 112 static void append_error(internal_request* req, grpc_error_handle error) {
 113   if (req->overall_error == GRPC_ERROR_NONE) {
 114     req->overall_error =
 115         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Failed HTTP/1 client request");
 116   }
 117   grpc_resolved_address* addr = &req->addresses->addrs[req->next_address - 1];
 118   std::string addr_text = grpc_sockaddr_to_uri(addr);
 119   req->overall_error = grpc_error_add_child(
 120       req->overall_error,
 121       grpc_error_set_str(error, GRPC_ERROR_STR_TARGET_ADDRESS,
 122                          grpc_slice_from_cpp_string(std::move(addr_text))));
 123 }
 124
 125 static void do_read(internal_request* req) {
 126   grpc_endpoint_read(req->ep, &req->incoming, &req->on_read, /*urgent=*/true);
 127 }
 128
 129 static void on_read(void* user_data, grpc_error_handle error) {
 130   internal_request* req = static_cast<internal_request*>(user_data);
 131   size_t i;
 132
 133   for (i = 0; i < req->incoming.count; i++) {
 134     if (GRPC_SLICE_LENGTH(req->incoming.slices[i])) {
 135       req->have_read_byte = 1;
 136       grpc_error_handle err = grpc_http_parser_parse(
 137           &req->parser, req->incoming.slices[i], nullptr);
 138       if (err != GRPC_ERROR_NONE) {
 139         finish(req, err);
 140         return;
 141       }
 142     }
 143   }
 144
 145   if (error == GRPC_ERROR_NONE) {
 146     do_read(req);
 147   } else if (!req->have_read_byte) {
 148     next_address(req, GRPC_ERROR_REF(error));
 149   } else {
 150     finish(req, grpc_http_parser_eof(&req->parser));
 151   }
 152 }
 153
 154 static void on_written(internal_request* req) { do_read(req); }
 155
 156 static void done_write(void* arg, grpc_error_handle error) {
 157   internal_request* req = static_cast<internal_request*>(arg);
 158   if (error == GRPC_ERROR_NONE) {
 159     on_written(req);
 160   } else {
 161     next_address(req, GRPC_ERROR_REF(error));
 162   }
 163 }
 164
 165 static void start_write(internal_request* req) {
 166   grpc_slice_ref_internal(req->request_text);
 167   grpc_slice_buffer_add(&req->outgoing, req->request_text);
 168   grpc_endpoint_write(req->ep, &req->outgoing, &req->done_write, nullptr);
 169 }
 170
 171 static void on_handshake_done(void* arg, grpc_endpoint* ep) {
 172   internal_request* req = static_cast<internal_request*>(arg);
 173
 174   if (!ep) {
 175     next_address(req, GRPC_ERROR_CREATE_FROM_STATIC_STRING(
 176                           "Unexplained handshake failure"));
 177     return;
 178   }
 179
 180   req->ep = ep;
 181   start_write(req);
 182 }
 183
 184 static void on_connected(void* arg, grpc_error_handle error) {
 185   internal_request* req = static_cast<internal_request*>(arg);
 186
 187   if (!req->ep) {
 188     next_address(req, GRPC_ERROR_REF(error));
 189     return;
 190   }
 191   req->handshaker->handshake(
 192       req, req->ep, req->ssl_host_override ? req->ssl_host_override : req->host,
 193       req->deadline, on_handshake_done);
 194 }
 195
 196 static void next_address(internal_request* req, grpc_error_handle error) {
 197   grpc_resolved_address* addr;
 198   if (error != GRPC_ERROR_NONE) {
 199     append_error(req, error);
 200   }
 201   if (req->next_address == req->addresses->naddrs) {
 202     finish(req,
 203            GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
 204                "Failed HTTP requests to all targets", &req->overall_error, 1));
 205     return;
 206   }
 207   addr = &req->addresses->addrs[req->next_address++];
 208   GRPC_CLOSURE_INIT(&req->connected, on_connected, req,
 209                     grpc_schedule_on_exec_ctx);
 210   grpc_tcp_client_connect(&req->connected, &req->ep,
 211                           grpc_slice_allocator_create(
 212                               req->resource_quota, grpc_sockaddr_to_uri(addr)),
 213                           req->context->pollset_set, nullptr, addr,
 214                           req->deadline);
 215 }
 216
 217 static void on_resolved(void* arg, grpc_error_handle error) {
 218   internal_request* req = static_cast<internal_request*>(arg);
 219   if (error != GRPC_ERROR_NONE) {
 220     finish(req, GRPC_ERROR_REF(error));
 221     return;
 222   }
 223   req->next_address = 0;
 224   next_address(req, GRPC_ERROR_NONE);
 225 }
 226
 227 static void internal_request_begin(grpc_httpcli_context* context,
 228                                    grpc_polling_entity* pollent,
 229                                    grpc_resource_quota* resource_quota,
 230                                    const grpc_httpcli_request* request,
 231                                    grpc_millis deadline, grpc_closure* on_done,
 232                                    grpc_httpcli_response* response,
 233                                    const char* name,
 234                                    const grpc_slice& request_text) {
 235   internal_request* req =
 236       static_cast<internal_request*>(gpr_malloc(sizeof(internal_request)));
 237   memset(req, 0, sizeof(*req));
 238   req->request_text = request_text;
 239   grpc_http_parser_init(&req->parser, GRPC_HTTP_RESPONSE, response);
 240   req->on_done = on_done;
 241   req->deadline = deadline;
 242   req->handshaker =
 243       request->handshaker ? request->handshaker : &grpc_httpcli_plaintext;
 244   req->context = context;
 245   req->pollent = pollent;
 246   req->overall_error = GRPC_ERROR_NONE;
 247   req->resource_quota = resource_quota;
 248   GRPC_CLOSURE_INIT(&req->on_read, on_read, req, grpc_schedule_on_exec_ctx);
 249   GRPC_CLOSURE_INIT(&req->done_write, done_write, req,
 250                     grpc_schedule_on_exec_ctx);
 251   grpc_slice_buffer_init(&req->incoming);
 252   grpc_slice_buffer_init(&req->outgoing);
 253   grpc_iomgr_register_object(&req->iomgr_obj, name);
 254   req->host = gpr_strdup(request->host);
 255   req->ssl_host_override = gpr_strdup(request->ssl_host_override);
 256
 257   GPR_ASSERT(pollent);
 258   grpc_polling_entity_add_to_pollset_set(req->pollent,
 259                                          req->context->pollset_set);
 260   grpc_resolve_address(
 261       request->host, req->handshaker->default_port, req->context->pollset_set,
 262       GRPC_CLOSURE_CREATE(on_resolved, req, grpc_schedule_on_exec_ctx),
 263       &req->addresses);
 264 }
 265
 266 void grpc_httpcli_get(grpc_httpcli_context* context,
 267                       grpc_polling_entity* pollent,
 268                       grpc_resource_quota* resource_quota,
 269                       const grpc_httpcli_request* request, grpc_millis deadline,
 270                       grpc_closure* on_done, grpc_httpcli_response* response) {
 271   if (g_get_override && g_get_override(request, deadline, on_done, response)) {
 272     grpc_resource_quota_unref_internal(resource_quota);
 273     return;
 274   }
 275   std::string name =
 276       absl::StrFormat("HTTP:GET:%s:%s", request->host, request->http.path);
 277   internal_request_begin(context, pollent, resource_quota, request, deadline,
 278                          on_done, response, name.c_str(),
 279                          grpc_httpcli_format_get_request(request));
 280 }
 281
 282 void grpc_httpcli_post(grpc_httpcli_context* context,
 283                        grpc_polling_entity* pollent,
 284                        grpc_resource_quota* resource_quota,
 285                        const grpc_httpcli_request* request,
 286                        const char* body_bytes, size_t body_size,
 287                        grpc_millis deadline, grpc_closure* on_done,
 288                        grpc_httpcli_response* response) {
 289   if (g_post_override && g_post_override(request, body_bytes, body_size,
 290                                          deadline, on_done, response)) {
 291     grpc_resource_quota_unref_internal(resource_quota);
 292     return;
 293   }
 294   std::string name =
 295       absl::StrFormat("HTTP:POST:%s:%s", request->host, request->http.path);
 296   internal_request_begin(
 297       context, pollent, resource_quota, request, deadline, on_done, response,
 298       name.c_str(),
 299       grpc_httpcli_format_post_request(request, body_bytes, body_size));
 300 }
 301
 302 void grpc_httpcli_set_override(grpc_httpcli_get_override get,
 303                                grpc_httpcli_post_override post) {
 304   g_get_override = get;
 305   g_post_override = post;
 306 }