diff --git a/benchmark/http2/full-duplex.js b/benchmark/http2/full-duplex.js new file mode 100644 index 000000000000..d2fec719f95a --- /dev/null +++ b/benchmark/http2/full-duplex.js @@ -0,0 +1,71 @@ +'use strict'; + +const common = require('../common.js'); +const fixtures = require('../../test/common/fixtures'); + +const bench = common.createBenchmark(main, { + n: [100], + streams: [2], + size: [4 * 1024 * 1024], + // Use the HTTP/2 protocol default. + window: [65535], +}, { + test: { size: 128 * 1024, window: 65535 }, +}); + +function main({ n, streams, size, window }) { + const http2 = require('http2'); + const payload = Buffer.alloc(size); + const server = http2.createSecureServer({ + key: fixtures.readKey('agent1-key.pem'), + cert: fixtures.readKey('agent1-cert.pem'), + settings: { initialWindowSize: window }, + }); + + let completed = 0; + let batches = 0; + + function onTransferComplete() { + if (++completed !== streams * 2) + return; + + if (++batches === n) { + // Report combined upload and download throughput in MiB/s. + bench.end(n * streams * size * 2 / (1024 * 1024)); + client.close(); + server.close(); + return; + } + + startBatch(); + } + + server.on('stream', (stream) => { + stream.resume(); + stream.on('end', onTransferComplete); + stream.respond(); + stream.end(payload); + }); + + let client; + function startBatch() { + completed = 0; + for (let i = 0; i < streams; i++) { + const request = client.request({ ':method': 'POST' }); + request.resume(); + request.on('end', onTransferComplete); + request.end(payload); + } + } + + server.listen(0, () => { + client = http2.connect(`https://localhost:${server.address().port}`, { + rejectUnauthorized: false, + settings: { initialWindowSize: window }, + }); + client.on('connect', () => { + bench.start(); + startBatch(); + }); + }); +} diff --git a/src/node_http2.cc b/src/node_http2.cc index 58f12c5561b5..4ce35dc37c23 100644 --- a/src/node_http2.cc +++ b/src/node_http2.cc @@ -981,45 +981,24 @@ ssize_t Http2Session::OnMaxFrameSizePadding(size_t frameLen, // quite expensive. This is a potential performance optimization target later. void Http2Session::ConsumeHTTP2Data() { CHECK_NOT_NULL(stream_buf_.base); - CHECK_LE(stream_buf_offset_, stream_buf_.len); - size_t read_len = stream_buf_.len - stream_buf_offset_; // multiple side effects. - Debug(this, "receiving %d bytes [wants data? %d]", - read_len, + Debug(this, + "receiving %d bytes [wants data? %d]", + stream_buf_.len, nghttp2_session_want_read(session_.get())); - set_receive_paused(false); custom_recv_error_code_ = nullptr; set_receiving(); ssize_t ret = - nghttp2_session_mem_recv(session_.get(), - reinterpret_cast(stream_buf_.base) + - stream_buf_offset_, - read_len); + nghttp2_session_mem_recv(session_.get(), + reinterpret_cast(stream_buf_.base), + stream_buf_.len); set_receiving(false); CHECK_NE(ret, NGHTTP2_ERR_NOMEM); CHECK_IMPLIES(custom_recv_error_code_ != nullptr, ret < 0); - if (is_receive_paused()) { - CHECK(is_reading_stopped()); - - CHECK_GT(ret, 0); - CHECK_LE(static_cast(ret), read_len); - - // Mark the remainder of the data as available for later consumption. - // Even if all bytes were received, a paused stream may delay the - // nghttp2_on_frame_recv_callback which may have an END_STREAM flag. - stream_buf_offset_ += ret; - // Still complete a Close() deferred during mem_recv; do not fall through - // to SendPendingData() here (paused receives historically skip that flush - // because a write may already be in progress). - MaybeFinishPendingClose(); - goto done; - } - // We are done processing the current input chunk. DecrementCurrentSessionMemory(stream_buf_.len); - stream_buf_offset_ = 0; stream_buf_ab_.Reset(); stream_buf_allocation_.reset(); stream_buf_ = uv_buf_init(nullptr, 0); @@ -1028,14 +1007,6 @@ void Http2Session::ConsumeHTTP2Data() { // not written after pending RST_STREAM frames. MaybeFinishPendingClose(); -done: - // Finish a Close() deferred above before flushing, so GOAWAY is not written - // after pending RST_STREAM frames. - if (is_close_pending() && !is_destroyed()) { - set_close_pending(false); - FinishClose(pending_close_code_, pending_close_socket_closed_); - } - // Send any data that was queued up while processing the received data. if (ret >= 0 && !is_destroyed()) { SendPendingData(); @@ -1485,15 +1456,6 @@ int Http2Session::OnDataChunkReceived(nghttp2_session* handle, } } while (len != 0); - // If we are currently waiting for a write operation to finish, we should - // tell nghttp2 that we want to wait before we process more input data. - if (session->is_write_in_progress()) { - CHECK(session->is_reading_stopped()); - session->set_receive_paused(); - Debug(session, "receive paused"); - return NGHTTP2_ERR_PAUSE; - } - return 0; } @@ -1576,7 +1538,6 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) { size_t offset = buf.base - session->stream_buf_.base; // Verify that the data offset is inside the current read buffer. - CHECK_GE(offset, session->stream_buf_offset_); CHECK_LE(offset, session->stream_buf_.len); CHECK_LE(offset + buf.len, session->stream_buf_.len); @@ -1891,11 +1852,6 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) { return; } - // If there is more incoming data queued up, consume it. - if (stream_buf_offset_ > 0) { - ConsumeHTTP2Data(); - } - if (!is_write_scheduled() && !is_destroyed()) { // Schedule a new write if nghttp2 wants to send data. MaybeScheduleWrite(); @@ -1942,7 +1898,7 @@ void Http2Session::MaybeStopReading() { if (is_reading_stopped() || is_closing()) return; int want_read = nghttp2_session_want_read(session_.get()); Debug(this, "wants read? %d", want_read); - if (want_read == 0 || is_write_in_progress()) { + if (want_read == 0) { set_reading_stopped(); stream_->ReadStop(); } @@ -2207,7 +2163,7 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { Context::Scope context_scope(env()->context()); Http2Scope h2scope(this); CHECK_NOT_NULL(stream_); - Debug(this, "receiving %d bytes, offset %d", nread, stream_buf_offset_); + Debug(this, "receiving %d bytes", nread); std::unique_ptr bs = env()->release_managed_buffer(buf_); // Only pass data on if nread > 0 @@ -2222,8 +2178,11 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { statistics_.data_received += nread; - if (stream_buf_offset_ == 0 && static_cast(nread) != bs->ByteLength()) - [[likely]] { + // ConsumeHTTP2Data() always consumes the whole chunk, so there is never a + // partially processed buffer left over from a previous read. + DCHECK_NULL(stream_buf_.base); + + if (static_cast(nread) != bs->ByteLength()) [[likely]] { // Shrink to the actual amount of used data. std::unique_ptr old_bs = std::move(bs); bs = ArrayBuffer::NewBackingStore( @@ -2231,31 +2190,6 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { nread, BackingStoreInitializationMode::kUninitialized); memcpy(bs->Data(), old_bs->Data(), nread); - } else { - // This is a very unlikely case, and should only happen if the ReadStart() - // call in OnStreamAfterWrite() immediately provides data. If that does - // happen, we concatenate the data we received with the already-stored - // pending input data, slicing off the already processed part. - size_t pending_len = stream_buf_.len - stream_buf_offset_; - std::unique_ptr new_bs = ArrayBuffer::NewBackingStore( - env()->isolate(), - pending_len + nread, - BackingStoreInitializationMode::kUninitialized); - memcpy(static_cast(new_bs->Data()), - stream_buf_.base + stream_buf_offset_, - pending_len); - memcpy(static_cast(new_bs->Data()) + pending_len, - bs->Data(), - nread); - - bs = std::move(new_bs); - nread = bs->ByteLength(); - stream_buf_offset_ = 0; - stream_buf_ab_.Reset(); - - // We have now fully processed the stream_buf_ input chunk (by moving the - // remaining part into buf, which will be accounted for below). - DecrementCurrentSessionMemory(stream_buf_.len); } IncrementCurrentSessionMemory(nread); diff --git a/src/node_http2.h b/src/node_http2.h index b81862b27bce..f3edcf658f80 100644 --- a/src/node_http2.h +++ b/src/node_http2.h @@ -85,9 +85,8 @@ constexpr int kSessionStateClosing = 0x8; constexpr int kSessionStateSending = 0x10; constexpr int kSessionStateWriteInProgress = 0x20; constexpr int kSessionStateReadingStopped = 0x40; -constexpr int kSessionStateReceivePaused = 0x80; -constexpr int kSessionStateReceiving = 0x100; -constexpr int kSessionStateClosePending = 0x200; +constexpr int kSessionStateReceiving = 0x80; +constexpr int kSessionStateClosePending = 0x100; // The Padding Strategy determines the method by which extra padding is // selected for HEADERS and DATA frames. These are configurable via the @@ -698,7 +697,6 @@ class Http2Session : public AsyncWrap, IS_FLAG(sending, kSessionStateSending) IS_FLAG(write_in_progress, kSessionStateWriteInProgress) IS_FLAG(reading_stopped, kSessionStateReadingStopped) - IS_FLAG(receive_paused, kSessionStateReceivePaused) IS_FLAG(receiving, kSessionStateReceiving) IS_FLAG(close_pending, kSessionStateClosePending) @@ -979,7 +977,6 @@ class Http2Session : public AsyncWrap, // will be set. stream_buf_ab_ is lazily created from stream_buf_allocation_. v8::Global stream_buf_ab_; std::unique_ptr stream_buf_allocation_; - size_t stream_buf_offset_ = 0; // Custom error code for errors that originated inside one of the callbacks // called by nghttp2_session_mem_recv. const char* custom_recv_error_code_ = nullptr; diff --git a/test/parallel/test-http2-bidirectional-write-deadlock.js b/test/parallel/test-http2-bidirectional-write-deadlock.js new file mode 100644 index 000000000000..9dd4b91c5234 --- /dev/null +++ b/test/parallel/test-http2-bidirectional-write-deadlock.js @@ -0,0 +1,107 @@ +'use strict'; + +// Regression test against deadlocks between two HTTP/2 peers that are both +// writing at the same time. +// +// To bound how much it buffered while output was backed up, an Http2Session +// used to stop reading from its socket whenever a write was in flight, and +// resume only once that write completed. When the peer was itself blocked +// writing, that write never completed, so the session never read again and +// the connection hung forever with no error and no timeout. +// +// Rather than relying on kernel socket buffers filling up - which depends on +// the platform and configured window sizes - this models one half of that +// cycle directly. The client's socket forwards a write but does not report it +// as complete, substituting for a write blocked because the peer is not +// reading. Only after that write is stalled does the server send its response +// body. A session that stops reading while writing never sees it. + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); +const assert = require('assert'); +const http2 = require('http2'); +const net = require('net'); +const { Duplex } = require('stream'); + +const BODY = 'the response body'; + +const heldCallbacks = []; +let serverStream; + +let stallWrites = false; + +// Client-side socket that forwards writes to a real connection, but can leave +// their completion callbacks pending to model a transport-blocked write. +class StalledClientSocket extends Duplex { + constructor(port) { + super(); + this.inner = net.connect(port, common.localhostIPv4); + this.inner.on('data', (chunk) => this.push(chunk)); + } + _read() { + // Incoming data is pushed as it arrives. + } + _write(chunk, encoding, callback) { + this.inner.write(chunk, encoding); + if (stallWrites) { + heldCallbacks.push(callback); + // Avoid writing from the server re-entrantly inside _write(). The + // ordering is still explicit: this callback is already held. + setImmediate(() => serverStream.end(BODY)); + return; + } + callback(); + } + _final(callback) { + callback(); + } + _destroy(err, callback) { + this.inner.destroy(); + callback(err); + } +} + +const server = http2.createServer(); + +server.on('stream', common.mustCall((stream) => { + // Send headers first. Their response event starts the stalled client write. + stream.respond(); + serverStream = stream; +})); + +server.listen(0, common.mustCall(() => { + const port = server.address().port; + + const client = http2.connect(`http://${common.localhostIPv4}:${port}`, { + createConnection: () => new StalledClientSocket(port), + }); + + const req = client.request({ ':method': 'POST' }); + + let received = ''; + + req.on('response', common.mustCall(() => { + // _write() will schedule the response body only after it has retained the + // callback, guaranteeing that the native write is still in progress. + stallWrites = true; + req.write(Buffer.alloc(256)); + })); + + req.on('data', (chunk) => { + received += chunk; + }); + + req.on('end', common.mustCall(() => { + assert.strictEqual(received, BODY); + assert.ok(heldCallbacks.length > 0, + 'test did not actually stall a socket write'); + + // Let the stalled writes complete so that everything can shut down. + stallWrites = false; + for (const callback of heldCallbacks) callback(); + + client.destroy(); + server.close(); + })); +})); diff --git a/test/parallel/test-http2-stream-backpressure-unread-peer.js b/test/parallel/test-http2-stream-backpressure-unread-peer.js new file mode 100644 index 000000000000..9667c1611b00 --- /dev/null +++ b/test/parallel/test-http2-stream-backpressure-unread-peer.js @@ -0,0 +1,133 @@ +'use strict'; + +// CVE-2019-9517 describes a peer that offers HTTP/2 flow-control credit +// but then doesn't read from its TCP connection (blocking TCP flow). +// A streaming response must stop producing data when the resulting server +// write becomes blocked by the transport. + +// We use a raw client here to directly control the flow of requests and +// stall responses to deterministically reproduce this on all platforms. + +const common = require('../common'); +if (!common.hasCrypto) + common.skip('missing crypto'); +const assert = require('assert'); +const http2 = require('http2'); +const { Duplex, Readable } = require('stream'); + +const { MAX_INITIAL_WINDOW_SIZE } = http2.constants; + +const FRAME_HEADERS = 1; +const FRAME_SETTINGS = 4; +const FRAME_WINDOW_UPDATE = 8; +const FLAG_END_STREAM = 1; +const FLAG_END_HEADERS = 4; +const INITIAL_CONNECTION_WINDOW_SIZE = 65_535; + +const preface = Buffer.from('PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'); + +function frame(type, flags, streamID, payload = Buffer.alloc(0)) { + const header = Buffer.alloc(9); + header.writeUIntBE(payload.length, 0, 3); + header[3] = type; + header[4] = flags; + header.writeUInt32BE(streamID, 5); + return Buffer.concat([header, payload]); +} + +function settings() { + const payload = http2.getPackedSettings({ + initialWindowSize: MAX_INITIAL_WINDOW_SIZE, + }); + return frame(FRAME_SETTINGS, 0, 0, payload); +} + +function connectionWindowUpdate() { + const payload = Buffer.alloc(4); + payload.writeUInt32BE( + MAX_INITIAL_WINDOW_SIZE - INITIAL_CONNECTION_WINDOW_SIZE, 0); + return frame(FRAME_WINDOW_UPDATE, 0, 0, payload); +} + +function request() { + return frame( + FRAME_HEADERS, + FLAG_END_HEADERS | FLAG_END_STREAM, + 1, + Buffer.from([ + 0x82, // :method: GET + 0x86, // :scheme: http + 0x84, // :path: / + 0x41, 0x01, 0x78, // :authority: x + ])); +} + +let onWriteStalled; +const socket = new Duplex({ + read() {}, + write(_data, _encoding, callback) { + if (onWriteStalled === undefined) { + callback(); + return; + } + + // Leave the write pending to model a peer that has stopped reading. + onWriteStalled(); + }, +}); + +// The 1MB body is well within the client's stream and connection windows +// (2^31-1), so HTTP/2 flow control will not limit the response. +const CHUNK_SIZE = 16 * 1024; +const RESPONSE_SIZE = 1024 * 1024; +const chunk = Buffer.alloc(CHUNK_SIZE); + +const serverSession = http2.performServerHandshake(socket); +serverSession.on('stream', common.mustCall((stream) => { + let produced = 0; + + // A datasource that produces chunks on demand + const source = new Readable({ + highWaterMark: CHUNK_SIZE, + read() { + if (produced === RESPONSE_SIZE) { + this.push(null); + return; + } + produced += CHUNK_SIZE; + this.push(chunk); + }, + }); + + // We set this once the stream opens, to stall all future writes once the + // initial preface & SETTINGS dance is completed. + onWriteStalled = common.mustCall(() => { + setImmediate(common.mustCall(() => { + // Backpressure should have been applied to the source when the + // socket writable writes stalled, even though there is still + // H2 flow-control credit available: + assert(source.isPaused()); + assert(stream.writableNeedDrain); + + // Four chunks fill the stream's 64KB writable buffer and one more is + // prefetched into the source's 16KB readable buffer. Only 80KB of the + // 1MB source is produced before TCP backpressure stops buffering. + assert.strictEqual(produced, CHUNK_SIZE * 5); + + serverSession.destroy(); + socket.destroy(); + source.destroy(); + })); + }); + + stream.respond(); + source.pipe(stream); +})); + +// Manually send a single request with all required preamble: +socket.push(Buffer.concat([ + preface, + settings(), + connectionWindowUpdate(), + request(), +])); diff --git a/test/parallel/test-stream-pipeline-http2.js b/test/parallel/test-stream-pipeline-http2.js index 8ffee7786838..0b26c95b3a4e 100644 --- a/test/parallel/test-stream-pipeline-http2.js +++ b/test/parallel/test-stream-pipeline-http2.js @@ -30,8 +30,7 @@ const http2 = require('http2'); let received = 0; req.on('data', (data) => { received += data.length; - // Bound the data that flows before teardown - bytes per data event vary - // by platform, and letting this run longer hangs on macOS. + // Use a byte threshold because data event chunking varies by platform. if (received >= 32 * 1024) rs.destroy(); }); }));