nghttpx: Retain memory in Router

src/allocator.h

@@ -54,7 +54,35 @@ struct BlockAllocator {
         block_size(block_size),
         isolation_threshold(std::min(block_size, isolation_threshold)) {}
 
-  ~BlockAllocator() {
+  ~BlockAllocator() { destroy(); }
+
+  BlockAllocator(BlockAllocator &&other) noexcept
+      : retain(other.retain),
+        head(other.head),
+        block_size(other.block_size),
+        isolation_threshold(other.isolation_threshold) {
+    other.retain = nullptr;
+    other.head = nullptr;
+  }
+
+  BlockAllocator &operator=(BlockAllocator &&other) noexcept {
+    destroy();
+
+    retain = other.retain;
+    head = other.head;
+    block_size = other.block_size;
+    isolation_threshold = other.isolation_threshold;
+
+    other.retain = nullptr;
+    other.head = nullptr;
+
+    return *this;
+  }
+
+  BlockAllocator(const BlockAllocator &) = delete;
+  BlockAllocator &operator=(const BlockAllocator &) = delete;
+
+  void destroy() {
     for (auto mb = retain; mb;) {
       auto next = mb->next;
       delete[] reinterpret_cast<uint8_t *>(mb);
@@ -62,12 +90,6 @@ struct BlockAllocator {
     }
   }
 
-  BlockAllocator(BlockAllocator &&) = default;
-  BlockAllocator &operator=(BlockAllocator &&) = default;
-
-  BlockAllocator(const BlockAllocator &) = delete;
-  BlockAllocator &operator=(const BlockAllocator &) = delete;
-
   MemBlock *alloc_mem_block(size_t size) {
     auto block = new uint8_t[sizeof(MemBlock) + size];
     auto mb = reinterpret_cast<MemBlock *>(block);

src/shrpx_api_downstream_connection.cc

@@ -32,7 +32,7 @@
 namespace shrpx {
 
 APIDownstreamConnection::APIDownstreamConnection(Worker *worker)
-    : worker_(worker) {}
+    : worker_(worker), abandoned_(false) {}
 
 APIDownstreamConnection::~APIDownstreamConnection() {}
 
@@ -41,14 +41,6 @@ int APIDownstreamConnection::attach_downstream(Downstream *downstream) {
     DCLOG(INFO, this) << "Attaching to DOWNSTREAM:" << downstream;
   }
 
-  auto &req = downstream->request();
-
-  if (req.path != StringRef::from_lit("/api/v1/dynamicconfig")) {
-    // TODO this will return 503 error, which is not nice.  We'd like
-    // to use 404 in this case.
-    return -1;
-  }
-
   downstream_ = downstream;
 
   return 0;
@@ -61,10 +53,48 @@ void APIDownstreamConnection::detach_downstream(Downstream *downstream) {
   downstream_ = nullptr;
 }
 
-int APIDownstreamConnection::push_request_headers() { return 0; }
+int APIDownstreamConnection::send_reply(unsigned int http_status,
+                                        const StringRef &body) {
+  abandoned_ = true;
+
+  auto upstream = downstream_->get_upstream();
+
+  auto &resp = downstream_->response();
+
+  resp.http_status = http_status;
+
+  auto &balloc = downstream_->get_block_allocator();
+
+  auto content_length = util::make_string_ref_uint(balloc, body.size());
+
+  resp.fs.add_header_token(StringRef::from_lit("content-length"),
+                           content_length, false, http2::HD_CONTENT_LENGTH);
+
+  if (upstream->send_reply(downstream_, body.byte(), body.size()) != 0) {
+    return -1;
+  }
+
+  return 0;
+}
+
+int APIDownstreamConnection::push_request_headers() {
+  auto &req = downstream_->request();
+
+  if (req.path != StringRef::from_lit("/api/v1alpha1/backend/replace")) {
+    send_reply(404, StringRef::from_lit("404 Not Found"));
+
+    return 0;
+  }
+
+  return 0;
+}
 
 int APIDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
                                                     size_t datalen) {
+  if (abandoned_) {
+    return 0;
+  }
+
   auto output = downstream_->get_request_buf();
 
   // TODO limit the maximum payload size
@@ -78,33 +108,39 @@ int APIDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
 }
 
 int APIDownstreamConnection::end_upload_data() {
-  // TODO process request payload here
+  if (abandoned_) {
-  (void)worker_;
+    return 0;
+  }
 
-  auto upstream = downstream_->get_upstream();
   auto output = downstream_->get_request_buf();
-  auto &resp = downstream_->response();
-  struct iovec iov;
 
+  struct iovec iov;
   auto iovcnt = output->riovec(&iov, 1);
 
-  constexpr auto body = StringRef::from_lit("OK");
+  constexpr auto body = StringRef::from_lit("200 OK");
+  constexpr auto error_body = StringRef::from_lit("400 Bad Request");
 
   if (iovcnt == 0) {
-    resp.http_status = 200;
+    send_reply(200, body);
-
-    upstream->send_reply(downstream_, body.byte(), body.size());
 
     return 0;
   }
 
   Config config{};
   config.conn.downstream = std::make_shared<DownstreamConfig>();
+  const auto &downstreamconf = config.conn.downstream;
+
+  auto src = get_config()->conn.downstream;
+
+  downstreamconf->timeout = src->timeout;
+  downstreamconf->connections_per_host = src->connections_per_host;
+  downstreamconf->connections_per_frontend = src->connections_per_frontend;
+  downstreamconf->request_buffer_size = src->request_buffer_size;
+  downstreamconf->response_buffer_size = src->response_buffer_size;
+  downstreamconf->family = src->family;
 
   std::set<StringRef> include_set;
 
-  constexpr auto error_body = StringRef::from_lit("invalid configuration");
-
   for (auto first = reinterpret_cast<const uint8_t *>(iov.iov_base),
             last = first + iov.iov_len;
        first != last;) {
@@ -120,17 +156,13 @@ int APIDownstreamConnection::end_upload_data() {
 
     auto eq = std::find(first, eol, '=');
     if (eq == eol) {
-      resp.http_status = 500;
+      send_reply(400, error_body);
-
-      upstream->send_reply(downstream_, error_body.byte(), error_body.size());
       return 0;
     }
 
     if (parse_config(&config, StringRef{first, eq}, StringRef{eq + 1, eol},
                      include_set) != 0) {
-      resp.http_status = 500;
+      send_reply(400, error_body);
-
-      upstream->send_reply(downstream_, error_body.byte(), error_body.size());
       return 0;
     }
 
@@ -140,16 +172,13 @@ int APIDownstreamConnection::end_upload_data() {
   auto &tlsconf = get_config()->tls;
   if (configure_downstream_group(&config, get_config()->http2_proxy, true,
                                  tlsconf) != 0) {
-    resp.http_status = 500;
+    send_reply(400, error_body);
-    upstream->send_reply(downstream_, error_body.byte(), error_body.size());
     return 0;
   }
 
-  worker_->replace_downstream_config(config.conn.downstream);
+  worker_->replace_downstream_config(downstreamconf);
 
-  resp.http_status = 200;
+  send_reply(200, body);
-
-  upstream->send_reply(downstream_, body.byte(), body.size());
 
   return 0;
 }

src/shrpx_api_downstream_connection.h

@@ -56,8 +56,11 @@ public:
 
   virtual DownstreamAddrGroup *get_downstream_addr_group() const;
 
+  int send_reply(unsigned int http_status, const StringRef &body);
+
 private:
   Worker *worker_;
+  bool abandoned_;
 };
 
 } // namespace shrpx

src/shrpx_config.cc

@@ -786,7 +786,10 @@ int parse_mapping(Config *config, DownstreamAddrConfig addr,
     if (done) {
       continue;
     }
-    DownstreamAddrGroupConfig g(StringRef{pattern});
+
+    auto idx = addr_groups.size();
+    addr_groups.emplace_back(StringRef{pattern});
+    auto &g = addr_groups.back();
     g.addrs.push_back(addr);
 
     if (pattern[0] == '*') {
@@ -804,19 +807,16 @@ int parse_mapping(Config *config, DownstreamAddrConfig addr,
           [&host](const WildcardPattern &wp) { return wp.host == host; });
 
       if (it == std::end(wildcard_patterns)) {
-        wildcard_patterns.push_back(
+        wildcard_patterns.emplace_back(host);
-            {ImmutableString{std::begin(host), std::end(host)}});
 
         auto &router = wildcard_patterns.back().router;
-        router.add_route(path, addr_groups.size());
+        router.add_route(path, idx);
       } else {
-        (*it).router.add_route(path, addr_groups.size());
+        (*it).router.add_route(path, idx);
       }
     } else {
-      downstreamconf.router.add_route(StringRef{g.pattern}, addr_groups.size());
+      downstreamconf.router.add_route(StringRef{g.pattern}, idx);
     }
-
-    addr_groups.push_back(std::move(g));
   }
   return 0;
 }

src/shrpx_config.h

@@ -591,6 +591,9 @@ struct RateLimitConfig {
 // field.  router includes all path pattern sharing same wildcard
 // host.
 struct WildcardPattern {
+  WildcardPattern(const StringRef &host)
+      : host(std::begin(host), std::end(host)) {}
+
   ImmutableString host;
   Router router;
 };

src/shrpx_router.cc

@@ -35,7 +35,9 @@ RNode::RNode() : s(nullptr), len(0), index(-1) {}
 RNode::RNode(const char *s, size_t len, size_t index)
     : s(s), len(len), index(index) {}
 
-Router::Router() : root_{} {}
+Router::Router() : balloc_(1024, 1024), root_{} {}
+
+Router::~Router() {}
 
 namespace {
 RNode *find_next_node(const RNode *node, char c) {
@@ -62,7 +64,8 @@ void add_next_node(RNode *node, std::unique_ptr<RNode> new_node) {
 
 void Router::add_node(RNode *node, const char *pattern, size_t patlen,
                       size_t index) {
-  auto new_node = make_unique<RNode>(pattern, patlen, index);
+  auto pat = make_string_ref(balloc_, StringRef{pattern, patlen});
+  auto new_node = make_unique<RNode>(pat.c_str(), pat.size(), index);
   add_next_node(node, std::move(new_node));
 }
 

src/shrpx_router.h

@@ -30,6 +30,8 @@
 #include <vector>
 #include <memory>
 
+#include "allocator.h"
+
 namespace shrpx {
 
 struct RNode {
@@ -55,6 +57,12 @@ struct RNode {
 class Router {
 public:
   Router();
+  ~Router();
+  Router(Router &&) = default;
+  Router(const Router &) = delete;
+  Router &operator=(Router &&) = default;
+  Router &operator=(const Router &) = delete;
+
   // Adds route |pattern| with its |index|.
   bool add_route(const StringRef &pattern, size_t index);
   // Returns the matched index of pattern.  -1 if there is no match.
@@ -65,6 +73,7 @@ public:
   void dump() const;
 
 private:
+  BlockAllocator balloc_;
   // The root node of Patricia tree.  This is special node and its s
   // field is nulptr, and len field is 0.
   RNode root_;

src/shrpx_worker_test.cc

@@ -38,20 +38,22 @@
 namespace shrpx {
 
 void test_shrpx_worker_match_downstream_addr_group(void) {
-  auto groups = std::vector<DownstreamAddrGroup>();
+  auto groups = std::vector<std::shared_ptr<DownstreamAddrGroup>>();
   for (auto &s : {"nghttp2.org/", "nghttp2.org/alpha/bravo/",
                   "nghttp2.org/alpha/charlie", "nghttp2.org/delta%3A",
                   "www.nghttp2.org/", "[::1]/", "nghttp2.org/alpha/bravo/delta",
                   // Check that match is done in the single node
                   "example.com/alpha/bravo", "192.168.0.1/alpha/", "/golf/"}) {
-    groups.push_back(DownstreamAddrGroup{ImmutableString(s)});
+    auto g = std::make_shared<DownstreamAddrGroup>();
+    g->pattern = ImmutableString(s);
+    groups.push_back(std::move(g));
   }
 
   Router router;
 
   for (size_t i = 0; i < groups.size(); ++i) {
     auto &g = groups[i];
-    router.add_route(StringRef{g.pattern}, i);
+    router.add_route(StringRef{g->pattern}, i);
   }
 
   std::vector<WildcardPattern> wp;
@@ -176,15 +178,18 @@ void test_shrpx_worker_match_downstream_addr_group(void) {
                        StringRef::from_lit("/"), groups, 255));
 
   // Test for wildcard hosts
-  groups.push_back(
+  auto g1 = std::make_shared<DownstreamAddrGroup>();
-      DownstreamAddrGroup{ImmutableString::from_lit("git.nghttp2.org")});
+  g1->pattern = ImmutableString::from_lit("git.nghttp2.org");
-  groups.push_back(
+  groups.push_back(std::move(g1));
-      DownstreamAddrGroup{ImmutableString::from_lit(".nghttp2.org")});
 
-  wp.push_back({ImmutableString("git.nghttp2.org")});
+  auto g2 = std::make_shared<DownstreamAddrGroup>();
+  g2->pattern = ImmutableString::from_lit(".nghttp2.org");
+  groups.push_back(std::move(g2));
+
+  wp.emplace_back(StringRef::from_lit("git.nghttp2.org"));
   wp.back().router.add_route(StringRef::from_lit("/echo/"), 10);
 
-  wp.push_back({ImmutableString(".nghttp2.org")});
+  wp.emplace_back(StringRef::from_lit(".nghttp2.org"));
   wp.back().router.add_route(StringRef::from_lit("/echo/"), 11);
   wp.back().router.add_route(StringRef::from_lit("/echo/foxtrot"), 12);