nghttpx: Add --backend-http1-connections-per-frontend option

src/shrpx.cc

@@ -712,6 +712,7 @@ void fill_default_config() {
   mod_config()->argc = 0;
   mod_config()->argv = nullptr;
   mod_config()->downstream_connections_per_host = 8;
+  mod_config()->downstream_connections_per_frontend = 0;
   mod_config()->listener_disable_timeout = 0.;
 }
 } // namespace
@@ -843,9 +844,20 @@ Performance:
   --backend-http1-connections-per-host=<NUM>
                      Set maximum  number of backend  concurrent HTTP/1
                      connections per host.   This option is meaningful
-                     when -s option is used.
+                     when -s option  is used.  To limit  the number of
+                     connections  per frontend  for default  mode, use
+                     --backend-http1-connections-per-frontend.
                      Default: )"
       << get_config()->downstream_connections_per_host << R"(
+  --backend-http1-connections-per-frontend=<NUM>
+                     Set maximum  number of backend  concurrent HTTP/1
+                     connections  per frontend.   This option  is only
+                     used for  default mode.   0 means  unlimited.  To
+                     limit  the number  of  connections  per host  for
+                     HTTP/2  or  SPDY  proxy  mode  (-s  option),  use
+                     --backend-http1-connections-per-host.
+                     Default: )"
+      << get_config()->downstream_connections_per_frontend << R"(
 
 Timeout:
   --frontend-http2-read-timeout=<SEC>
@@ -1215,6 +1227,8 @@ int main(int argc, char **argv) {
         {"listener-disable-timeout", required_argument, &flag, 64},
         {"strip-incoming-x-forwarded-for", no_argument, &flag, 65},
         {"accesslog-format", required_argument, &flag, 66},
+        {"backend-http1-connections-per-frontend", required_argument, &flag,
+         67},
         {nullptr, 0, nullptr, 0}};
 
     int option_index = 0;
@@ -1519,6 +1533,11 @@ int main(int argc, char **argv) {
         // --accesslog-format
         cmdcfgs.emplace_back(SHRPX_OPT_ACCESSLOG_FORMAT, optarg);
         break;
+      case 67:
+        // --backend-http1-connections-per-frontend
+        cmdcfgs.emplace_back(SHRPX_OPT_BACKEND_HTTP1_CONNECTIONS_PER_FRONTEND,
+                             optarg);
+        break;
       default:
         break;
       }

src/shrpx_config.cc

@@ -134,6 +134,8 @@ const char SHRPX_OPT_WORKER_FRONTEND_CONNECTIONS[] =
 const char SHRPX_OPT_NO_LOCATION_REWRITE[] = "no-location-rewrite";
 const char SHRPX_OPT_BACKEND_HTTP1_CONNECTIONS_PER_HOST[] =
     "backend-http1-connections-per-host";
+const char SHRPX_OPT_BACKEND_HTTP1_CONNECTIONS_PER_FRONTEND[] =
+    "backend-http1-connections-per-frontend";
 const char SHRPX_OPT_LISTENER_DISABLE_TIMEOUT[] = "listener-disable-timeout";
 
 namespace {
@@ -1030,6 +1032,24 @@ int parse_config(const char *opt, const char *optarg) {
     return 0;
   }
 
+  if (util::strieq(opt, SHRPX_OPT_BACKEND_HTTP1_CONNECTIONS_PER_FRONTEND)) {
+    int n;
+
+    if (parse_uint(&n, opt, optarg) != 0) {
+      return -1;
+    }
+
+    if (n < 0) {
+      LOG(ERROR) << opt << ": specify the integer more than or equal to 0";
+
+      return -1;
+    }
+
+    mod_config()->downstream_connections_per_frontend = n;
+
+    return 0;
+  }
+
   if (util::strieq(opt, SHRPX_OPT_LISTENER_DISABLE_TIMEOUT)) {
     return parse_timeval(&mod_config()->listener_disable_timeout, opt, optarg);
   }

src/shrpx_config.h

@@ -125,6 +125,7 @@ extern const char SHRPX_OPT_ADD_RESPONSE_HEADER[];
 extern const char SHRPX_OPT_WORKER_FRONTEND_CONNECTIONS[];
 extern const char SHRPX_OPT_NO_LOCATION_REWRITE[];
 extern const char SHRPX_OPT_BACKEND_HTTP1_CONNECTIONS_PER_HOST[];
+extern const char SHRPX_OPT_BACKEND_HTTP1_CONNECTIONS_PER_FRONTEND[];
 extern const char SHRPX_OPT_LISTENER_DISABLE_TIMEOUT[];
 
 union sockaddr_union {
@@ -230,6 +231,7 @@ struct Config {
   size_t http2_upstream_connection_window_bits;
   size_t http2_downstream_connection_window_bits;
   size_t downstream_connections_per_host;
+  size_t downstream_connections_per_frontend;
   // actual size of downstream_http_proxy_addr
   size_t downstream_http_proxy_addrlen;
   size_t read_rate;

src/shrpx_downstream_queue.cc

@@ -33,10 +33,11 @@ namespace shrpx {
 
 DownstreamQueue::HostEntry::HostEntry() : num_active(0) {}
 
-DownstreamQueue::DownstreamQueue(size_t conn_max_per_host)
+DownstreamQueue::DownstreamQueue(size_t conn_max_per_host, bool unified_host)
     : conn_max_per_host_(conn_max_per_host == 0
                              ? std::numeric_limits<size_t>::max()
-                             : conn_max_per_host) {}
+                             : conn_max_per_host),
+      unified_host_(unified_host) {}
 
 DownstreamQueue::~DownstreamQueue() {}
 
@@ -59,8 +60,19 @@ DownstreamQueue::find_host_entry(const std::string &host) {
   return (*itr).second;
 }
 
+const std::string &
+DownstreamQueue::make_host_key(const std::string &host) const {
+  static std::string empty_key;
+  return unified_host_ ? empty_key : host;
+}
+
+const std::string &
+DownstreamQueue::make_host_key(Downstream *downstream) const {
+  return make_host_key(downstream->get_request_http2_authority());
+}
+
 void DownstreamQueue::add_active(std::unique_ptr<Downstream> downstream) {
-  auto &ent = find_host_entry(downstream->get_request_http2_authority());
+  auto &ent = find_host_entry(make_host_key(downstream.get()));
   ++ent.num_active;
 
   auto stream_id = downstream->get_stream_id();
@@ -68,14 +80,14 @@ void DownstreamQueue::add_active(std::unique_ptr<Downstream> downstream) {
 }
 
 void DownstreamQueue::add_blocked(std::unique_ptr<Downstream> downstream) {
-  auto &ent = find_host_entry(downstream->get_request_http2_authority());
+  auto &ent = find_host_entry(make_host_key(downstream.get()));
   auto stream_id = downstream->get_stream_id();
   ent.blocked.insert(stream_id);
   blocked_downstreams_[stream_id] = std::move(downstream);
 }
 
 bool DownstreamQueue::can_activate(const std::string &host) const {
-  auto itr = host_entries_.find(host);
+  auto itr = host_entries_.find(make_host_key(host));
   if (itr == std::end(host_entries_)) {
     return true;
   }
@@ -119,7 +131,7 @@ DownstreamQueue::remove_and_pop_blocked(int32_t stream_id) {
 
   if (kv != std::end(active_downstreams_)) {
     auto downstream = pop_downstream(kv, active_downstreams_);
-    auto &host = downstream->get_request_http2_authority();
+    auto &host = make_host_key(downstream.get());
     auto &ent = find_host_entry(host);
     --ent.num_active;
 
@@ -148,7 +160,7 @@ DownstreamQueue::remove_and_pop_blocked(int32_t stream_id) {
 
   if (kv != std::end(blocked_downstreams_)) {
     auto downstream = pop_downstream(kv, blocked_downstreams_);
-    auto &host = downstream->get_request_http2_authority();
+    auto &host = make_host_key(downstream.get());
     auto &ent = find_host_entry(host);
     ent.blocked.erase(stream_id);
 

src/shrpx_downstream_queue.h

@@ -52,7 +52,7 @@ public:
   typedef std::map<std::string, HostEntry> HostEntryMap;
 
   // conn_max_per_host == 0 means no limit for downstream connection.
-  DownstreamQueue(size_t conn_max_per_host = 0);
+  DownstreamQueue(size_t conn_max_per_host = 0, bool unified_host = true);
   ~DownstreamQueue();
   void add_pending(std::unique_ptr<Downstream> downstream);
   void add_failure(std::unique_ptr<Downstream> downstream);
@@ -82,6 +82,8 @@ public:
   Downstream *find(int32_t stream_id);
   const DownstreamMap &get_active_downstreams() const;
   HostEntry &find_host_entry(const std::string &host);
+  const std::string &make_host_key(const std::string &host) const;
+  const std::string &make_host_key(Downstream *downstream) const;
 
   // Maximum number of concurrent connections to the same host.
   size_t conn_max_per_host_;
@@ -98,6 +100,9 @@ private:
   DownstreamMap active_downstreams_;
   // Downstream objects, blocked by conn_max_per_host_
   DownstreamMap blocked_downstreams_;
+  // true if downstream host is treated as the same.  Used for reverse
+  // proxying.
+  bool unified_host_;
 };
 
 } // namespace shrpx

src/shrpx_http2_upstream.cc

@@ -569,9 +569,13 @@ void settings_timeout_cb(struct ev_loop *loop, ev_timer *w, int revents) {
 } // namespace
 
 Http2Upstream::Http2Upstream(ClientHandler *handler)
-    : downstream_queue_(get_config()->http2_proxy
+    : downstream_queue_(
-                            ? get_config()->downstream_connections_per_host
+          get_config()->http2_proxy
-                            : 0),
+              ? get_config()->downstream_connections_per_host
+              : get_config()->downstream_proto == PROTO_HTTP
+                    ? get_config()->downstream_connections_per_frontend
+                    : 0,
+          !get_config()->http2_proxy),
       handler_(handler), session_(nullptr), data_pending_(nullptr),
       data_pendinglen_(0), deferred_(false) {
 

src/shrpx_spdy_upstream.cc

@@ -400,9 +400,13 @@ uint32_t infer_upstream_rst_stream_status_code(uint32_t downstream_error_code) {
 } // namespace
 
 SpdyUpstream::SpdyUpstream(uint16_t version, ClientHandler *handler)
-    : downstream_queue_(get_config()->http2_proxy
+    : downstream_queue_(
-                            ? get_config()->downstream_connections_per_host
+          get_config()->http2_proxy
-                            : 0),
+              ? get_config()->downstream_connections_per_host
+              : get_config()->downstream_proto == PROTO_HTTP
+                    ? get_config()->downstream_connections_per_frontend
+                    : 0,
+          !get_config()->http2_proxy),
       handler_(handler), session_(nullptr) {
   spdylay_session_callbacks callbacks;
   memset(&callbacks, 0, sizeof(callbacks));