Adds methods to retrieve the receive window size for the connection or a particular stream. Also renames GetPeerConnectionWindow() to GetSendWindowSize(), and implements proper receive-side flow control accounting in OgHttp2Session. PiperOrigin-RevId: 378752864
diff --git a/http2/adapter/http2_adapter.h b/http2/adapter/http2_adapter.h index 2dd7d3f..e53918c 100644 --- a/http2/adapter/http2_adapter.h +++ b/http2/adapter/http2_adapter.h
@@ -69,8 +69,16 @@ // Invokes the visitor's OnReadyToSend() method for serialized frame data. virtual void Send() = 0; - // Returns the connection-level flow control window for the peer. - virtual int GetPeerConnectionWindow() const = 0; + // Returns the connection-level flow control window advertised by the peer. + virtual int GetSendWindowSize() const = 0; + + // Returns the amount of data a peer could send on a given stream. This is + // the outstanding stream receive window. + virtual int GetStreamReceiveWindowSize(Http2StreamId stream_id) const = 0; + + // Returns the total amount of data a peer could send on the connection. This + // is the outstanding connection receive window. + virtual int GetReceiveWindowSize() const = 0; // Gets the highest stream ID value seen in a frame received by this endpoint. // This method is only guaranteed to work for server endpoints.
diff --git a/http2/adapter/nghttp2_adapter.cc b/http2/adapter/nghttp2_adapter.cc index 6ec2a14..d4e3365 100644 --- a/http2/adapter/nghttp2_adapter.cc +++ b/http2/adapter/nghttp2_adapter.cc
@@ -102,10 +102,19 @@ } } -int NgHttp2Adapter::GetPeerConnectionWindow() const { +int NgHttp2Adapter::GetSendWindowSize() const { return session_->GetRemoteWindowSize(); } +int NgHttp2Adapter::GetStreamReceiveWindowSize(Http2StreamId stream_id) const { + return nghttp2_session_get_stream_local_window_size(session_->raw_ptr(), + stream_id); +} + +int NgHttp2Adapter::GetReceiveWindowSize() const { + return nghttp2_session_get_local_window_size(session_->raw_ptr()); +} + Http2StreamId NgHttp2Adapter::GetHighestReceivedStreamId() const { return nghttp2_session_get_last_proc_stream_id(session_->raw_ptr()); }
diff --git a/http2/adapter/nghttp2_adapter.h b/http2/adapter/nghttp2_adapter.h index a3f32df..2fdf3ee 100644 --- a/http2/adapter/nghttp2_adapter.h +++ b/http2/adapter/nghttp2_adapter.h
@@ -49,7 +49,9 @@ void Send() override; - int GetPeerConnectionWindow() const override; + int GetSendWindowSize() const override; + int GetStreamReceiveWindowSize(Http2StreamId stream_id) const override; + int GetReceiveWindowSize() const override; Http2StreamId GetHighestReceivedStreamId() const override;
diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc index 38f2301..df5d186 100644 --- a/http2/adapter/nghttp2_adapter_test.cc +++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -84,8 +84,7 @@ const ssize_t initial_result = adapter->ProcessBytes(initial_frames); EXPECT_EQ(initial_frames.size(), initial_result); - EXPECT_EQ(adapter->GetPeerConnectionWindow(), - kInitialFlowControlWindowSize + 1000); + EXPECT_EQ(adapter->GetSendWindowSize(), kInitialFlowControlWindowSize + 1000); // Some bytes should have been serialized. adapter->Send(); EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::SETTINGS, @@ -143,6 +142,18 @@ spdy::SpdyFrameType::HEADERS})); visitor.Clear(); + // All streams are active and have not yet received any data, so the receive + // window should be at the initial value. + EXPECT_EQ(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(stream_id1)); + EXPECT_EQ(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(stream_id2)); + EXPECT_EQ(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(stream_id3)); + + // Connection has not yet received any data. + EXPECT_EQ(kInitialFlowControlWindowSize, adapter->GetReceiveWindowSize()); + EXPECT_EQ(0, adapter->GetHighestReceivedStreamId()); EXPECT_EQ(kSentinel1, adapter->GetStreamUserData(stream_id1)); @@ -180,6 +191,19 @@ const ssize_t stream_result = adapter->ProcessBytes(stream_frames); EXPECT_EQ(stream_frames.size(), stream_result); + // First stream has received some data. + EXPECT_GT(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(stream_id1)); + // Second stream was closed. + EXPECT_EQ(-1, adapter->GetStreamReceiveWindowSize(stream_id2)); + // Third stream has not received any data. + EXPECT_EQ(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(stream_id3)); + + // Connection window should be the same as the first stream. + EXPECT_EQ(adapter->GetReceiveWindowSize(), + adapter->GetStreamReceiveWindowSize(stream_id1)); + // Should be 3, but this method only works for server adapters. EXPECT_EQ(0, adapter->GetHighestReceivedStreamId()); @@ -250,6 +274,11 @@ EXPECT_GT(stream_id, 0); EXPECT_TRUE(adapter->session().want_write()); adapter->Send(); + + EXPECT_EQ(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(stream_id)); + EXPECT_EQ(kInitialFlowControlWindowSize, adapter->GetReceiveWindowSize()); + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS, spdy::SpdyFrameType::DATA})); EXPECT_THAT(visitor.data(), testing::HasSubstr(kBody)); @@ -489,6 +518,11 @@ EXPECT_EQ(kSentinel1, adapter->GetStreamUserData(1)); + EXPECT_GT(kInitialFlowControlWindowSize, + adapter->GetStreamReceiveWindowSize(1)); + EXPECT_EQ(adapter->GetStreamReceiveWindowSize(1), + adapter->GetReceiveWindowSize()); + // Because stream 3 has already been closed, it's not possible to set user // data. const char* kSentinel3 = "another arbitrary pointer"; @@ -497,8 +531,7 @@ EXPECT_EQ(3, adapter->GetHighestReceivedStreamId()); - EXPECT_EQ(adapter->GetPeerConnectionWindow(), - kInitialFlowControlWindowSize + 1000); + EXPECT_EQ(adapter->GetSendWindowSize(), kInitialFlowControlWindowSize + 1000); EXPECT_TRUE(adapter->session().want_write()); // Some bytes should have been serialized.
diff --git a/http2/adapter/oghttp2_adapter.cc b/http2/adapter/oghttp2_adapter.cc index 9f57aec..11aefb2 100644 --- a/http2/adapter/oghttp2_adapter.cc +++ b/http2/adapter/oghttp2_adapter.cc
@@ -83,10 +83,18 @@ session_->Send(); } -int OgHttp2Adapter::GetPeerConnectionWindow() const { +int OgHttp2Adapter::GetSendWindowSize() const { return session_->GetRemoteWindowSize(); } +int OgHttp2Adapter::GetStreamReceiveWindowSize(Http2StreamId stream_id) const { + return session_->GetStreamReceiveWindowSize(stream_id); +} + +int OgHttp2Adapter::GetReceiveWindowSize() const { + return session_->GetReceiveWindowSize(); +} + Http2StreamId OgHttp2Adapter::GetHighestReceivedStreamId() const { return session_->GetHighestReceivedStreamId(); }
diff --git a/http2/adapter/oghttp2_adapter.h b/http2/adapter/oghttp2_adapter.h index bbdd2a4..477a6e0 100644 --- a/http2/adapter/oghttp2_adapter.h +++ b/http2/adapter/oghttp2_adapter.h
@@ -35,7 +35,9 @@ int window_increment) override; void SubmitMetadata(Http2StreamId stream_id, bool fin) override; void Send() override; - int GetPeerConnectionWindow() const override; + int GetSendWindowSize() const override; + int GetStreamReceiveWindowSize(Http2StreamId stream_id) const override; + int GetReceiveWindowSize() const override; Http2StreamId GetHighestReceivedStreamId() const override; void MarkDataConsumedForStream(Http2StreamId stream_id, size_t num_bytes) override;
diff --git a/http2/adapter/oghttp2_adapter_test.cc b/http2/adapter/oghttp2_adapter_test.cc index bee8640..1675bda 100644 --- a/http2/adapter/oghttp2_adapter_test.cc +++ b/http2/adapter/oghttp2_adapter_test.cc
@@ -43,9 +43,9 @@ EXPECT_QUICHE_BUG(adapter_->SubmitMetadata(3, true), "Not implemented"); } -TEST_F(OgHttp2AdapterTest, GetPeerConnectionWindow) { - const int peer_window = adapter_->GetPeerConnectionWindow(); - EXPECT_GT(peer_window, 0); +TEST_F(OgHttp2AdapterTest, GetSendWindowSize) { + const int peer_window = adapter_->GetSendWindowSize(); + EXPECT_EQ(peer_window, kInitialFlowControlWindowSize); } TEST_F(OgHttp2AdapterTest, MarkDataConsumedForStream) {
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc index 5927914..853c374 100644 --- a/http2/adapter/oghttp2_session.cc +++ b/http2/adapter/oghttp2_session.cc
@@ -23,7 +23,14 @@ } OgHttp2Session::OgHttp2Session(Http2VisitorInterface& visitor, Options options) - : visitor_(visitor), headers_handler_(visitor), options_(options) { + : visitor_(visitor), + headers_handler_(visitor), + connection_window_manager_(kInitialFlowControlWindowSize, + [this](size_t window_update_delta) { + SendWindowUpdate(kConnectionStreamId, + window_update_delta); + }), + options_(options) { decoder_.set_visitor(this); if (options_.perspective == Perspective::kServer) { remaining_preface_ = {spdy::kHttp2ConnectionHeaderPrefix, @@ -59,6 +66,18 @@ return true; } +int OgHttp2Session::GetStreamReceiveWindowSize(Http2StreamId stream_id) const { + auto it = stream_map_.find(stream_id); + if (it != stream_map_.end()) { + return it->second.window_manager.CurrentWindowSize(); + } + return -1; +} + +int OgHttp2Session::GetReceiveWindowSize() const { + return connection_window_manager_.CurrentWindowSize(); +} + ssize_t OgHttp2Session::ProcessBytes(absl::string_view bytes) { ssize_t preface_consumed = 0; if (!remaining_preface_.empty()) { @@ -95,6 +114,7 @@ } else { it->second.window_manager.MarkDataFlushed(num_bytes); } + connection_window_manager_.MarkDataFlushed(num_bytes); return 0; // Remove? } @@ -369,6 +389,7 @@ void OgHttp2Session::OnStreamFrameData(spdy::SpdyStreamId stream_id, const char* data, size_t len) { + MarkDataBuffered(stream_id, len); visitor_.OnDataForStream(stream_id, absl::string_view(data, len)); } @@ -380,13 +401,15 @@ visitor_.OnEndStream(stream_id); } -void OgHttp2Session::OnStreamPadLength(spdy::SpdyStreamId /*stream_id*/, - size_t /*value*/) { - // TODO(181586191): handle padding +void OgHttp2Session::OnStreamPadLength(spdy::SpdyStreamId stream_id, + size_t value) { + MarkDataBuffered(stream_id, 1 + value); + // TODO(181586191): Pass padding to the visitor? } void OgHttp2Session::OnStreamPadding(spdy::SpdyStreamId stream_id, size_t len) { - // TODO(181586191): handle padding + // Flow control was accounted for in OnStreamPadLength(). + // TODO(181586191): Pass padding to the visitor? } spdy::SpdyHeadersHandlerInterface* OgHttp2Session::OnHeaderFrameStart( @@ -559,5 +582,12 @@ } } +void OgHttp2Session::MarkDataBuffered(Http2StreamId stream_id, size_t bytes) { + connection_window_manager_.MarkDataBuffered(bytes); + if (auto it = stream_map_.find(stream_id); it != stream_map_.end()) { + it->second.window_manager.MarkDataBuffered(bytes); + } +} + } // namespace adapter } // namespace http2
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h index e2ab3fd..9634400 100644 --- a/http2/adapter/oghttp2_session.h +++ b/http2/adapter/oghttp2_session.h
@@ -57,6 +57,13 @@ // Resumes a stream that was previously blocked. Returns true on success. bool ResumeStream(Http2StreamId stream_id); + // Returns the outstanding stream receive window, or -1 if the stream does not + // exist. + int GetStreamReceiveWindowSize(Http2StreamId stream_id) const; + + // Returns the outstanding connection receive window. + int GetReceiveWindowSize() const; + // From Http2Session. ssize_t ProcessBytes(absl::string_view bytes) override; int Consume(Http2StreamId stream_id, size_t num_bytes) override; @@ -171,6 +178,9 @@ // Section 8.1. void MaybeCloseWithRstStream(Http2StreamId stream_id, StreamState& state); + // Performs flow control accounting for data sent by the peer. + void MarkDataBuffered(Http2StreamId stream_id, size_t bytes); + // Receives events when inbound frames are parsed. Http2VisitorInterface& visitor_; @@ -199,6 +209,8 @@ // session. absl::string_view remaining_preface_; + WindowManager connection_window_manager_; + Http2StreamId next_stream_id_ = 1; Http2StreamId highest_received_stream_id_ = 0; int connection_send_window_ = kInitialFlowControlWindowSize;
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc index 3869db1..cdaacdc 100644 --- a/http2/adapter/oghttp2_session_test.cc +++ b/http2/adapter/oghttp2_session_test.cc
@@ -67,30 +67,46 @@ kInitialFlowControlWindowSize + 1000); EXPECT_EQ(0, session.GetHighestReceivedStreamId()); + // Connection has not yet received any data. + EXPECT_EQ(kInitialFlowControlWindowSize, session.GetReceiveWindowSize()); + // Should OgHttp2Session require that streams 1 and 3 have been created? + // Submit a request to ensure the first stream is created. + const char* kSentinel1 = "arbitrary pointer 1"; + TestDataFrameSource body1(visitor, "This is an example request body."); + int stream_id = + session.SubmitRequest(ToHeaders({{":method", "POST"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}), + &body1, const_cast<char*>(kSentinel1)); + EXPECT_EQ(stream_id, 1); + const std::string stream_frames = TestFrameSequence() - .Headers(1, + .Headers(stream_id, {{":status", "200"}, {"server", "my-fake-server"}, {"date", "Tue, 6 Apr 2021 12:54:01 GMT"}}, /*fin=*/false) - .Data(1, "This is the response body.") + .Data(stream_id, "This is the response body.") .RstStream(3, Http2ErrorCode::INTERNAL_ERROR) .GoAway(5, Http2ErrorCode::ENHANCE_YOUR_CALM, "calm down!!") .Serialize(); - EXPECT_CALL(visitor, OnFrameHeader(1, _, HEADERS, 4)); - EXPECT_CALL(visitor, OnBeginHeadersForStream(1)); - EXPECT_CALL(visitor, OnHeaderForStream(1, ":status", "200")); - EXPECT_CALL(visitor, OnHeaderForStream(1, "server", "my-fake-server")); + EXPECT_CALL(visitor, OnFrameHeader(stream_id, _, HEADERS, 4)); + EXPECT_CALL(visitor, OnBeginHeadersForStream(stream_id)); + EXPECT_CALL(visitor, OnHeaderForStream(stream_id, ":status", "200")); EXPECT_CALL(visitor, - OnHeaderForStream(1, "date", "Tue, 6 Apr 2021 12:54:01 GMT")); - EXPECT_CALL(visitor, OnEndHeadersForStream(1)); - EXPECT_CALL(visitor, OnFrameHeader(1, 26, DATA, 0)); - EXPECT_CALL(visitor, OnBeginDataForStream(1, 26)); - EXPECT_CALL(visitor, OnDataForStream(1, "This is the response body.")); + OnHeaderForStream(stream_id, "server", "my-fake-server")); + EXPECT_CALL(visitor, OnHeaderForStream(stream_id, "date", + "Tue, 6 Apr 2021 12:54:01 GMT")); + EXPECT_CALL(visitor, OnEndHeadersForStream(stream_id)); + EXPECT_CALL(visitor, OnFrameHeader(stream_id, 26, DATA, 0)); + EXPECT_CALL(visitor, OnBeginDataForStream(stream_id, 26)); + EXPECT_CALL(visitor, + OnDataForStream(stream_id, "This is the response body.")); EXPECT_CALL(visitor, OnFrameHeader(3, 4, RST_STREAM, 0)); EXPECT_CALL(visitor, OnRstStream(3, Http2ErrorCode::INTERNAL_ERROR)); EXPECT_CALL(visitor, OnCloseStream(3, Http2ErrorCode::INTERNAL_ERROR)); @@ -99,6 +115,13 @@ const ssize_t stream_result = session.ProcessBytes(stream_frames); EXPECT_EQ(stream_frames.size(), stream_result); EXPECT_EQ(3, session.GetHighestReceivedStreamId()); + + // The first stream is active and has received some data. + EXPECT_GT(kInitialFlowControlWindowSize, + session.GetStreamReceiveWindowSize(stream_id)); + // Connection receive window is equivalent to the first stream's. + EXPECT_EQ(session.GetReceiveWindowSize(), + session.GetStreamReceiveWindowSize(stream_id)); } // Verifies that a client session enqueues initial SETTINGS if Send() is called @@ -404,6 +427,13 @@ EXPECT_EQ(kSentinel1, session.GetStreamUserData(1)); + // The first stream is active and has received some data. + EXPECT_GT(kInitialFlowControlWindowSize, + session.GetStreamReceiveWindowSize(1)); + // Connection receive window is equivalent to the first stream's. + EXPECT_EQ(session.GetReceiveWindowSize(), + session.GetStreamReceiveWindowSize(1)); + // TODO(birenroy): drop stream state when streams are closed. It should no // longer be possible to set user data. const char* kSentinel3 = "another arbitrary pointer";