src/shrpx_connection_handler.cc

   1 /*
   2  * nghttp2 - HTTP/2 C Library
   3  *
   4  * Copyright (c) 2012 Tatsuhiro Tsujikawa
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining
   7  * a copy of this software and associated documentation files (the
   8  * "Software"), to deal in the Software without restriction, including
   9  * without limitation the rights to use, copy, modify, merge, publish,
  10  * distribute, sublicense, and/or sell copies of the Software, and to
  11  * permit persons to whom the Software is furnished to do so, subject to
  12  * the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be
  15  * included in all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  21  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  22  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  23  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  24  */
  25 #include "shrpx_connection_handler.h"
  26
  27 #include <unistd.h>
  28
  29 #include <cerrno>
  30 #include <thread>
  31
  32 #include "shrpx_client_handler.h"
  33 #include "shrpx_ssl.h"
  34 #include "shrpx_worker.h"
  35 #include "shrpx_worker_config.h"
  36 #include "shrpx_config.h"
  37 #include "shrpx_http2_session.h"
  38 #include "shrpx_connect_blocker.h"
  39 #include "shrpx_downstream_connection.h"
  40 #include "shrpx_accept_handler.h"
  41 #include "util.h"
  42 #include "template.h"
  43
  44 using namespace nghttp2;
  45
  46 namespace shrpx {
  47
  48 namespace {
  49 void acceptor_disable_cb(struct ev_loop *loop, ev_timer *w, int revent) {
  50   auto h = static_cast<ConnectionHandler *>(w->data);
  51
  52   // If we are in graceful shutdown period, we must not enable
  53   // acceptors again.
  54   if (worker_config->graceful_shutdown) {
  55     return;
  56   }
  57
  58   h->enable_acceptor();
  59 }
  60 } // namespace
  61
  62 ConnectionHandler::ConnectionHandler(struct ev_loop *loop)
  63     : loop_(loop), sv_ssl_ctx_(nullptr), cl_ssl_ctx_(nullptr),
  64       // rate_limit_group_(bufferevent_rate_limit_group_new(
  65       //     evbase, get_config()->worker_rate_limit_cfg)),
  66       worker_stat_(make_unique<WorkerStat>()), worker_round_robin_cnt_(0) {
  67   ev_timer_init(&disable_acceptor_timer_, acceptor_disable_cb, 0., 0.);
  68   disable_acceptor_timer_.data = this;
  69 }
  70
  71 ConnectionHandler::~ConnectionHandler() {
  72   //  bufferevent_rate_limit_group_free(rate_limit_group_);
  73   ev_timer_stop(loop_, &disable_acceptor_timer_);
  74 }
  75
  76 void ConnectionHandler::create_ssl_context() {
  77   sv_ssl_ctx_ = ssl::setup_server_ssl_context();
  78   cl_ssl_ctx_ = ssl::setup_client_ssl_context();
  79 }
  80
  81 void ConnectionHandler::worker_reopen_log_files() {
  82   WorkerEvent wev;
  83
  84   memset(&wev, 0, sizeof(wev));
  85   wev.type = REOPEN_LOG;
  86
  87   for (auto &worker : workers_) {
  88     worker->send(wev);
  89   }
  90 }
  91
  92 void ConnectionHandler::worker_renew_ticket_keys(
  93     const std::shared_ptr<TicketKeys> &ticket_keys) {
  94   WorkerEvent wev;
  95
  96   memset(&wev, 0, sizeof(wev));
  97   wev.type = RENEW_TICKET_KEYS;
  98   wev.ticket_keys = ticket_keys;
  99
 100   for (auto &worker : workers_) {
 101     worker->send(wev);
 102   }
 103 }
 104
 105 void ConnectionHandler::create_worker_thread(size_t num) {
 106 #ifndef NOTHREADS
 107   assert(workers_.size() == 0);
 108
 109   for (size_t i = 0; i < num; ++i) {
 110     workers_.push_back(make_unique<Worker>(sv_ssl_ctx_, cl_ssl_ctx_,
 111                                            worker_config->cert_tree,
 112                                            worker_config->ticket_keys));
 113
 114     if (LOG_ENABLED(INFO)) {
 115       LLOG(INFO, this) << "Created thread #" << workers_.size() - 1;
 116     }
 117   }
 118 #endif // NOTHREADS
 119 }
 120
 121 void ConnectionHandler::join_worker() {
 122 #ifndef NOTHREADS
 123   int n = 0;
 124
 125   if (LOG_ENABLED(INFO)) {
 126     LLOG(INFO, this) << "Waiting for worker thread to join: n="
 127                      << workers_.size();
 128   }
 129
 130   for (auto &worker : workers_) {
 131     worker->wait();
 132     if (LOG_ENABLED(INFO)) {
 133       LLOG(INFO, this) << "Thread #" << n << " joined";
 134     }
 135     ++n;
 136   }
 137 #endif // NOTHREADS
 138 }
 139
 140 void ConnectionHandler::graceful_shutdown_worker() {
 141   if (get_config()->num_worker == 1) {
 142     return;
 143   }
 144
 145   WorkerEvent wev;
 146   memset(&wev, 0, sizeof(wev));
 147   wev.type = GRACEFUL_SHUTDOWN;
 148
 149   if (LOG_ENABLED(INFO)) {
 150     LLOG(INFO, this) << "Sending graceful shutdown signal to worker";
 151   }
 152
 153   for (auto &worker : workers_) {
 154
 155     worker->send(wev);
 156   }
 157 }
 158
 159 int ConnectionHandler::handle_connection(int fd, sockaddr *addr, int addrlen) {
 160   if (LOG_ENABLED(INFO)) {
 161     LLOG(INFO, this) << "Accepted connection. fd=" << fd;
 162   }
 163
 164   if (get_config()->num_worker == 1) {
 165
 166     if (worker_stat_->num_connections >=
 167         get_config()->worker_frontend_connections) {
 168
 169       if (LOG_ENABLED(INFO)) {
 170         LLOG(INFO, this) << "Too many connections >="
 171                          << get_config()->worker_frontend_connections;
 172       }
 173
 174       close(fd);
 175       return -1;
 176     }
 177
 178     auto client = ssl::accept_connection(loop_, sv_ssl_ctx_, fd, addr, addrlen,
 179                                          worker_stat_.get(), &dconn_pool_);
 180     if (!client) {
 181       LLOG(ERROR, this) << "ClientHandler creation failed";
 182
 183       close(fd);
 184       return -1;
 185     }
 186
 187     client->set_http2_session(http2session_.get());
 188     client->set_http1_connect_blocker(http1_connect_blocker_.get());
 189
 190     return 0;
 191   }
 192
 193   size_t idx = worker_round_robin_cnt_ % workers_.size();
 194   ++worker_round_robin_cnt_;
 195   WorkerEvent wev;
 196   memset(&wev, 0, sizeof(wev));
 197   wev.type = NEW_CONNECTION;
 198   wev.client_fd = fd;
 199   memcpy(&wev.client_addr, addr, addrlen);
 200   wev.client_addrlen = addrlen;
 201
 202   workers_[idx]->send(wev);
 203
 204   return 0;
 205 }
 206
 207 struct ev_loop *ConnectionHandler::get_loop() const {
 208   return loop_;
 209 }
 210
 211 void ConnectionHandler::create_http2_session() {
 212   http2session_ = make_unique<Http2Session>(loop_, cl_ssl_ctx_);
 213 }
 214
 215 void ConnectionHandler::create_http1_connect_blocker() {
 216   http1_connect_blocker_ = make_unique<ConnectBlocker>(loop_);
 217 }
 218
 219 const WorkerStat *ConnectionHandler::get_worker_stat() const {
 220   return worker_stat_.get();
 221 }
 222
 223 void ConnectionHandler::set_acceptor4(std::unique_ptr<AcceptHandler> h) {
 224   acceptor4_ = std::move(h);
 225 }
 226
 227 AcceptHandler *ConnectionHandler::get_acceptor4() const {
 228   return acceptor4_.get();
 229 }
 230
 231 void ConnectionHandler::set_acceptor6(std::unique_ptr<AcceptHandler> h) {
 232   acceptor6_ = std::move(h);
 233 }
 234
 235 AcceptHandler *ConnectionHandler::get_acceptor6() const {
 236   return acceptor6_.get();
 237 }
 238
 239 void ConnectionHandler::enable_acceptor() {
 240   if (acceptor4_) {
 241     acceptor4_->enable();
 242   }
 243
 244   if (acceptor6_) {
 245     acceptor6_->enable();
 246   }
 247 }
 248
 249 void ConnectionHandler::disable_acceptor() {
 250   if (acceptor4_) {
 251     acceptor4_->disable();
 252   }
 253
 254   if (acceptor6_) {
 255     acceptor6_->disable();
 256   }
 257 }
 258
 259 void ConnectionHandler::disable_acceptor_temporary(ev_tstamp t) {
 260   if (t == 0. || ev_is_active(&disable_acceptor_timer_)) {
 261     return;
 262   }
 263
 264   disable_acceptor();
 265
 266   ev_timer_set(&disable_acceptor_timer_, t, 0.);
 267   ev_timer_start(loop_, &disable_acceptor_timer_);
 268 }
 269
 270 void ConnectionHandler::accept_pending_connection() {
 271   if (acceptor4_) {
 272     acceptor4_->accept_connection();
 273   }
 274   if (acceptor6_) {
 275     acceptor6_->accept_connection();
 276   }
 277 }
 278
 279 } // namespace shrpx