src/shrpx_worker.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_worker.h"
  26
  27 #ifdef HAVE_UNISTD_H
  28 #include <unistd.h>
  29 #endif // HAVE_UNISTD_H
  30
  31 #include <memory>
  32
  33 #include "shrpx_ssl.h"
  34 #include "shrpx_log.h"
  35 #include "shrpx_client_handler.h"
  36 #include "shrpx_http2_session.h"
  37 #include "shrpx_log_config.h"
  38 #include "shrpx_connect_blocker.h"
  39 #include "util.h"
  40 #include "template.h"
  41
  42 namespace shrpx {
  43
  44 namespace {
  45 void eventcb(struct ev_loop *loop, ev_async *w, int revents) {
  46   auto worker = static_cast<Worker *>(w->data);
  47   worker->process_events();
  48 }
  49 } // namespace
  50
  51 namespace {
  52 void mcpool_clear_cb(struct ev_loop *loop, ev_timer *w, int revents) {
  53   auto worker = static_cast<Worker *>(w->data);
  54   if (worker->get_worker_stat()->num_connections != 0) {
  55     return;
  56   }
  57   worker->get_mcpool()->clear();
  58 }
  59 } // namespace
  60
  61 Worker::Worker(struct ev_loop *loop, SSL_CTX *sv_ssl_ctx, SSL_CTX *cl_ssl_ctx,
  62                ssl::CertLookupTree *cert_tree,
  63                const std::shared_ptr<TicketKeys> &ticket_keys)
  64     : next_http2session_(0), loop_(loop), sv_ssl_ctx_(sv_ssl_ctx),
  65       cl_ssl_ctx_(cl_ssl_ctx), cert_tree_(cert_tree), ticket_keys_(ticket_keys),
  66       connect_blocker_(make_unique<ConnectBlocker>(loop_)),
  67       graceful_shutdown_(false) {
  68   ev_async_init(&w_, eventcb);
  69   w_.data = this;
  70   ev_async_start(loop_, &w_);
  71
  72   ev_timer_init(&mcpool_clear_timer_, mcpool_clear_cb, 0., 0.);
  73   mcpool_clear_timer_.data = this;
  74
  75   if (get_config()->downstream_proto == PROTO_HTTP2) {
  76     auto n = get_config()->http2_downstream_connections_per_worker;
  77     for (; n > 0; --n) {
  78       http2sessions_.push_back(make_unique<Http2Session>(
  79           loop_, cl_ssl_ctx, connect_blocker_.get(), this));
  80     }
  81   }
  82 }
  83
  84 Worker::~Worker() {
  85   ev_async_stop(loop_, &w_);
  86   ev_timer_stop(loop_, &mcpool_clear_timer_);
  87 }
  88
  89 void Worker::schedule_clear_mcpool() {
  90   // libev manual says: "If the watcher is already active nothing will
  91   // happen."  Since we don't change any timeout here, we don't have
  92   // to worry about querying ev_is_active.
  93   ev_timer_start(loop_, &mcpool_clear_timer_);
  94 }
  95
  96 void Worker::wait() {
  97 #ifndef NOTHREADS
  98   fut_.get();
  99 #endif // !NOTHREADS
 100 }
 101
 102 void Worker::run_async() {
 103 #ifndef NOTHREADS
 104   fut_ = std::async(std::launch::async, [this] {
 105     (void)reopen_log_files();
 106     ev_run(loop_);
 107   });
 108 #endif // !NOTHREADS
 109 }
 110
 111 void Worker::send(const WorkerEvent &event) {
 112   {
 113     std::lock_guard<std::mutex> g(m_);
 114
 115     q_.push_back(event);
 116   }
 117
 118   ev_async_send(loop_, &w_);
 119 }
 120
 121 void Worker::process_events() {
 122   std::deque<WorkerEvent> q;
 123   {
 124     std::lock_guard<std::mutex> g(m_);
 125     q.swap(q_);
 126   }
 127   for (auto &wev : q) {
 128     switch (wev.type) {
 129     case NEW_CONNECTION: {
 130       if (LOG_ENABLED(INFO)) {
 131         WLOG(INFO, this) << "WorkerEvent: client_fd=" << wev.client_fd
 132                          << ", addrlen=" << wev.client_addrlen;
 133       }
 134
 135       if (worker_stat_.num_connections >=
 136           get_config()->worker_frontend_connections) {
 137
 138         if (LOG_ENABLED(INFO)) {
 139           WLOG(INFO, this) << "Too many connections >= "
 140                            << get_config()->worker_frontend_connections;
 141         }
 142
 143         close(wev.client_fd);
 144
 145         break;
 146       }
 147
 148       auto client_handler = ssl::accept_connection(
 149           this, wev.client_fd, &wev.client_addr.sa, wev.client_addrlen);
 150       if (!client_handler) {
 151         if (LOG_ENABLED(INFO)) {
 152           WLOG(ERROR, this) << "ClientHandler creation failed";
 153         }
 154         close(wev.client_fd);
 155         break;
 156       }
 157
 158       if (LOG_ENABLED(INFO)) {
 159         WLOG(INFO, this) << "CLIENT_HANDLER:" << client_handler << " created ";
 160       }
 161
 162       break;
 163     }
 164     case RENEW_TICKET_KEYS:
 165       WLOG(NOTICE, this) << "Renew ticket keys: worker(" << this << ")";
 166
 167       ticket_keys_ = wev.ticket_keys;
 168
 169       break;
 170     case REOPEN_LOG:
 171       WLOG(NOTICE, this) << "Reopening log files: worker(" << this << ")";
 172
 173       reopen_log_files();
 174
 175       break;
 176     case GRACEFUL_SHUTDOWN:
 177       WLOG(NOTICE, this) << "Graceful shutdown commencing";
 178
 179       graceful_shutdown_ = true;
 180
 181       if (worker_stat_.num_connections == 0) {
 182         ev_break(loop_);
 183
 184         return;
 185       }
 186
 187       break;
 188     default:
 189       if (LOG_ENABLED(INFO)) {
 190         WLOG(INFO, this) << "unknown event type " << wev.type;
 191       }
 192     }
 193   }
 194 }
 195
 196 ssl::CertLookupTree *Worker::get_cert_lookup_tree() const { return cert_tree_; }
 197
 198 const std::shared_ptr<TicketKeys> &Worker::get_ticket_keys() const {
 199   return ticket_keys_;
 200 }
 201
 202 void Worker::set_ticket_keys(std::shared_ptr<TicketKeys> ticket_keys) {
 203   ticket_keys_ = std::move(ticket_keys);
 204 }
 205
 206 WorkerStat *Worker::get_worker_stat() { return &worker_stat_; }
 207
 208 DownstreamConnectionPool *Worker::get_dconn_pool() { return &dconn_pool_; }
 209
 210 Http2Session *Worker::next_http2_session() {
 211   if (http2sessions_.empty()) {
 212     return nullptr;
 213   }
 214
 215   auto res = http2sessions_[next_http2session_].get();
 216   ++next_http2session_;
 217   if (next_http2session_ >= http2sessions_.size()) {
 218     next_http2session_ = 0;
 219   }
 220
 221   return res;
 222 }
 223
 224 ConnectBlocker *Worker::get_connect_blocker() const {
 225   return connect_blocker_.get();
 226 }
 227
 228 struct ev_loop *Worker::get_loop() const {
 229   return loop_;
 230 }
 231
 232 SSL_CTX *Worker::get_sv_ssl_ctx() const { return sv_ssl_ctx_; }
 233
 234 SSL_CTX *Worker::get_cl_ssl_ctx() const { return cl_ssl_ctx_; }
 235
 236 void Worker::set_graceful_shutdown(bool f) { graceful_shutdown_ = f; }
 237
 238 bool Worker::get_graceful_shutdown() const { return graceful_shutdown_; }
 239
 240 MemchunkPool *Worker::get_mcpool() { return &mcpool_; }
 241
 242 } // namespace shrpx