src/shrpx_worker.h - master - githubmirror/nghttp2

src/shrpx_worker.h
488 строк15 KB
Tatsuhiro Tsujikawa
src: MSDMI part2
12 май 2026, 15:11
12 май 2026, 15:11f1aefd0
/*
 * nghttp2 - HTTP/2 C Library
 *
 * 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.
 */
#ifndef SHRPX_WORKER_H
#define SHRPX_WORKER_H

#include "shrpx.h"

#include <mutex>
#include <vector>
#include <random>
#include <unordered_map>
#include <deque>
#include <thread>
#include <queue>
#ifndef NOTHREADS
#  include <future>
#endif // !defined(NOTHREADS)
#include <expected>

#include "ssl_compat.h"

#ifdef NGHTTP2_OPENSSL_IS_WOLFSSL
#  include <wolfssl/options.h>
#  include <wolfssl/openssl/ssl.h>
#  include <wolfssl/openssl/err.h>
#else // !defined(NGHTTP2_OPENSSL_IS_WOLFSSL)
#  include <openssl/ssl.h>
#  include <openssl/err.h>
#endif // !defined(NGHTTP2_OPENSSL_IS_WOLFSSL)

#include <ev.h>

#include "shrpx_config.h"
#include "shrpx_downstream_connection_pool.h"
#include "memchunk.h"
#include "shrpx_tls.h"
#include "shrpx_live_check.h"
#include "shrpx_connect_blocker.h"
#include "shrpx_dns_tracker.h"
#ifdef ENABLE_HTTP3
#  include "shrpx_quic_connection_handler.h"
#  include "shrpx_quic.h"
#endif // defined(ENABLE_HTTP3)
#include "allocator.h"
#include "errors.h"

using namespace nghttp2;

namespace shrpx {

class Http2Session;
class ConnectBlocker;
struct UpstreamAddr;
class ConnectionHandler;
class AcceptHandler;
#ifdef ENABLE_HTTP3
class QUICListener;
#endif // defined(ENABLE_HTTP3)

#ifdef HAVE_MRUBY
namespace mruby {

class MRubyContext;

} // namespace mruby
#endif // defined(HAVE_MRUBY)

namespace tls {
class CertLookupTree;
} // namespace tls

struct WeightGroup;

struct DownstreamAddr {
  Address addr;
  // backend address.  If |host_unix| is true, this is UNIX domain
  // socket path.
  std::string_view host;
  std::string_view hostport;
  // backend port.  0 if |host_unix| is true.
  uint16_t port;
  // true if |host| contains UNIX domain socket path.
  bool host_unix;

  // sni field to send remote server if TLS is enabled.
  std::string_view sni;

  std::unique_ptr<ConnectBlocker> connect_blocker;
  std::unique_ptr<LiveCheck> live_check;
  // Connection pool for this particular address if session affinity
  // is enabled
  std::unique_ptr<DownstreamConnectionPool> dconn_pool;
  size_t fall;
  size_t rise;
  // Client side TLS session cache
  tls::TLSSessionCache tls_session_cache;
  // List of Http2Session which is not fully utilized (i.e., the
  // server advertised maximum concurrency is not reached).  We will
  // coalesce as much stream as possible in one Http2Session to fully
  // utilize TCP connection.
  DList<Http2Session> http2_extra_freelist;
  WeightGroup *wg;
  // total number of streams created in HTTP/2 connections for this
  // address.
  size_t num_dconn;
  // the sequence number of this address to randomize the order access
  // threads.
  size_t seq;
  // Application protocol used in this backend
  Proto proto;
  // cycle is used to prioritize this address.  Lower value takes
  // higher priority.
  uint32_t cycle;
  // penalty which is applied to the next cycle calculation.
  uint32_t pending_penalty;
  // Weight of this address inside a weight group.  Its range is [1,
  // 256], inclusive.
  uint32_t weight;
  // name of group which this address belongs to.
  std::string_view group;
  // Weight of the weight group which this address belongs to.  Its
  // range is [1, 256], inclusive.
  uint32_t group_weight;
  // affinity hash for this address.  It is assigned when strict
  // stickiness is enabled.
  uint32_t affinity_hash;
  // true if TLS is used in this backend
  bool tls;
  // true if dynamic DNS is enabled
  bool dns;
  // true if :scheme pseudo header field should be upgraded to secure
  // variant (e.g., "https") when forwarding request to a backend
  // connected by TLS connection.
  bool upgrade_scheme;
  // true if this address is queued.
  bool queued;
};

inline constexpr uint32_t MAX_DOWNSTREAM_ADDR_WEIGHT = 256;

struct DownstreamAddrEntry {
  DownstreamAddr *addr;
  size_t seq;
  uint32_t cycle;
};

struct DownstreamAddrEntryGreater {
  bool operator()(const DownstreamAddrEntry &lhs,
                  const DownstreamAddrEntry &rhs) const {
    auto d = lhs.cycle - rhs.cycle;
    if (d == 0) {
      return rhs.seq < lhs.seq;
    }
    return d <= 2 * MAX_DOWNSTREAM_ADDR_WEIGHT - 1;
  }
};

struct WeightGroup {
  std::priority_queue<DownstreamAddrEntry, std::vector<DownstreamAddrEntry>,
                      DownstreamAddrEntryGreater>
    pq;
  std::string_view name;
  size_t seq;
  uint32_t weight;
  uint32_t cycle;
  uint32_t pending_penalty;
  // true if this object is queued.
  bool queued;
};

struct WeightGroupEntry {
  WeightGroup *wg;
  size_t seq;
  uint32_t cycle;
};

struct WeightGroupEntryGreater {
  bool operator()(const WeightGroupEntry &lhs,
                  const WeightGroupEntry &rhs) const {
    auto d = lhs.cycle - rhs.cycle;
    if (d == 0) {
      return rhs.seq < lhs.seq;
    }
    return d <= 2 * MAX_DOWNSTREAM_ADDR_WEIGHT - 1;
  }
};

struct SharedDownstreamAddr {
  SharedDownstreamAddr() noexcept = default;
  SharedDownstreamAddr(const SharedDownstreamAddr &) = delete;
  SharedDownstreamAddr(SharedDownstreamAddr &&) = delete;
  SharedDownstreamAddr &operator=(const SharedDownstreamAddr &) = delete;
  SharedDownstreamAddr &operator=(SharedDownstreamAddr &&) = delete;

  BlockAllocator balloc{1024, 1024};
  std::vector<DownstreamAddr> addrs;
  std::vector<WeightGroup> wgs;
  std::priority_queue<WeightGroupEntry, std::vector<WeightGroupEntry>,
                      WeightGroupEntryGreater>
    pq;
  // Bunch of session affinity hash.  Only used if affinity ==
  // SessionAffinity::IP.
  std::vector<AffinityHash> affinity_hash;
  // Maps affinity hash of each DownstreamAddr to its index in addrs.
  // It is only assigned when strict stickiness is enabled.
  std::unordered_map<uint32_t, size_t> affinity_hash_map;
#ifdef HAVE_MRUBY
  std::shared_ptr<mruby::MRubyContext> mruby_ctx;
#endif // defined(HAVE_MRUBY)
  // Configuration for session affinity
  AffinityConfig affinity{SessionAffinity::NONE};
  // Session affinity
  // true if this group requires that client connection must be TLS,
  // and the request must be redirected to https URI.
  bool redirect_if_not_tls{};
  // true if a request should not be forwarded to a backend.
  bool dnf{};
  // Timeouts for backend connection.
  struct {
    ev_tstamp read;
    ev_tstamp write;
  } timeout{};
};

struct DownstreamAddrGroup {
  DownstreamAddrGroup() noexcept = default;
  ~DownstreamAddrGroup();

  DownstreamAddrGroup(const DownstreamAddrGroup &) = delete;
  DownstreamAddrGroup(DownstreamAddrGroup &&) = delete;
  DownstreamAddrGroup &operator=(const DownstreamAddrGroup &) = delete;
  DownstreamAddrGroup &operator=(DownstreamAddrGroup &&) = delete;

  ImmutableString pattern;
  std::shared_ptr<SharedDownstreamAddr> shared_addr;
  // true if this group is no longer used for new request.  If this is
  // true, the connection made using one of address in shared_addr
  // must not be pooled.
  bool retired{};
};

struct WorkerStat {
  size_t num_connections;
  size_t num_close_waits;
};

#ifdef ENABLE_HTTP3
struct QUICPacket {
  QUICPacket() noexcept = default;
  QUICPacket(size_t upstream_addr_index, const Address &remote_addr,
             const Address &local_addr, const ngtcp2_pkt_info &pi,
             std::span<const uint8_t> data)
    : upstream_addr_index{upstream_addr_index},
      remote_addr{remote_addr},
      local_addr{local_addr},
      pi{pi},
      data{std::ranges::begin(data), std::ranges::end(data)} {}
  size_t upstream_addr_index{};
  Address remote_addr{};
  Address local_addr{};
  ngtcp2_pkt_info pi{};
  std::vector<uint8_t> data;
};
#endif // defined(ENABLE_HTTP3)

enum class WorkerEventType {
  REOPEN_LOG = 0x02,
  GRACEFUL_SHUTDOWN = 0x03,
  REPLACE_DOWNSTREAM = 0x04,
#ifdef ENABLE_HTTP3
  QUIC_PKT_FORWARD = 0x05,
#endif // defined(ENABLE_HTTP3)
};

struct WorkerEvent {
  WorkerEventType type;
  std::shared_ptr<DownstreamConfig> downstreamconf;
#ifdef ENABLE_HTTP3
  std::unique_ptr<QUICPacket> quic_pkt;
#endif // defined(ENABLE_HTTP3)
};

class Worker {
public:
  Worker(struct ev_loop *loop, SSL_CTX *sv_ssl_ctx, SSL_CTX *cl_ssl_ctx,
         tls::CertLookupTree *cert_tree,
#ifdef ENABLE_HTTP3
         SSL_CTX *quic_sv_ssl_ctx, tls::CertLookupTree *quic_cert_tree,
         WorkerID wid,
#endif // defined(ENABLE_HTTP3)
         size_t index, const std::shared_ptr<TicketKeys> &ticket_keys,
         ConnectionHandler *conn_handler,
         std::shared_ptr<DownstreamConfig> downstreamconf);
  ~Worker();
  void run_async();
  void wait();
  void process_events();
  void send(WorkerEvent event);

  tls::CertLookupTree *get_cert_lookup_tree() const;
#ifdef ENABLE_HTTP3
  tls::CertLookupTree *get_quic_cert_lookup_tree() const;
#endif // defined(ENABLE_HTTP3)

  // These 2 functions make a lock m_ to get/set ticket keys
  // atomically.
  std::shared_ptr<TicketKeys> get_ticket_keys();
  void set_ticket_keys(std::shared_ptr<TicketKeys> ticket_keys);

  WorkerStat *get_worker_stat();
  struct ev_loop *get_loop() const;
  SSL_CTX *get_sv_ssl_ctx() const;
  SSL_CTX *get_cl_ssl_ctx() const;
#ifdef ENABLE_HTTP3
  SSL_CTX *get_quic_sv_ssl_ctx() const;
#endif // defined(ENABLE_HTTP3)

  void set_graceful_shutdown(bool f);
  bool get_graceful_shutdown() const;

  MemchunkPool *get_mcpool();
  void schedule_clear_mcpool();

  std::mt19937 &get_randgen();

#ifdef HAVE_MRUBY
  std::expected<void, Error> create_mruby_context();

  mruby::MRubyContext *get_mruby_context() const;
#endif // defined(HAVE_MRUBY)

  std::vector<std::shared_ptr<DownstreamAddrGroup>> &
  get_downstream_addr_groups();

  ConnectBlocker *get_connect_blocker() const;

  const DownstreamConfig *get_downstream_config() const;

  void
  replace_downstream_config(std::shared_ptr<DownstreamConfig> downstreamconf);

  ConnectionHandler *get_connection_handler() const;

  std::expected<void, Error> setup_server_socket();
  void drain_and_delete_listener();
  std::expected<void, Error> create_tcp_server_socket(UpstreamAddr &addr);
  void enable_listener();
  void disable_listener();
  void sleep_listener(ev_tstamp t);

#ifdef ENABLE_HTTP3
  QUICConnectionHandler *get_quic_connection_handler();

  std::expected<void, Error> setup_quic_server_socket();

  const WorkerID &get_worker_id() const;

#  ifdef HAVE_LIBBPF
  bool should_attach_bpf() const;

  bool should_update_bpf_map() const;

  uint32_t compute_sk_index() const;
#  endif // defined(HAVE_LIBBPF)

  std::expected<void, Error> create_quic_server_socket(UpstreamAddr &addr);

  // Returns a pointer to UpstreamAddr which matches |local_addr|.
  std::expected<const UpstreamAddr *, Error>
  find_quic_upstream_addr(const Address &local_addr);

  std::expected<void, Error>
  setup_quic_keying_materials(const std::unique_ptr<QUICKeyingMaterials> &qkms);

  [[nodiscard]] const std::unique_ptr<QUICKeyingMaterials> &
  get_quic_keying_materials() const {
    return quic_keying_materials_;
  }
#endif // defined(ENABLE_HTTP3)

  DNSTracker *get_dns_tracker();

  std::expected<void, Error> handle_connection(int fd, const sockaddr *addr,
                                               socklen_t addrlen,
                                               const UpstreamAddr *faddr);

private:
#ifndef NOTHREADS
  std::future<void> fut_;
#endif // !defined(NOTHREADS)
  // Unique index of this worker.
  size_t index_;
  std::mutex m_;
  std::deque<WorkerEvent> q_;
  std::mt19937 randgen_;
  ev_async w_;
  ev_timer mcpool_clear_timer_;
  ev_timer proc_wev_timer_;
  ev_timer disable_listener_timer_;
  MemchunkPool mcpool_;
  WorkerStat worker_stat_;
  DNSTracker dns_tracker_;

  std::vector<UpstreamAddr> upstream_addrs_;
  std::vector<std::unique_ptr<AcceptHandler>> listeners_;

#ifdef ENABLE_HTTP3
  WorkerID worker_id_;
  std::unique_ptr<QUICKeyingMaterials> quic_keying_materials_;
  std::vector<UpstreamAddr> quic_upstream_addrs_;
  std::vector<std::unique_ptr<QUICListener>> quic_listeners_;
#endif // defined(ENABLE_HTTP3)

  std::shared_ptr<DownstreamConfig> downstreamconf_;
#ifdef HAVE_MRUBY
  std::unique_ptr<mruby::MRubyContext> mruby_ctx_;
#endif // defined(HAVE_MRUBY)
  struct ev_loop *loop_;

  // Following fields are shared across threads if
  // get_config()->tls_ctx_per_worker == true.
  SSL_CTX *sv_ssl_ctx_;
  SSL_CTX *cl_ssl_ctx_;
  tls::CertLookupTree *cert_tree_;
  ConnectionHandler *conn_handler_;
#ifdef ENABLE_HTTP3
  SSL_CTX *quic_sv_ssl_ctx_;
  tls::CertLookupTree *quic_cert_tree_;

  QUICConnectionHandler quic_conn_handler_;
#endif // defined(ENABLE_HTTP3)

#ifdef HAVE_ATOMIC_STD_SHARED_PTR
  std::atomic<std::shared_ptr<TicketKeys>> ticket_keys_;
#else  // !defined(HAVE_ATOMIC_STD_SHARED_PTR)
  std::mutex ticket_keys_m_;
  std::shared_ptr<TicketKeys> ticket_keys_;
#endif // !defined(HAVE_ATOMIC_STD_SHARED_PTR)
  std::vector<std::shared_ptr<DownstreamAddrGroup>> downstream_addr_groups_;
  // Worker level blocker for downstream connection.  For example,
  // this is used when file descriptor is exhausted.
  std::unique_ptr<ConnectBlocker> connect_blocker_;

  bool graceful_shutdown_{};
};

// Selects group based on request's |hostport| and |path|.  |hostport|
// is the value taken from :authority or host header field, and may
// contain port.  The |path| may contain query part.  We require the
// catch-all pattern in place, so this function always selects one
// group.  The catch-all group index is given in |catch_all|.  All
// patterns are given in |groups|.
size_t match_downstream_addr_group(
  const RouterConfig &routerconfig, std::string_view hostport,
  std::string_view path,
  const std::vector<std::shared_ptr<DownstreamAddrGroup>> &groups,
  size_t catch_all, BlockAllocator &balloc);

// Calls this function if connecting to backend failed.  |raddr| is
// the actual address used to connect to backend, and it could be
// nullptr.  This function may schedule live check.
void downstream_failure(DownstreamAddr *addr, const Address *raddr);

} // namespace shrpx

#endif // !defined(SHRPX_WORKER_H)