nghttp2/src/shrpx_worker.cc

261 lines
7.0 KiB
C++
Raw Normal View History

2012-06-05 18:26:04 +02:00
/*
2014-03-30 12:09:21 +02:00
* nghttp2 - HTTP/2 C Library
2012-06-05 18:26:04 +02:00
*
* Copyright (c) 2012 Tatsuhiro Tsujikawa
*
* Permission is hereby granted, free of charge, to any person obtaining
* a copy of this software and associated documentation files (the
* "Software"), to deal in the Software without restriction, including
* without limitation the rights to use, copy, modify, merge, publish,
* distribute, sublicense, and/or sell copies of the Software, and to
* permit persons to whom the Software is furnished to do so, subject to
* the following conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
* LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
* OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include "shrpx_worker.h"
2015-05-13 15:30:35 +02:00
#ifdef HAVE_UNISTD_H
2012-06-05 18:26:04 +02:00
#include <unistd.h>
2015-05-13 15:30:35 +02:00
#endif // HAVE_UNISTD_H
#include <memory>
2012-06-05 18:26:04 +02:00
#include "shrpx_ssl.h"
#include "shrpx_log.h"
#include "shrpx_client_handler.h"
#include "shrpx_http2_session.h"
#include "shrpx_log_config.h"
#include "shrpx_connect_blocker.h"
#include "util.h"
#include "template.h"
2012-06-05 18:26:04 +02:00
namespace shrpx {
namespace {
void eventcb(struct ev_loop *loop, ev_async *w, int revents) {
auto worker = static_cast<Worker *>(w->data);
worker->process_events();
2012-06-05 18:26:04 +02:00
}
} // namespace
namespace {
void mcpool_clear_cb(struct ev_loop *loop, ev_timer *w, int revents) {
auto worker = static_cast<Worker *>(w->data);
if (worker->get_worker_stat()->num_connections != 0) {
return;
}
worker->get_mcpool()->clear();
}
} // namespace
2015-02-11 11:18:41 +01:00
Worker::Worker(struct ev_loop *loop, SSL_CTX *sv_ssl_ctx, SSL_CTX *cl_ssl_ctx,
ssl::CertLookupTree *cert_tree,
const std::shared_ptr<TicketKeys> &ticket_keys)
: dconn_pool_(get_config()->downstream_addr_groups.size()),
worker_stat_(get_config()->downstream_addr_groups.size()),
dgrps_(get_config()->downstream_addr_groups.size()), loop_(loop),
sv_ssl_ctx_(sv_ssl_ctx), cl_ssl_ctx_(cl_ssl_ctx), cert_tree_(cert_tree),
ticket_keys_(ticket_keys),
connect_blocker_(make_unique<ConnectBlocker>(loop_)),
graceful_shutdown_(false) {
ev_async_init(&w_, eventcb);
w_.data = this;
ev_async_start(loop_, &w_);
ev_timer_init(&mcpool_clear_timer_, mcpool_clear_cb, 0., 0.);
mcpool_clear_timer_.data = this;
2015-02-11 11:18:41 +01:00
if (get_config()->downstream_proto == PROTO_HTTP2) {
auto n = get_config()->http2_downstream_connections_per_worker;
size_t group = 0;
for (auto &dgrp : dgrps_) {
auto m = n;
if (m == 0) {
m = get_config()->downstream_addr_groups[group].addrs.size();
}
for (; m; --m) {
dgrp.http2sessions.push_back(make_unique<Http2Session>(
loop_, cl_ssl_ctx, connect_blocker_.get(), this, group));
}
++group;
}
2015-02-11 11:18:41 +01:00
}
}
2012-06-05 18:26:04 +02:00
Worker::~Worker() {
ev_async_stop(loop_, &w_);
ev_timer_stop(loop_, &mcpool_clear_timer_);
}
void Worker::schedule_clear_mcpool() {
// libev manual says: "If the watcher is already active nothing will
// happen." Since we don't change any timeout here, we don't have
// to worry about querying ev_is_active.
ev_timer_start(loop_, &mcpool_clear_timer_);
}
void Worker::wait() {
#ifndef NOTHREADS
fut_.get();
#endif // !NOTHREADS
}
2015-02-11 11:18:41 +01:00
void Worker::run_async() {
#ifndef NOTHREADS
fut_ = std::async(std::launch::async, [this] {
(void)reopen_log_files();
ev_run(loop_);
});
#endif // !NOTHREADS
}
void Worker::send(const WorkerEvent &event) {
{
std::lock_guard<std::mutex> g(m_);
q_.push_back(event);
}
ev_async_send(loop_, &w_);
}
void Worker::process_events() {
std::vector<WorkerEvent> q;
{
std::lock_guard<std::mutex> g(m_);
q.swap(q_);
}
for (auto &wev : q) {
switch (wev.type) {
case NEW_CONNECTION: {
if (LOG_ENABLED(INFO)) {
WLOG(INFO, this) << "WorkerEvent: client_fd=" << wev.client_fd
<< ", addrlen=" << wev.client_addrlen;
}
2015-02-11 11:18:41 +01:00
if (worker_stat_.num_connections >=
get_config()->worker_frontend_connections) {
if (LOG_ENABLED(INFO)) {
WLOG(INFO, this) << "Too many connections >= "
<< get_config()->worker_frontend_connections;
}
close(wev.client_fd);
break;
}
auto client_handler = ssl::accept_connection(
2015-02-11 11:18:41 +01:00
this, wev.client_fd, &wev.client_addr.sa, wev.client_addrlen);
if (!client_handler) {
if (LOG_ENABLED(INFO)) {
WLOG(ERROR, this) << "ClientHandler creation failed";
}
close(wev.client_fd);
break;
}
if (LOG_ENABLED(INFO)) {
WLOG(INFO, this) << "CLIENT_HANDLER:" << client_handler << " created ";
}
break;
}
case RENEW_TICKET_KEYS:
WLOG(NOTICE, this) << "Renew ticket keys: worker(" << this << ")";
2015-02-11 11:18:41 +01:00
ticket_keys_ = wev.ticket_keys;
break;
case REOPEN_LOG:
WLOG(NOTICE, this) << "Reopening log files: worker(" << this << ")";
reopen_log_files();
break;
case GRACEFUL_SHUTDOWN:
WLOG(NOTICE, this) << "Graceful shutdown commencing";
graceful_shutdown_ = true;
2015-02-11 11:18:41 +01:00
if (worker_stat_.num_connections == 0) {
ev_break(loop_);
return;
}
break;
default:
if (LOG_ENABLED(INFO)) {
WLOG(INFO, this) << "unknown event type " << wev.type;
}
}
}
2012-06-05 18:26:04 +02:00
}
2015-02-11 11:18:41 +01:00
ssl::CertLookupTree *Worker::get_cert_lookup_tree() const { return cert_tree_; }
const std::shared_ptr<TicketKeys> &Worker::get_ticket_keys() const {
return ticket_keys_;
}
void Worker::set_ticket_keys(std::shared_ptr<TicketKeys> ticket_keys) {
ticket_keys_ = std::move(ticket_keys);
}
WorkerStat *Worker::get_worker_stat() { return &worker_stat_; }
DownstreamConnectionPool *Worker::get_dconn_pool() { return &dconn_pool_; }
Http2Session *Worker::next_http2_session(size_t group) {
auto &dgrp = dgrps_[group];
auto &http2sessions = dgrp.http2sessions;
if (http2sessions.empty()) {
return nullptr;
}
auto res = http2sessions[dgrp.next_http2session].get();
++dgrp.next_http2session;
if (dgrp.next_http2session >= http2sessions.size()) {
dgrp.next_http2session = 0;
}
return res;
}
2015-02-11 11:18:41 +01:00
ConnectBlocker *Worker::get_connect_blocker() const {
return connect_blocker_.get();
2015-02-11 11:18:41 +01:00
}
struct ev_loop *Worker::get_loop() const {
return loop_;
}
SSL_CTX *Worker::get_sv_ssl_ctx() const { return sv_ssl_ctx_; }
SSL_CTX *Worker::get_cl_ssl_ctx() const { return cl_ssl_ctx_; }
void Worker::set_graceful_shutdown(bool f) { graceful_shutdown_ = f; }
bool Worker::get_graceful_shutdown() const { return graceful_shutdown_; }
2015-04-07 15:13:01 +02:00
MemchunkPool *Worker::get_mcpool() { return &mcpool_; }
DownstreamGroup *Worker::get_dgrp(size_t group) {
assert(group < dgrps_.size());
return &dgrps_[group];
}
2012-06-05 18:26:04 +02:00
} // namespace shrpx