http: use intrusive lists in ConnectionsList

Every HTTP message was performing multiple erase and insert operations
on two std::set instances ordered by a mutating key, showing up as ~2%
of CPU cycles in a hello-world server profile due to red-black tree
rebalancing and node allocations.

Replace both sets with intrusive doubly-linked lists. Membership in the
list of all connections no longer changes per message, and updating the
active connections list is now O(1) with no allocations. Appending to
the tail keeps the active list ordered by last_message_start_ because
uv_hrtime() is monotonic.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

src/node_http_parser.cc

@@ -246,55 +246,63 @@ struct StringPtr {
   size_t size_ = 0;
 };
 
-struct ParserComparator {
-  bool operator()(const Parser* lhs, const Parser* rhs) const;
+// Intrusive doubly-linked list node, linked to itself when not in a list.
+struct ParserListNode {
+  ParserListNode* prev = this;
+  ParserListNode* next = this;
+
+  ParserListNode() = default;
+  ~ParserListNode() { Remove(); }
+
+  ParserListNode(const ParserListNode&) = delete;
+  ParserListNode& operator=(const ParserListNode&) = delete;
+
+  void Remove() {
+    prev->next = next;
+    next->prev = prev;
+    prev = this;
+    next = this;
+  }
 };
 
 class ConnectionsList : public BaseObject {
  public:
-    static void New(const FunctionCallbackInfo<Value>& args);
+  static void New(const FunctionCallbackInfo<Value>& args);
 
-    static void All(const FunctionCallbackInfo<Value>& args);
+  static void All(const FunctionCallbackInfo<Value>& args);
 
-    static void Idle(const FunctionCallbackInfo<Value>& args);
+  static void Idle(const FunctionCallbackInfo<Value>& args);
 
-    static void Active(const FunctionCallbackInfo<Value>& args);
+  static void Active(const FunctionCallbackInfo<Value>& args);
 
-    static void Expired(const FunctionCallbackInfo<Value>& args);
+  static void Expired(const FunctionCallbackInfo<Value>& args);
 
-    void Push(Parser* parser) {
-      all_connections_.insert(parser);
-    }
+  inline void Push(Parser* parser);
 
-    void Pop(Parser* parser) {
-      all_connections_.erase(parser);
-    }
+  inline void Pop(Parser* parser);
 
-    void PushActive(Parser* parser) {
-      active_connections_.insert(parser);
-    }
+  inline void PushActive(Parser* parser);
 
-    void PopActive(Parser* parser) {
-      active_connections_.erase(parser);
-    }
+  inline void PopActive(Parser* parser);
 
-    SET_NO_MEMORY_INFO()
-    SET_MEMORY_INFO_NAME(ConnectionsList)
-    SET_SELF_SIZE(ConnectionsList)
+  SET_NO_MEMORY_INFO()
+  SET_MEMORY_INFO_NAME(ConnectionsList)
+  SET_SELF_SIZE(ConnectionsList)
 
  private:
-    ConnectionsList(Environment* env, Local<Object> object)
+  ConnectionsList(Environment* env, Local<Object> object)
       : BaseObject(env, object) {
-        MakeWeak();
-      }
+    MakeWeak();
+  }
 
-    std::set<Parser*, ParserComparator> all_connections_;
-    std::set<Parser*, ParserComparator> active_connections_;
+  // active_connections_ is ordered by last_message_start_, as parsers are
+  // appended right after it is assigned from the monotonic uv_hrtime().
+  ParserListNode all_connections_;
+  ParserListNode active_connections_;
 };
 
 class Parser : public AsyncWrap, public StreamListener {
   friend class ConnectionsList;
-  friend struct ParserComparator;
 
  public:
   Parser(BindingData* binding_data, Local<Object> wrap)
@@ -308,13 +316,6 @@ class Parser : public AsyncWrap, public StreamListener {
   SET_SELF_SIZE(Parser)
 
   int on_message_begin() {
-    // Important: Pop from the lists BEFORE resetting the last_message_start_
-    // otherwise std::set.erase will fail.
-    if (connectionsList_ != nullptr) {
-      connectionsList_->Pop(this);
-      connectionsList_->PopActive(this);
-    }
-
     num_fields_ = num_values_ = 0;
     headers_completed_ = false;
     chunk_extensions_nread_ = 0;
@@ -325,7 +326,6 @@ class Parser : public AsyncWrap, public StreamListener {
     status_message_.Reset();
 
     if (connectionsList_ != nullptr) {
-      connectionsList_->Push(this);
       connectionsList_->PushActive(this);
     }
 
@@ -344,7 +344,6 @@ class Parser : public AsyncWrap, public StreamListener {
     return 0;
   }
 
-
   int on_url(const char* at, size_t length) {
     int rv = TrackHeader(length);
     if (rv != 0) {
@@ -542,19 +541,12 @@ class Parser : public AsyncWrap, public StreamListener {
   int on_message_complete() {
     HandleScope scope(env()->isolate());
 
-    // Important: Pop from the lists BEFORE resetting the last_message_start_
-    // otherwise std::set.erase will fail.
     if (connectionsList_ != nullptr) {
-      connectionsList_->Pop(this);
       connectionsList_->PopActive(this);
     }
 
     last_message_start_ = 0;
 
-    if (connectionsList_ != nullptr) {
-      connectionsList_->Push(this);
-    }
-
     if (num_fields_)
       Flush();  // Flush trailing HTTP headers.
 
@@ -740,8 +732,6 @@ class Parser : public AsyncWrap, public StreamListener {
       // server.timeout is left to the default value of zero.
       parser->last_message_start_ = uv_hrtime();
 
-      // Important: Push into the lists AFTER setting the last_message_start_
-      // otherwise std::set.erase will fail later.
       parser->connectionsList_->Push(parser);
       parser->connectionsList_->PushActive(parser);
     } else {
@@ -1116,6 +1106,8 @@ class Parser : public AsyncWrap, public StreamListener {
   uint64_t max_http_header_size_;
   uint64_t last_message_start_;
   ConnectionsList* connectionsList_;
+  ParserListNode all_node_;
+  ParserListNode active_node_;
 
   BaseObjectPtr<BindingData> binding_data_;
 
@@ -1143,18 +1135,34 @@ class Parser : public AsyncWrap, public StreamListener {
   static const llhttp_settings_t settings;
 };
 
-bool ParserComparator::operator()(const Parser* lhs, const Parser* rhs) const {
-  if (lhs->last_message_start_ == 0 && rhs->last_message_start_ == 0) {
-    // When both parsers are idle, guarantee strict order by
-    // comparing pointers as ints.
-    return lhs < rhs;
-  } else if (lhs->last_message_start_ == 0) {
-    return true;
-  } else if (rhs->last_message_start_ == 0) {
-    return false;
-  }
+namespace {
+
+// Append `node` at the tail of the list headed by `head`, unlinking it from
+// any list it is currently in.
+void ListPushBack(ParserListNode* head, ParserListNode* node) {
+  node->Remove();
+  node->prev = head->prev;
+  node->next = head;
+  head->prev->next = node;
+  head->prev = node;
+}
+
+}  // anonymous namespace
 
-  return lhs->last_message_start_ < rhs->last_message_start_;
+void ConnectionsList::Push(Parser* parser) {
+  ListPushBack(&all_connections_, &parser->all_node_);
+}
+
+void ConnectionsList::Pop(Parser* parser) {
+  parser->all_node_.Remove();
+}
+
+void ConnectionsList::PushActive(Parser* parser) {
+  ListPushBack(&active_connections_, &parser->active_node_);
+}
+
+void ConnectionsList::PopActive(Parser* parser) {
+  parser->active_node_.Remove();
 }
 
 void ConnectionsList::New(const FunctionCallbackInfo<Value>& args) {
@@ -1172,8 +1180,10 @@ void ConnectionsList::All(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&list, args.This());
 
   LocalVector<Value> result(isolate);
-  result.reserve(list->all_connections_.size());
-  for (auto parser : list->all_connections_) {
+  for (ParserListNode* node = list->all_connections_.next;
+       node != &list->all_connections_;
+       node = node->next) {
+    Parser* parser = ContainerOf(&Parser::all_node_, node);
     result.emplace_back(parser->object());
   }
 
@@ -1189,8 +1199,10 @@ void ConnectionsList::Idle(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&list, args.This());
 
   LocalVector<Value> result(isolate);
-  result.reserve(list->all_connections_.size());
-  for (auto parser : list->all_connections_) {
+  for (ParserListNode* node = list->all_connections_.next;
+       node != &list->all_connections_;
+       node = node->next) {
+    Parser* parser = ContainerOf(&Parser::all_node_, node);
     if (parser->last_message_start_ == 0 || !parser->received_data_) {
       result.emplace_back(parser->object());
     }
@@ -1208,8 +1220,10 @@ void ConnectionsList::Active(const FunctionCallbackInfo<Value>& args) {
   ASSIGN_OR_RETURN_UNWRAP(&list, args.This());
 
   LocalVector<Value> result(isolate);
-  result.reserve(list->active_connections_.size());
-  for (auto parser : list->active_connections_) {
+  for (ParserListNode* node = list->active_connections_.next;
+       node != &list->active_connections_;
+       node = node->next) {
+    Parser* parser = ContainerOf(&Parser::active_node_, node);
     result.emplace_back(parser->object());
   }
 
@@ -1253,14 +1267,11 @@ void ConnectionsList::Expired(const FunctionCallbackInfo<Value>& args) {
     return args.GetReturnValue().Set(Array::New(isolate, 0));
   }
 
-  auto iter = list->active_connections_.begin();
-  auto end = list->active_connections_.end();
-
   LocalVector<Value> result(isolate);
-  result.reserve(list->active_connections_.size());
-  while (iter != end) {
-    Parser* parser = *iter;
-    iter++;
+  ParserListNode* node = list->active_connections_.next;
+  while (node != &list->active_connections_) {
+    Parser* parser = ContainerOf(&Parser::active_node_, node);
+    node = node->next;
 
     // Check for expiration.
     if (
@@ -1272,7 +1283,7 @@ void ConnectionsList::Expired(const FunctionCallbackInfo<Value>& args) {
     ) {
       result.emplace_back(parser->object());
 
-      list->active_connections_.erase(parser);
+      parser->active_node_.Remove();
     }
   }