src/core/lib/iomgr/tcp_server_custom.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 <grpc/support/port_platform.h>
  20
  21 #include "src/core/lib/iomgr/port.h"
  22
  23 #include <assert.h>
  24 #include <string.h>
  25
  26 #include <grpc/support/alloc.h>
  27 #include <grpc/support/log.h>
  28
  29 #include "src/core/lib/iomgr/error.h"
  30 #include "src/core/lib/iomgr/exec_ctx.h"
  31 #include "src/core/lib/iomgr/iomgr_custom.h"
  32 #include "src/core/lib/iomgr/sockaddr.h"
  33 #include "src/core/lib/iomgr/sockaddr_utils.h"
  34 #include "src/core/lib/iomgr/tcp_custom.h"
  35 #include "src/core/lib/iomgr/tcp_server.h"
  36
  37 extern grpc_core::TraceFlag grpc_tcp_trace;
  38
  39 extern grpc_socket_vtable* grpc_custom_socket_vtable;
  40
  41 /* one listening port */
  42 struct grpc_tcp_listener {
  43   grpc_tcp_server* server;
  44   unsigned port_index;
  45   int port;
  46
  47   grpc_custom_socket* socket;
  48
  49   /* linked list */
  50   struct grpc_tcp_listener* next;
  51
  52   bool closed;
  53 };
  54
  55 struct grpc_tcp_server {
  56   gpr_refcount refs;
  57
  58   /* Called whenever accept() succeeds on a server port. */
  59   grpc_tcp_server_cb on_accept_cb;
  60   void* on_accept_cb_arg;
  61
  62   int open_ports;
  63
  64   /* linked list of server ports */
  65   grpc_tcp_listener* head;
  66   grpc_tcp_listener* tail;
  67
  68   /* List of closures passed to shutdown_starting_add(). */
  69   grpc_closure_list shutdown_starting;
  70
  71   /* shutdown callback */
  72   grpc_closure* shutdown_complete;
  73
  74   bool shutdown;
  75
  76   grpc_resource_quota* resource_quota;
  77 };
  78
  79 static grpc_error* tcp_server_create(grpc_closure* shutdown_complete,
  80                                      const grpc_channel_args* args,
  81                                      grpc_tcp_server** server) {
  82   grpc_tcp_server* s = (grpc_tcp_server*)gpr_malloc(sizeof(grpc_tcp_server));
  83   s->resource_quota = grpc_resource_quota_create(nullptr);
  84   for (size_t i = 0; i < (args == nullptr ? 0 : args->num_args); i++) {
  85     if (0 == strcmp(GRPC_ARG_RESOURCE_QUOTA, args->args[i].key)) {
  86       if (args->args[i].type == GRPC_ARG_POINTER) {
  87         grpc_resource_quota_unref_internal(s->resource_quota);
  88         s->resource_quota = grpc_resource_quota_ref_internal(
  89             (grpc_resource_quota*)args->args[i].value.pointer.p);
  90       } else {
  91         grpc_resource_quota_unref_internal(s->resource_quota);
  92         gpr_free(s);
  93         return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  94             GRPC_ARG_RESOURCE_QUOTA " must be a pointer to a buffer pool");
  95       }
  96     }
  97   }
  98   gpr_ref_init(&s->refs, 1);
  99   s->on_accept_cb = nullptr;
 100   s->on_accept_cb_arg = nullptr;
 101   s->open_ports = 0;
 102   s->head = nullptr;
 103   s->tail = nullptr;
 104   s->shutdown_starting.head = nullptr;
 105   s->shutdown_starting.tail = nullptr;
 106   s->shutdown_complete = shutdown_complete;
 107   s->shutdown = false;
 108   *server = s;
 109   return GRPC_ERROR_NONE;
 110 }
 111
 112 static grpc_tcp_server* tcp_server_ref(grpc_tcp_server* s) {
 113   GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
 114   gpr_ref(&s->refs);
 115   return s;
 116 }
 117
 118 static void tcp_server_shutdown_starting_add(grpc_tcp_server* s,
 119                                              grpc_closure* shutdown_starting) {
 120   grpc_closure_list_append(&s->shutdown_starting, shutdown_starting,
 121                            GRPC_ERROR_NONE);
 122 }
 123
 124 static void finish_shutdown(grpc_tcp_server* s) {
 125   GPR_ASSERT(s->shutdown);
 126   if (s->shutdown_complete != nullptr) {
 127     GRPC_CLOSURE_SCHED(s->shutdown_complete, GRPC_ERROR_NONE);
 128   }
 129
 130   while (s->head) {
 131     grpc_tcp_listener* sp = s->head;
 132     s->head = sp->next;
 133     sp->next = nullptr;
 134     gpr_free(sp);
 135   }
 136   grpc_resource_quota_unref_internal(s->resource_quota);
 137   gpr_free(s);
 138 }
 139
 140 static void custom_close_callback(grpc_custom_socket* socket) {
 141   grpc_tcp_listener* sp = socket->listener;
 142   if (sp) {
 143     grpc_core::ExecCtx exec_ctx;
 144     sp->server->open_ports--;
 145     if (sp->server->open_ports == 0 && sp->server->shutdown) {
 146       finish_shutdown(sp->server);
 147     }
 148   }
 149   socket->refs--;
 150   if (socket->refs == 0) {
 151     grpc_custom_socket_vtable->destroy(socket);
 152     gpr_free(socket);
 153   }
 154 }
 155
 156 void grpc_custom_close_server_callback(grpc_tcp_listener* sp) {
 157   if (sp) {
 158     grpc_core::ExecCtx exec_ctx;
 159     sp->server->open_ports--;
 160     if (sp->server->open_ports == 0 && sp->server->shutdown) {
 161       finish_shutdown(sp->server);
 162     }
 163   }
 164 }
 165
 166 static void close_listener(grpc_tcp_listener* sp) {
 167   grpc_custom_socket* socket = sp->socket;
 168   if (!sp->closed) {
 169     sp->closed = true;
 170     grpc_custom_socket_vtable->close(socket, custom_close_callback);
 171   }
 172 }
 173
 174 static void tcp_server_destroy(grpc_tcp_server* s) {
 175   int immediately_done = 0;
 176   grpc_tcp_listener* sp;
 177
 178   GPR_ASSERT(!s->shutdown);
 179   s->shutdown = true;
 180
 181   if (s->open_ports == 0) {
 182     immediately_done = 1;
 183   }
 184   for (sp = s->head; sp; sp = sp->next) {
 185     close_listener(sp);
 186   }
 187
 188   if (immediately_done) {
 189     finish_shutdown(s);
 190   }
 191 }
 192
 193 static void tcp_server_unref(grpc_tcp_server* s) {
 194   GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
 195   if (gpr_unref(&s->refs)) {
 196     /* Complete shutdown_starting work before destroying. */
 197     grpc_core::ExecCtx exec_ctx;
 198     GRPC_CLOSURE_LIST_SCHED(&s->shutdown_starting);
 199     grpc_core::ExecCtx::Get()->Flush();
 200     tcp_server_destroy(s);
 201   }
 202 }
 203
 204 static void finish_accept(grpc_tcp_listener* sp, grpc_custom_socket* socket) {
 205   grpc_tcp_server_acceptor* acceptor =
 206       (grpc_tcp_server_acceptor*)gpr_malloc(sizeof(*acceptor));
 207   grpc_endpoint* ep = nullptr;
 208   grpc_resolved_address peer_name;
 209   char* peer_name_string;
 210   grpc_error* err;
 211
 212   peer_name_string = nullptr;
 213   memset(&peer_name, 0, sizeof(grpc_resolved_address));
 214   peer_name.len = GRPC_MAX_SOCKADDR_SIZE;
 215   err = grpc_custom_socket_vtable->getpeername(
 216       socket, (grpc_sockaddr*)&peer_name.addr, (int*)&peer_name.len);
 217   if (err == GRPC_ERROR_NONE) {
 218     peer_name_string = grpc_sockaddr_to_uri(&peer_name);
 219   } else {
 220     GRPC_LOG_IF_ERROR("getpeername error", err);
 221     GRPC_ERROR_UNREF(err);
 222   }
 223   if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) {
 224     if (peer_name_string) {
 225       gpr_log(GPR_INFO, "SERVER_CONNECT: %p accepted connection: %s",
 226               sp->server, peer_name_string);
 227     } else {
 228       gpr_log(GPR_INFO, "SERVER_CONNECT: %p accepted connection", sp->server);
 229     }
 230   }
 231   ep = custom_tcp_endpoint_create(socket, sp->server->resource_quota,
 232                                   peer_name_string);
 233   acceptor->from_server = sp->server;
 234   acceptor->port_index = sp->port_index;
 235   acceptor->fd_index = 0;
 236   sp->server->on_accept_cb(sp->server->on_accept_cb_arg, ep, nullptr, acceptor);
 237   gpr_free(peer_name_string);
 238 }
 239
 240 static void custom_accept_callback(grpc_custom_socket* socket,
 241                                    grpc_custom_socket* client,
 242                                    grpc_error* error);
 243
 244 static void custom_accept_callback(grpc_custom_socket* socket,
 245                                    grpc_custom_socket* client,
 246                                    grpc_error* error) {
 247   grpc_core::ExecCtx exec_ctx;
 248   grpc_tcp_listener* sp = socket->listener;
 249   if (error != GRPC_ERROR_NONE) {
 250     if (!sp->closed) {
 251       gpr_log(GPR_ERROR, "Accept failed: %s", grpc_error_string(error));
 252     }
 253     gpr_free(client);
 254     GRPC_ERROR_UNREF(error);
 255     return;
 256   }
 257   finish_accept(sp, client);
 258   if (!sp->closed) {
 259     grpc_custom_socket* new_socket =
 260         (grpc_custom_socket*)gpr_malloc(sizeof(grpc_custom_socket));
 261     new_socket->endpoint = nullptr;
 262     new_socket->listener = nullptr;
 263     new_socket->connector = nullptr;
 264     new_socket->refs = 1;
 265     grpc_custom_socket_vtable->accept(sp->socket, new_socket,
 266                                       custom_accept_callback);
 267   }
 268 }
 269
 270 static grpc_error* add_socket_to_server(grpc_tcp_server* s,
 271                                         grpc_custom_socket* socket,
 272                                         const grpc_resolved_address* addr,
 273                                         unsigned port_index,
 274                                         grpc_tcp_listener** listener) {
 275   grpc_tcp_listener* sp = nullptr;
 276   int port = -1;
 277   grpc_error* error;
 278   grpc_resolved_address sockname_temp;
 279
 280   // The last argument to uv_tcp_bind is flags
 281   error = grpc_custom_socket_vtable->bind(socket, (grpc_sockaddr*)addr->addr,
 282                                           addr->len, 0);
 283   if (error != GRPC_ERROR_NONE) {
 284     return error;
 285   }
 286
 287   error = grpc_custom_socket_vtable->listen(socket);
 288   if (error != GRPC_ERROR_NONE) {
 289     return error;
 290   }
 291
 292   sockname_temp.len = GRPC_MAX_SOCKADDR_SIZE;
 293   error = grpc_custom_socket_vtable->getsockname(
 294       socket, (grpc_sockaddr*)&sockname_temp.addr, (int*)&sockname_temp.len);
 295   if (error != GRPC_ERROR_NONE) {
 296     return error;
 297   }
 298
 299   port = grpc_sockaddr_get_port(&sockname_temp);
 300
 301   GPR_ASSERT(port >= 0);
 302   GPR_ASSERT(!s->on_accept_cb && "must add ports before starting server");
 303   sp = (grpc_tcp_listener*)gpr_zalloc(sizeof(grpc_tcp_listener));
 304   sp->next = nullptr;
 305   if (s->head == nullptr) {
 306     s->head = sp;
 307   } else {
 308     s->tail->next = sp;
 309   }
 310   s->tail = sp;
 311   sp->server = s;
 312   sp->socket = socket;
 313   sp->port = port;
 314   sp->port_index = port_index;
 315   sp->closed = false;
 316   s->open_ports++;
 317   *listener = sp;
 318
 319   return GRPC_ERROR_NONE;
 320 }
 321
 322 static grpc_error* tcp_server_add_port(grpc_tcp_server* s,
 323                                        const grpc_resolved_address* addr,
 324                                        int* port) {
 325   // This function is mostly copied from tcp_server_windows.c
 326   grpc_tcp_listener* sp = nullptr;
 327   grpc_custom_socket* socket;
 328   grpc_resolved_address addr6_v4mapped;
 329   grpc_resolved_address wildcard;
 330   grpc_resolved_address* allocated_addr = nullptr;
 331   grpc_resolved_address sockname_temp;
 332   unsigned port_index = 0;
 333   grpc_error* error = GRPC_ERROR_NONE;
 334   int family;
 335
 336   GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
 337
 338   if (s->tail != nullptr) {
 339     port_index = s->tail->port_index + 1;
 340   }
 341
 342   /* Check if this is a wildcard port, and if so, try to keep the port the same
 343      as some previously created listener. */
 344   if (grpc_sockaddr_get_port(addr) == 0) {
 345     for (sp = s->head; sp; sp = sp->next) {
 346       socket = sp->socket;
 347       sockname_temp.len = GRPC_MAX_SOCKADDR_SIZE;
 348       if (nullptr == grpc_custom_socket_vtable->getsockname(
 349                          socket, (grpc_sockaddr*)&sockname_temp.addr,
 350                          (int*)&sockname_temp.len)) {
 351         *port = grpc_sockaddr_get_port(&sockname_temp);
 352         if (*port > 0) {
 353           allocated_addr =
 354               (grpc_resolved_address*)gpr_malloc(sizeof(grpc_resolved_address));
 355           memcpy(allocated_addr, addr, sizeof(grpc_resolved_address));
 356           grpc_sockaddr_set_port(allocated_addr, *port);
 357           addr = allocated_addr;
 358           break;
 359         }
 360       }
 361     }
 362   }
 363
 364   if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
 365     addr = &addr6_v4mapped;
 366   }
 367
 368   /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */
 369   if (grpc_sockaddr_is_wildcard(addr, port)) {
 370     grpc_sockaddr_make_wildcard6(*port, &wildcard);
 371
 372     addr = &wildcard;
 373   }
 374
 375   if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) {
 376     char* port_string;
 377     grpc_sockaddr_to_string(&port_string, addr, 0);
 378     const char* str = grpc_error_string(error);
 379     if (port_string) {
 380       gpr_log(GPR_INFO, "SERVER %p add_port %s error=%s", s, port_string, str);
 381       gpr_free(port_string);
 382     } else {
 383       gpr_log(GPR_INFO, "SERVER %p add_port error=%s", s, str);
 384     }
 385   }
 386
 387   family = grpc_sockaddr_get_family(addr);
 388   socket = (grpc_custom_socket*)gpr_malloc(sizeof(grpc_custom_socket));
 389   socket->refs = 1;
 390   socket->endpoint = nullptr;
 391   socket->listener = nullptr;
 392   socket->connector = nullptr;
 393   grpc_custom_socket_vtable->init(socket, family);
 394
 395   if (error == GRPC_ERROR_NONE) {
 396     error = add_socket_to_server(s, socket, addr, port_index, &sp);
 397   }
 398   gpr_free(allocated_addr);
 399
 400   if (error != GRPC_ERROR_NONE) {
 401     grpc_error* error_out = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
 402         "Failed to add port to server", &error, 1);
 403     GRPC_ERROR_UNREF(error);
 404     error = error_out;
 405     *port = -1;
 406   } else {
 407     GPR_ASSERT(sp != nullptr);
 408     *port = sp->port;
 409   }
 410   socket->listener = sp;
 411   return error;
 412 }
 413
 414 static void tcp_server_start(grpc_tcp_server* server, grpc_pollset** pollsets,
 415                              size_t pollset_count,
 416                              grpc_tcp_server_cb on_accept_cb, void* cb_arg) {
 417   grpc_tcp_listener* sp;
 418   (void)pollsets;
 419   (void)pollset_count;
 420   GRPC_CUSTOM_IOMGR_ASSERT_SAME_THREAD();
 421   if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) {
 422     gpr_log(GPR_INFO, "SERVER_START %p", server);
 423   }
 424   GPR_ASSERT(on_accept_cb);
 425   GPR_ASSERT(!server->on_accept_cb);
 426   server->on_accept_cb = on_accept_cb;
 427   server->on_accept_cb_arg = cb_arg;
 428   for (sp = server->head; sp; sp = sp->next) {
 429     grpc_custom_socket* new_socket =
 430         (grpc_custom_socket*)gpr_malloc(sizeof(grpc_custom_socket));
 431     new_socket->endpoint = nullptr;
 432     new_socket->listener = nullptr;
 433     new_socket->connector = nullptr;
 434     new_socket->refs = 1;
 435     grpc_custom_socket_vtable->accept(sp->socket, new_socket,
 436                                       custom_accept_callback);
 437   }
 438 }
 439
 440 static unsigned tcp_server_port_fd_count(grpc_tcp_server* s,
 441                                          unsigned port_index) {
 442   return 0;
 443 }
 444
 445 static int tcp_server_port_fd(grpc_tcp_server* s, unsigned port_index,
 446                               unsigned fd_index) {
 447   return -1;
 448 }
 449
 450 static void tcp_server_shutdown_listeners(grpc_tcp_server* s) {
 451   for (grpc_tcp_listener* sp = s->head; sp; sp = sp->next) {
 452     if (!sp->closed) {
 453       sp->closed = true;
 454       grpc_custom_socket_vtable->close(sp->socket, custom_close_callback);
 455     }
 456   }
 457 }
 458
 459 grpc_tcp_server_vtable custom_tcp_server_vtable = {
 460     tcp_server_create,
 461     tcp_server_start,
 462     tcp_server_add_port,
 463     tcp_server_port_fd_count,
 464     tcp_server_port_fd,
 465     tcp_server_ref,
 466     tcp_server_shutdown_starting_add,
 467     tcp_server_unref,
 468     tcp_server_shutdown_listeners};
 469
 470 #ifdef GRPC_UV_TEST
 471 grpc_tcp_server_vtable* default_tcp_server_vtable = &custom_tcp_server_vtable;
 472 #endif