Skip to content
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

Jump to bottom

http2: minor C++ cleanups #16461

Merged
merged 5 commits into from
Oct 30, 2017
Merged

http2: minor C++ cleanups #16461

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
aeed4a1
http2: move uv_prepare handle to `Http2Session`
addaleax Oct 24, 2017
41aaacb
src: add `InternalCallbackScope` util constructor
addaleax Oct 24, 2017
0c90c9c
http2: track async state for sending
addaleax Oct 24, 2017
3ddd151
http2: remove unused assignment
addaleax Oct 24, 2017
c8a00fd
http2: make sessions garbage-collectible
addaleax Oct 24, 2017
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
6 changes: 6 additions & 0 deletions src/node.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1373,6 +1373,12 @@ CallbackScope::~CallbackScope() {
delete private_;
}

InternalCallbackScope::InternalCallbackScope(AsyncWrap* async_wrap)
: InternalCallbackScope(async_wrap->env(),
async_wrap->object(),
{ async_wrap->get_async_id(),
async_wrap->get_trigger_async_id() }) {}

InternalCallbackScope::InternalCallbackScope(Environment* env,
Local<Object> object,
const async_context& asyncContext,
Expand Down
54 changes: 51 additions & 3 deletions src/node_http2.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,8 +69,56 @@ Http2Options::Http2Options(Environment* env) {
}
}

void Http2Session::OnFreeSession() {
::delete this;

Http2Session::Http2Session(Environment* env,
Local<Object> wrap,
nghttp2_session_type type)
: AsyncWrap(env, wrap, AsyncWrap::PROVIDER_HTTP2SESSION),
StreamBase(env) {
MakeWeak<Http2Session>(this);

Http2Options opts(env);

padding_strategy_ = opts.GetPaddingStrategy();

Init(type, *opts);

// For every node::Http2Session instance, there is a uv_prepare_t handle
// whose callback is triggered on every tick of the event loop. When
// run, nghttp2 is prompted to send any queued data it may have stored.
prep_ = new uv_prepare_t();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

btw, definitely not for this PR, but down the road... I plan to refactor this so that there is one uv_prepare_t for N session instances, and loop through each on every uv_prepare_t tick... rather than registering one uv_prepare_t for each session. Will need to benchmark that tho...

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍 – I can leave a TODO if you want.

I’ve also wondered if it might worth hopping onto the same underlying mechanism that setImmediate uses…

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nah, don't need a todo.

... the same underlying mechanism that setImmediate uses...

I've considered that also but haven't looked into it yet.

uv_prepare_init(env->event_loop(), prep_);
prep_->data = static_cast<void*>(this);
uv_prepare_start(prep_, [](uv_prepare_t* t) {
Http2Session* session = static_cast<Http2Session*>(t->data);
HandleScope scope(session->env()->isolate());
Context::Scope context_scope(session->env()->context());

// Sending data may call arbitrary JS code, so keep track of
// async context.
InternalCallbackScope callback_scope(session);
session->SendPendingData();
});
}

Http2Session::~Http2Session() {
CHECK(persistent().IsEmpty());
Close();
}

void Http2Session::Close() {
if (!object().IsEmpty())
ClearWrap(object());
persistent().Reset();

this->Nghttp2Session::Close();
// Stop the loop
CHECK_EQ(uv_prepare_stop(prep_), 0);
auto prep_close = [](uv_handle_t* handle) {
delete reinterpret_cast<uv_prepare_t*>(handle);
};
uv_close(reinterpret_cast<uv_handle_t*>(prep_), prep_close);
prep_ = nullptr;
}

ssize_t Http2Session::OnMaxFrameSizePadding(size_t frameLen,
Expand Down Expand Up @@ -364,7 +412,7 @@ void Http2Session::Destroy(const FunctionCallbackInfo<Value>& args) {

if (!skipUnconsume)
session->Unconsume();
session->Free();
session->Close();
}

void Http2Session::Destroying(const FunctionCallbackInfo<Value>& args) {
Expand Down
29 changes: 9 additions & 20 deletions src/node_http2.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,24 +343,8 @@ class Http2Session : public AsyncWrap,
public:
Http2Session(Environment* env,
Local<Object> wrap,
nghttp2_session_type type) :
AsyncWrap(env, wrap, AsyncWrap::PROVIDER_HTTP2SESSION),
StreamBase(env) {
Wrap(object(), this);

Http2Options opts(env);

padding_strategy_ = opts.GetPaddingStrategy();

Init(env->event_loop(), type, *opts);
}

~Http2Session() override {
CHECK_EQ(false, persistent().IsEmpty());
ClearWrap(object());
persistent().Reset();
CHECK_EQ(true, persistent().IsEmpty());
}
nghttp2_session_type type);
~Http2Session() override;

static void OnStreamAllocImpl(size_t suggested_size,
uv_buf_t* buf,
Expand All @@ -369,9 +353,8 @@ class Http2Session : public AsyncWrap,
const uv_buf_t* bufs,
uv_handle_type pending,
void* ctx);
protected:
void OnFreeSession() override;

protected:
ssize_t OnMaxFrameSizePadding(size_t frameLength,
size_t maxPayloadLen);

Expand Down Expand Up @@ -449,6 +432,9 @@ class Http2Session : public AsyncWrap,
return 0;
}

uv_loop_t* event_loop() const override {
return env()->event_loop();
}
public:
void Consume(Local<External> external);
void Unconsume();
Expand Down Expand Up @@ -488,6 +474,8 @@ class Http2Session : public AsyncWrap,
return stream_buf_;
}

void Close() override;

private:
StreamBase* stream_;
StreamResource::Callback<StreamResource::AllocCb> prev_alloc_cb_;
Expand All @@ -496,6 +484,7 @@ class Http2Session : public AsyncWrap,

// use this to allow timeout tracking during long-lasting writes
uint32_t chunks_sent_since_last_write_ = 0;
uv_prepare_t* prep_ = nullptr;

char stream_buf_[kAllocBufferSize];
};
Expand Down
42 changes: 12 additions & 30 deletions src/node_http2_core-inl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,8 @@ inline ssize_t Nghttp2Session::OnStreamReadFD(nghttp2_session* session,
uv_fs_t read_req;

if (length > 0) {
numchars = uv_fs_read(handle->loop_,
// TODO(addaleax): Never use synchronous I/O on the main thread.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

re: the todo....

just keep in mind that nghttp2 requires that the data be provided synchronously here...

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm – from the nghttp2 header:

If the application wants to postpone DATA frames (e.g.,
asynchronous I/O, or reading data blocks for long time), it is
achieved by returning :enum:NGHTTP2_ERR_DEFERRED without reading
any data in this invocation. The library removes DATA frame from
the outgoing queue temporarily. To move back deferred DATA frame
to outgoing queue, call nghttp2_session_resume_data().

It even comes with instructions, so that sounds like it’s definitely supported?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes and no. I was being a bit simplistic. nghttp2_session_resume_data triggers this callback to be invoked again to resume the data flow. Within this method, when the data is available, it must be copied in synchronously. So to achieve what you're looking for it would require a bit more mechanism... e.g.

  1. attempt read,
    • no data? initiate uv_fs_read, return DEFER ... (wait for Step 2) ... keeping in mind that the buffer pointer nghttp2 hands us will be invalid when this function exits, so we'd need to allocate our own buffer for uv_fs_read.
    • got data? copy into nghttp2 provided buffer ...
      • all done? set EOF flag and return.
      • not all done? repeat step 1
  2. uv_fs_read callback calls session_resume_data ... flow returns back to step 1

This can be further complicated using the NO_COPY flag, which causes another callback to be called telling our code to send our buffers without the need to memcpy, but there are some complications with that also.

So, yes, it's certainly possible, and likely desirable, just non-trivial.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A bit more background... nghttp2 calls this method while it is attempting to serialize a DATA frame for transmission. Calls to this method are synchronous when ng's send or send_mem function is called. Essentially, nghttp2 goes through it's queue of pending frames and serializes each. When it comes to this method, it will synchronously call it over and over until it receives either a DEFER, an error, or an EOF response. If it receives a DEFER, it takes that stream temporarily out of it's send queue. Calling session_resume_data inserts it back into the queue, so that the next time send or mem_send is called, serialization of the DATA frame can be attempted again.

numchars = uv_fs_read(handle->event_loop(),
&read_req,
fd, &data, 1,
offset, nullptr);
Expand Down Expand Up @@ -541,11 +542,9 @@ inline void Nghttp2Session::SendPendingData() {
// Initialize the Nghttp2Session handle by creating and
// assigning the Nghttp2Session instance and associated
// uv_loop_t.
inline int Nghttp2Session::Init(uv_loop_t* loop,
const nghttp2_session_type type,
nghttp2_option* options,
nghttp2_mem* mem) {
loop_ = loop;
inline int Nghttp2Session::Init(const nghttp2_session_type type,
nghttp2_option* options,
nghttp2_mem* mem) {
session_type_ = type;
DEBUG_HTTP2("Nghttp2Session %s: initializing session\n", TypeName());
destroying_ = false;
Expand Down Expand Up @@ -581,41 +580,25 @@ inline int Nghttp2Session::Init(uv_loop_t* loop,
nghttp2_option_del(opts);
}

// For every node::Http2Session instance, there is a uv_prep_t handle
// whose callback is triggered on every tick of the event loop. When
// run, nghttp2 is prompted to send any queued data it may have stored.
uv_prepare_init(loop_, &prep_);
uv_prepare_start(&prep_, [](uv_prepare_t* t) {
Nghttp2Session* session = ContainerOf(&Nghttp2Session::prep_, t);
session->SendPendingData();
});
return ret;
}

inline void Nghttp2Session::MarkDestroying() {
destroying_ = true;
}

inline int Nghttp2Session::Free() {
#if defined(DEBUG) && DEBUG
CHECK(session_ != nullptr);
#endif
inline Nghttp2Session::~Nghttp2Session() {
Close();
}

inline void Nghttp2Session::Close() {
if (IsClosed())
return;
DEBUG_HTTP2("Nghttp2Session %s: freeing session\n", TypeName());
// Stop the loop
CHECK_EQ(uv_prepare_stop(&prep_), 0);
auto PrepClose = [](uv_handle_t* handle) {
Nghttp2Session* session =
ContainerOf(&Nghttp2Session::prep_,
reinterpret_cast<uv_prepare_t*>(handle));
session->OnFreeSession();
};
uv_close(reinterpret_cast<uv_handle_t*>(&prep_), PrepClose);
nghttp2_session_terminate_session(session_, NGHTTP2_NO_ERROR);
nghttp2_session_del(session_);
session_ = nullptr;
loop_ = nullptr;
DEBUG_HTTP2("Nghttp2Session %s: session freed\n", TypeName());
return 1;
}

// Write data received from the socket to the underlying nghttp2_session.
Expand Down Expand Up @@ -813,7 +796,6 @@ inline int Nghttp2Stream::SubmitFile(int fd,
DEBUG_HTTP2("Nghttp2Stream %d: submitting file\n", id_);
getTrailers_ = options & STREAM_OPTION_GET_TRAILERS;
nghttp2_data_provider prov;
prov.source.ptr = this;
prov.source.fd = fd;
prov.read_callback = Nghttp2Session::OnStreamReadFD;

Expand Down
15 changes: 9 additions & 6 deletions src/node_http2_core.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,13 +92,12 @@ class Nghttp2Session {
public:
// Initializes the session instance
inline int Init(
uv_loop_t*,
const nghttp2_session_type type = NGHTTP2_SESSION_SERVER,
nghttp2_option* options = nullptr,
nghttp2_mem* mem = nullptr);

// Frees this session instance
inline int Free();
inline ~Nghttp2Session();
inline void MarkDestroying();
bool IsDestroying() {
return destroying_;
Expand Down Expand Up @@ -141,6 +140,8 @@ class Nghttp2Session {
// Returns the nghttp2 library session
inline nghttp2_session* session() const { return session_; }

inline bool IsClosed() const { return session_ == nullptr; }

nghttp2_session_type type() const {
return session_type_;
}
Expand Down Expand Up @@ -175,7 +176,6 @@ class Nghttp2Session {
int error_code) {}
virtual ssize_t GetPadding(size_t frameLength,
size_t maxFrameLength) { return 0; }
virtual void OnFreeSession() {}
virtual void AllocateSend(uv_buf_t* buf) = 0;

virtual bool HasGetPaddingCallback() { return false; }
Expand All @@ -199,8 +199,13 @@ class Nghttp2Session {
virtual void OnTrailers(Nghttp2Stream* stream,
const SubmitTrailers& submit_trailers) {}

private:
inline void SendPendingData();

virtual uv_loop_t* event_loop() const = 0;

virtual void Close();

private:
inline void HandleHeadersFrame(const nghttp2_frame* frame);
inline void HandlePriorityFrame(const nghttp2_frame* frame);
inline void HandleDataFrame(const nghttp2_frame* frame);
Expand Down Expand Up @@ -281,8 +286,6 @@ class Nghttp2Session {
static Callbacks callback_struct_saved[2];

nghttp2_session* session_;
uv_loop_t* loop_;
uv_prepare_t prep_;
nghttp2_session_type session_type_;
std::unordered_map<int32_t, Nghttp2Stream*> streams_;
bool destroying_ = false;
Expand Down
2 changes: 2 additions & 0 deletions src/node_internals.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,8 @@ class InternalCallbackScope {
v8::Local<v8::Object> object,
const async_context& asyncContext,
ResourceExpectation expect = kRequireResource);
// Utility that can be used by AsyncWrap classes.
explicit InternalCallbackScope(AsyncWrap* async_wrap);
~InternalCallbackScope();
void Close();

Expand Down