nghttpx: Pin frontend to specific HTTP/2 session object per group

src/shrpx_client_handler.cc

@@ -361,6 +361,11 @@ ClientHandler::ClientHandler(Worker *worker, int fd, SSL *ssl,
             get_config()->upstream_read_timeout, get_config()->write_rate,
             get_config()->write_burst, get_config()->read_rate,
             get_config()->read_burst, writecb, readcb, timeoutcb, this),
+      pinned_http2sessions_(
+          get_config()->downstream_proto == PROTO_HTTP2
+              ? make_unique<std::vector<ssize_t>>(
+                    get_config()->downstream_addr_groups.size(), -1)
+              : nullptr),
       ipaddr_(ipaddr), port_(port), worker_(worker),
       left_connhd_len_(NGHTTP2_CLIENT_MAGIC_LEN),
       should_close_after_write_(false) {
@@ -653,7 +658,15 @@ ClientHandler::get_downstream_connection(Downstream *downstream) {
     auto dconn_pool = worker_->get_dconn_pool();
 
     if (get_config()->downstream_proto == PROTO_HTTP2) {
-      auto http2session = worker_->next_http2_session(group);
+      Http2Session *http2session;
+      auto &pinned = (*pinned_http2sessions_)[group];
+      if (pinned == -1) {
+        http2session = worker_->next_http2_session(group);
+        pinned = http2session->get_index();
+      } else {
+        auto dgrp = worker_->get_dgrp(group);
+        http2session = dgrp->http2sessions[pinned].get();
+      }
       dconn = make_unique<Http2DownstreamConnection>(dconn_pool, http2session);
     } else {
       dconn =

src/shrpx_client_handler.h

@@ -134,6 +134,7 @@ private:
   Connection conn_;
   ev_timer reneg_shutdown_timer_;
   std::unique_ptr<Upstream> upstream_;
+  std::unique_ptr<std::vector<ssize_t>> pinned_http2sessions_;
   std::string ipaddr_;
   std::string port_;
   // The ALPN identifier negotiated for this connection.

src/shrpx_http2_session.cc

@@ -143,13 +143,13 @@ void writecb(struct ev_loop *loop, ev_io *w, int revents) {
 
 Http2Session::Http2Session(struct ev_loop *loop, SSL_CTX *ssl_ctx,
                            ConnectBlocker *connect_blocker, Worker *worker,
-                           size_t group)
+                           size_t group, size_t idx)
     : conn_(loop, -1, nullptr, get_config()->downstream_write_timeout,
             get_config()->downstream_read_timeout, 0, 0, 0, 0, writecb, readcb,
             timeoutcb, this),
       worker_(worker), connect_blocker_(connect_blocker), ssl_ctx_(ssl_ctx),
       session_(nullptr), data_pending_(nullptr), data_pendinglen_(0),
-      addr_idx_(0), group_(group), state_(DISCONNECTED),
+      addr_idx_(0), group_(group), index_(idx), state_(DISCONNECTED),
       connection_check_state_(CONNECTION_CHECK_NONE), flow_control_(false) {
 
   read_ = write_ = &Http2Session::noop;
@@ -1755,4 +1755,6 @@ size_t Http2Session::get_addr_idx() const { return addr_idx_; }
 
 size_t Http2Session::get_group() const { return group_; }
 
+size_t Http2Session::get_index() const { return index_; }
+
 } // namespace shrpx

src/shrpx_http2_session.h

@@ -58,7 +58,8 @@ struct StreamData {
 class Http2Session {
 public:
   Http2Session(struct ev_loop *loop, SSL_CTX *ssl_ctx,
-               ConnectBlocker *connect_blocker, Worker *worker, size_t group);
+               ConnectBlocker *connect_blocker, Worker *worker, size_t group,
+               size_t idx);
   ~Http2Session();
 
   int check_cert();
@@ -153,6 +154,8 @@ public:
 
   size_t get_group() const;
 
+  size_t get_index() const;
+
   enum {
     // Disconnected
     DISCONNECTED,
@@ -206,6 +209,9 @@ private:
   // index of get_config()->downstream_addrs this object uses
   size_t addr_idx_;
   size_t group_;
+  // index inside group, this is used to pin frontend to certain
+  // HTTP/2 backend for better throughput.
+  size_t index_;
   int state_;
   int connection_check_state_;
   bool flow_control_;

src/shrpx_worker.cc

@@ -83,9 +83,9 @@ Worker::Worker(struct ev_loop *loop, SSL_CTX *sv_ssl_ctx, SSL_CTX *cl_ssl_ctx,
       if (m == 0) {
         m = get_config()->downstream_addr_groups[group].addrs.size();
       }
-      for (; m; --m) {
+      for (size_t idx = 0; idx < m; ++idx) {
         dgrp.http2sessions.push_back(make_unique<Http2Session>(
-            loop_, cl_ssl_ctx, connect_blocker_.get(), this, group));
+            loop_, cl_ssl_ctx, connect_blocker_.get(), this, group, idx));
       }
       ++group;
     }