OBS-URL: https://build.opensuse.org/package/show/devel:languages:nodejs/nodejs10?expand=0&rev=179
337 lines
12 KiB
Diff
337 lines
12 KiB
Diff
diff --git a/src/js_stream.cc b/src/js_stream.cc
|
|
index 4054e90b2e..e3d734c015 100644
|
|
--- a/src/js_stream.cc
|
|
+++ b/src/js_stream.cc
|
|
@@ -105,7 +105,7 @@ int JSStream::DoShutdown(ShutdownWrap* req_wrap) {
|
|
}
|
|
|
|
|
|
-int JSStream::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
+int JSStream::DoWrite(WriteWrap* w,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) {
|
|
@@ -122,7 +122,7 @@ int JSStream::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
}
|
|
|
|
Local<Value> argv[] = {
|
|
- w.get()->object(),
|
|
+ w->object(),
|
|
bufs_arr
|
|
};
|
|
|
|
diff --git a/src/js_stream.h b/src/js_stream.h
|
|
index bf0d15d462..6612e558ae 100644
|
|
--- a/src/js_stream.h
|
|
+++ b/src/js_stream.h
|
|
@@ -22,7 +22,7 @@ class JSStream : public AsyncWrap, public StreamBase {
|
|
int ReadStop() override;
|
|
|
|
int DoShutdown(ShutdownWrap* req_wrap) override;
|
|
- int DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
+ int DoWrite(WriteWrap* w,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) override;
|
|
diff --git a/src/node_file.h b/src/node_file.h
|
|
index b440c143b9..cbbb8b037d 100644
|
|
--- a/src/node_file.h
|
|
+++ b/src/node_file.h
|
|
@@ -287,7 +287,7 @@ class FileHandle : public AsyncWrap, public StreamBase {
|
|
ShutdownWrap* CreateShutdownWrap(v8::Local<v8::Object> object) override;
|
|
int DoShutdown(ShutdownWrap* req_wrap) override;
|
|
|
|
- int DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
+ int DoWrite(WriteWrap* w,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) override {
|
|
diff --git a/src/node_http2.cc b/src/node_http2.cc
|
|
index c81e202943..8cc1d705a9 100644
|
|
--- a/src/node_http2.cc
|
|
+++ b/src/node_http2.cc
|
|
@@ -2236,6 +2236,24 @@ int Http2Stream::SubmitPriority(nghttp2_priority_spec* prispec,
|
|
void Http2Stream::SubmitRstStream(const uint32_t code) {
|
|
CHECK(!this->IsDestroyed());
|
|
code_ = code;
|
|
+
|
|
+ auto is_stream_cancel = [](const uint32_t code) {
|
|
+ return code == NGHTTP2_CANCEL;
|
|
+ };
|
|
+
|
|
+ // If RST_STREAM frame is received with error code NGHTTP2_CANCEL,
|
|
+ // add it to the pending list and don't force purge the data. It is
|
|
+ // to avoids the double free error due to unwanted behavior of nghttp2.
|
|
+
|
|
+ // Add stream to the pending list only if it is received with scope
|
|
+ // below in the stack. The pending list may not get processed
|
|
+ // if RST_STREAM received is not in scope and added to the list
|
|
+ // causing endpoint to hang.
|
|
+ if (session_->is_in_scope() && is_stream_cancel(code)) {
|
|
+ session_->AddPendingRstStream(id_);
|
|
+ return;
|
|
+ }
|
|
+
|
|
// If possible, force a purge of any currently pending data here to make sure
|
|
// it is sent before closing the stream. If it returns non-zero then we need
|
|
// to wait until the current write finishes and try again to avoid nghttp2
|
|
@@ -2315,7 +2333,7 @@ int Http2Stream::ReadStop() {
|
|
// chunks of data have been flushed to the underlying nghttp2_session.
|
|
// Note that this does *not* mean that the data has been flushed
|
|
// to the socket yet.
|
|
-int Http2Stream::DoWrite(std::unique_ptr<WriteWrap>& req_wrap,
|
|
+int Http2Stream::DoWrite(WriteWrap* req_wrap,
|
|
uv_buf_t* bufs,
|
|
size_t nbufs,
|
|
uv_stream_t* send_handle) {
|
|
@@ -2330,7 +2348,7 @@ int Http2Stream::DoWrite(std::unique_ptr<WriteWrap>& req_wrap,
|
|
// Store the req_wrap on the last write info in the queue, so that it is
|
|
// only marked as finished once all buffers associated with it are finished.
|
|
queue_.emplace(nghttp2_stream_write {
|
|
- i == nbufs - 1 ? req_wrap.get() : nullptr,
|
|
+ i == nbufs - 1 ? req_wrap : nullptr,
|
|
bufs[i]
|
|
});
|
|
IncrementAvailableOutboundLength(bufs[i].len);
|
|
diff --git a/src/node_http2.h b/src/node_http2.h
|
|
index d1d523edcb..486a968392 100644
|
|
--- a/src/node_http2.h
|
|
+++ b/src/node_http2.h
|
|
@@ -568,7 +568,7 @@ class Http2Stream : public AsyncWrap,
|
|
|
|
AsyncWrap* GetAsyncWrap() override { return this; }
|
|
|
|
- int DoWrite(std::unique_ptr<WriteWrap>& w, uv_buf_t* bufs, size_t count,
|
|
+ int DoWrite(WriteWrap* w, uv_buf_t* bufs, size_t count,
|
|
uv_stream_t* send_handle) override;
|
|
|
|
void MemoryInfo(MemoryTracker* tracker) const override {
|
|
@@ -756,6 +756,22 @@ class Http2Session : public AsyncWrap, public StreamListener {
|
|
return (flags_ & SESSION_STATE_CLOSED) || session_ == nullptr;
|
|
}
|
|
|
|
+
|
|
+ // The changes are backported and exposes APIs to check the
|
|
+ // status flag of `Http2Session`
|
|
+#define IS_FLAG(name, flag) \
|
|
+ bool is_##name() const { return flags_ & flag; }
|
|
+
|
|
+ IS_FLAG(in_scope, SESSION_STATE_HAS_SCOPE)
|
|
+ IS_FLAG(write_scheduled, SESSION_STATE_WRITE_SCHEDULED)
|
|
+ IS_FLAG(closing, SESSION_STATE_CLOSING)
|
|
+ IS_FLAG(sending, SESSION_STATE_SENDING)
|
|
+ IS_FLAG(write_in_progress, SESSION_STATE_WRITE_IN_PROGRESS)
|
|
+ IS_FLAG(reading_stopped, SESSION_STATE_READING_STOPPED)
|
|
+ IS_FLAG(receive_paused, SESSION_STATE_NGHTTP2_RECV_PAUSED)
|
|
+
|
|
+#undef IS_FLAG
|
|
+
|
|
// Schedule a write if nghttp2 indicates it wants to write to the socket.
|
|
void MaybeScheduleWrite();
|
|
|
|
diff --git a/src/stream_base-inl.h b/src/stream_base-inl.h
|
|
index dca02ac76a..027b938d30 100644
|
|
--- a/src/stream_base-inl.h
|
|
+++ b/src/stream_base-inl.h
|
|
@@ -216,14 +216,14 @@ inline StreamWriteResult StreamBase::Write(
|
|
}
|
|
|
|
AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope(GetAsyncWrap());
|
|
- std::unique_ptr<WriteWrap> req_wrap{CreateWriteWrap(req_wrap_obj)};
|
|
+ WriteWrap* req_wrap = CreateWriteWrap(req_wrap_obj);
|
|
|
|
err = DoWrite(req_wrap, bufs, count, send_handle);
|
|
bool async = err == 0;
|
|
|
|
- if (!async && req_wrap != nullptr) {
|
|
+ if (!async) {
|
|
req_wrap->Dispose();
|
|
- req_wrap.release();
|
|
+ req_wrap = nullptr;
|
|
}
|
|
|
|
const char* msg = Error();
|
|
@@ -232,7 +232,7 @@ inline StreamWriteResult StreamBase::Write(
|
|
ClearError();
|
|
}
|
|
|
|
- return StreamWriteResult { async, err, req_wrap.release(), total_bytes };
|
|
+ return StreamWriteResult { async, err, req_wrap, total_bytes };
|
|
}
|
|
|
|
template <typename OtherBase>
|
|
diff --git a/src/stream_base.h b/src/stream_base.h
|
|
index 3e922a4ac4..65abd4dcf4 100644
|
|
--- a/src/stream_base.h
|
|
+++ b/src/stream_base.h
|
|
@@ -215,11 +215,10 @@ class StreamResource {
|
|
virtual int DoTryWrite(uv_buf_t** bufs, size_t* count);
|
|
// Perform a write of data, and either call req_wrap->Done() when finished
|
|
// and return 0, or return a libuv error code for synchronous failures.
|
|
- virtual int DoWrite(
|
|
- /* NOLINT (runtime/references) */ std::unique_ptr<WriteWrap>& w,
|
|
- uv_buf_t* bufs,
|
|
- size_t count,
|
|
- uv_stream_t* send_handle) = 0;
|
|
+ virtual int DoWrite(WriteWrap* w,
|
|
+ uv_buf_t* bufs,
|
|
+ size_t count,
|
|
+ uv_stream_t* send_handle) = 0;
|
|
|
|
// Returns true if the stream supports the `OnStreamWantsWrite()` interface.
|
|
virtual bool HasWantsWrite() const { return false; }
|
|
diff --git a/src/stream_wrap.cc b/src/stream_wrap.cc
|
|
index bd512e3349..10444fea4a 100644
|
|
--- a/src/stream_wrap.cc
|
|
+++ b/src/stream_wrap.cc
|
|
@@ -351,11 +351,11 @@ int LibuvStreamWrap::DoTryWrite(uv_buf_t** bufs, size_t* count) {
|
|
}
|
|
|
|
|
|
-int LibuvStreamWrap::DoWrite(std::unique_ptr<WriteWrap>& req_wrap,
|
|
+int LibuvStreamWrap::DoWrite(WriteWrap* req_wrap,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) {
|
|
- LibuvWriteWrap* w = static_cast<LibuvWriteWrap*>(req_wrap.get());
|
|
+ LibuvWriteWrap* w = static_cast<LibuvWriteWrap*>(req_wrap);
|
|
int r;
|
|
if (send_handle == nullptr) {
|
|
r = w->Dispatch(uv_write, stream(), bufs, count, AfterUvWrite);
|
|
diff --git a/src/stream_wrap.h b/src/stream_wrap.h
|
|
index 3c00d339af..98f0ca4ac4 100644
|
|
--- a/src/stream_wrap.h
|
|
+++ b/src/stream_wrap.h
|
|
@@ -51,7 +51,7 @@ class LibuvStreamWrap : public HandleWrap, public StreamBase {
|
|
// Resource implementation
|
|
int DoShutdown(ShutdownWrap* req_wrap) override;
|
|
int DoTryWrite(uv_buf_t** bufs, size_t* count) override;
|
|
- int DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
+ int DoWrite(WriteWrap* w,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) override;
|
|
diff --git a/src/tls_wrap.cc b/src/tls_wrap.cc
|
|
index 65ea8841b7..ce46e2163e 100644
|
|
--- a/src/tls_wrap.cc
|
|
+++ b/src/tls_wrap.cc
|
|
@@ -91,7 +91,8 @@ bool TLSWrap::InvokeQueued(int status, const char* error_str) {
|
|
return false;
|
|
|
|
if (current_write_ != nullptr) {
|
|
- WriteWrap* w = current_write_.release();
|
|
+ WriteWrap* w = current_write_;
|
|
+ current_write_ = nullptr;
|
|
w->Done(status, error_str);
|
|
}
|
|
|
|
@@ -616,7 +617,7 @@ void TLSWrap::ClearError() {
|
|
|
|
|
|
// Called by StreamBase::Write() to request async write of clear text into SSL.
|
|
-int TLSWrap::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
+int TLSWrap::DoWrite(WriteWrap* w,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) {
|
|
@@ -650,7 +651,7 @@ int TLSWrap::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
if (BIO_pending(enc_out_) == 0) {
|
|
Debug(this, "No pending encrypted output, writing to underlying stream");
|
|
CHECK_NULL(current_empty_write_);
|
|
- current_empty_write_ = w.get();
|
|
+ current_empty_write_ = w;
|
|
StreamWriteResult res =
|
|
underlying_stream()->Write(bufs, count, send_handle);
|
|
if (!res.async) {
|
|
@@ -665,7 +666,7 @@ int TLSWrap::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
|
|
// Store the current write wrap
|
|
CHECK_NULL(current_write_);
|
|
- current_write_ = std::move(w);
|
|
+ current_write_ = w;
|
|
|
|
// Write encrypted data to underlying stream and call Done().
|
|
if (length == 0) {
|
|
@@ -704,7 +705,7 @@ int TLSWrap::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
// If we stopped writing because of an error, it's fatal, discard the data.
|
|
if (!arg.IsEmpty()) {
|
|
Debug(this, "Got SSL error (%d), returning UV_EPROTO", err);
|
|
- current_write_.release();
|
|
+ current_write_ = nullptr;
|
|
return UV_EPROTO;
|
|
}
|
|
|
|
@@ -717,8 +718,6 @@ int TLSWrap::DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
// Write any encrypted/handshake output that may be ready.
|
|
EncOut();
|
|
|
|
- w.reset(current_write_.get());
|
|
-
|
|
return 0;
|
|
}
|
|
|
|
diff --git a/src/tls_wrap.h b/src/tls_wrap.h
|
|
index e2e748b859..bfcf07bbc8 100644
|
|
--- a/src/tls_wrap.h
|
|
+++ b/src/tls_wrap.h
|
|
@@ -67,7 +67,7 @@ class TLSWrap : public AsyncWrap,
|
|
ShutdownWrap* CreateShutdownWrap(
|
|
v8::Local<v8::Object> req_wrap_object) override;
|
|
int DoShutdown(ShutdownWrap* req_wrap) override;
|
|
- int DoWrite(std::unique_ptr<WriteWrap>& w,
|
|
+ int DoWrite(WriteWrap* w,
|
|
uv_buf_t* bufs,
|
|
size_t count,
|
|
uv_stream_t* send_handle) override;
|
|
@@ -170,7 +170,7 @@ class TLSWrap : public AsyncWrap,
|
|
// Waiting for ClearIn() to pass to SSL_write().
|
|
std::vector<char> pending_cleartext_input_;
|
|
size_t write_size_ = 0;
|
|
- std::unique_ptr<WriteWrap> current_write_ = nullptr;
|
|
+ WriteWrap* current_write_ = nullptr;
|
|
WriteWrap* current_empty_write_ = nullptr;
|
|
bool write_callback_scheduled_ = false;
|
|
bool started_ = false;
|
|
diff --git a/test/parallel/test-http2-cancel-while-client-reading.js b/test/parallel/test-http2-cancel-while-client-reading.js
|
|
new file mode 100644
|
|
index 0000000000..0605a02e11
|
|
--- /dev/null
|
|
+++ b/test/parallel/test-http2-cancel-while-client-reading.js
|
|
@@ -0,0 +1,36 @@
|
|
+'use strict';
|
|
+const common = require('../common');
|
|
+const fixtures = require('../common/fixtures');
|
|
+if (!common.hasCrypto) {
|
|
+ common.skip('missing crypto');
|
|
+}
|
|
+
|
|
+const http2 = require('http2');
|
|
+const key = fixtures.readKey('agent1-key.pem', 'binary');
|
|
+const cert = fixtures.readKey('agent1-cert.pem', 'binary');
|
|
+
|
|
+const server = http2.createSecureServer({ key, cert });
|
|
+
|
|
+let client_stream;
|
|
+
|
|
+server.on('stream', common.mustCall(function(stream) {
|
|
+ stream.resume();
|
|
+ stream.on('data', function(chunk) {
|
|
+ stream.write(chunk);
|
|
+ client_stream.pause();
|
|
+ client_stream.close(http2.constants.NGHTTP2_CANCEL);
|
|
+ });
|
|
+}));
|
|
+
|
|
+server.listen(0, function() {
|
|
+ const client = http2.connect(`https://localhost:${server.address().port}`,
|
|
+ { rejectUnauthorized: false }
|
|
+ );
|
|
+ client_stream = client.request({ ':method': 'POST' });
|
|
+ client_stream.on('close', common.mustCall(() => {
|
|
+ client.close();
|
|
+ server.close();
|
|
+ }));
|
|
+ client_stream.resume();
|
|
+ client_stream.write(Buffer.alloc(1024 * 1024));
|
|
+});
|