Skip to content

http: use intrusive lists in ConnectionsList #65296

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mcollina wants to merge 1 commit into
base: main
Choose a base branch
from
+71 −59
Open

http: use intrusive lists in ConnectionsList #65296

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 71 additions & 59 deletions src/node_http_parser.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,8 +246,23 @@ 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 {
Expand All @@ -262,21 +277,13 @@ class ConnectionsList : public BaseObject {

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)
Expand All @@ -288,13 +295,14 @@ class ConnectionsList : public BaseObject {
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)
Expand All @@ -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;
Expand All @@ -325,7 +326,6 @@ class Parser : public AsyncWrap, public StreamListener {
status_message_.Reset();

if (connectionsList_ != nullptr) {
connectionsList_->Push(this);
connectionsList_->PushActive(this);
}

Expand Down Expand Up @@ -542,19 +542,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.

Expand Down Expand Up @@ -740,8 +733,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 {
Expand Down Expand Up @@ -1116,6 +1107,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_;

Expand Down Expand Up @@ -1143,18 +1136,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

void ConnectionsList::Push(Parser* parser) {
ListPushBack(&all_connections_, &parser->all_node_);
}

void ConnectionsList::Pop(Parser* parser) {
parser->all_node_.Remove();
}

return lhs->last_message_start_ < rhs->last_message_start_;
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) {
Expand All @@ -1172,8 +1181,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());
}

Expand All @@ -1189,8 +1200,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());
}
Expand All @@ -1208,8 +1221,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());
}

Expand Down Expand Up @@ -1253,14 +1268,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 (
Expand All @@ -1272,7 +1284,7 @@ void ConnectionsList::Expired(const FunctionCallbackInfo<Value>& args) {
) {
result.emplace_back(parser->object());

list->active_connections_.erase(parser);
parser->active_node_.Remove();
}
}

Expand Down
Loading