Consolidates stream close behavior in a smaller number of places.
This change also adds a map counting queued frames per stream, so the library does not accidentally close a stream before sending all of the queued frames.
PiperOrigin-RevId: 410097334
diff --git a/http2/adapter/oghttp2_adapter_test.cc b/http2/adapter/oghttp2_adapter_test.cc
index 09f9d0c..78e3f94 100644
--- a/http2/adapter/oghttp2_adapter_test.cc
+++ b/http2/adapter/oghttp2_adapter_test.cc
@@ -1273,7 +1273,6 @@
EXPECT_CALL(http2_visitor_, OnFrameSent(PRIORITY, 3, _, 0x0, 0));
EXPECT_CALL(http2_visitor_, OnBeforeFrameSent(RST_STREAM, 3, _, 0x0));
EXPECT_CALL(http2_visitor_, OnFrameSent(RST_STREAM, 3, _, 0x0, 0x8));
- EXPECT_CALL(http2_visitor_, OnCloseStream(3, Http2ErrorCode::NO_ERROR));
EXPECT_CALL(http2_visitor_, OnBeforeFrameSent(PING, 0, _, 0x0));
EXPECT_CALL(http2_visitor_, OnFrameSent(PING, 0, _, 0x0, 0));
EXPECT_CALL(http2_visitor_, OnBeforeFrameSent(GOAWAY, 0, _, 0x0));
@@ -1525,9 +1524,6 @@
const int64_t result = adapter->ProcessBytes(frames);
EXPECT_EQ(frames.size(), static_cast<size_t>(result));
- // BUG: OnCloseStream() should be invoked after OnFrameSent() for the response
- // headers.
- EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
int submit_result =
adapter->SubmitResponse(1, ToHeaders({{":status", "200"}}), nullptr);
EXPECT_EQ(submit_result, 0);
@@ -1539,6 +1535,7 @@
EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x1, 0));
EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5));
EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0));
+ EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
int send_result = adapter->Send();
EXPECT_EQ(0, send_result);
@@ -1665,7 +1662,6 @@
EXPECT_FALSE(adapter->want_write());
// The body source has been exhausted by the call to Send() above.
- EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
int trailer_result =
adapter->SubmitTrailer(1, ToHeaders({{":final-status", "a-ok"}}));
ASSERT_EQ(trailer_result, 0);
@@ -1673,6 +1669,7 @@
EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5));
EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0));
+ EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
send_result = adapter->Send();
EXPECT_EQ(0, send_result);
@@ -1777,9 +1774,6 @@
EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5));
EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0));
- EXPECT_CALL(visitor, OnBeforeFrameSent(RST_STREAM, 1, _, 0x0));
- EXPECT_CALL(visitor, OnFrameSent(RST_STREAM, 1, _, 0x0, 0));
- EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
send_result = adapter->Send();
EXPECT_EQ(0, send_result);
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc
index d9bfd1d..5ab95a6 100644
--- a/http2/adapter/oghttp2_session.cc
+++ b/http2/adapter/oghttp2_session.cc
@@ -122,7 +122,12 @@
class RunOnExit {
public:
+ RunOnExit() = default;
explicit RunOnExit(std::function<void()> f) : f_(std::move(f)) {}
+
+ RunOnExit(RunOnExit&& other) = default;
+ RunOnExit& operator=(RunOnExit&& other) = default;
+
~RunOnExit() {
if (f_) {
f_();
@@ -393,6 +398,12 @@
streams_reset_.insert(frame->stream_id());
}
}
+ if (frame->stream_id() != 0) {
+ auto result = queued_frames_.insert({frame->stream_id(), 1});
+ if (!result.second) {
+ ++(result.first->second);
+ }
+ }
frames_.push_back(std::move(frame));
}
@@ -470,14 +481,8 @@
// Write blocked.
return SendResult::SEND_BLOCKED;
} else {
- visitor_.OnFrameSent(c.frame_type(), c.stream_id(), frame_payload_length,
- c.flags(), c.error_code());
- if (static_cast<FrameType>(c.frame_type()) == FrameType::RST_STREAM) {
- // If this endpoint is resetting the stream, the stream should be
- // closed. This endpoint is already aware of the outbound RST_STREAM and
- // its error code, so close with NO_ERROR.
- CloseStream(c.stream_id(), Http2ErrorCode::NO_ERROR);
- }
+ AfterFrameSent(c.frame_type(), c.stream_id(), frame_payload_length,
+ c.flags(), c.error_code());
frames_.pop_front();
if (static_cast<size_t>(result) < frame.size()) {
@@ -490,6 +495,24 @@
return SendResult::SEND_OK;
}
+void OgHttp2Session::AfterFrameSent(uint8_t frame_type, uint32_t stream_id,
+ size_t payload_length, uint8_t flags,
+ uint32_t error_code) {
+ visitor_.OnFrameSent(frame_type, stream_id, payload_length, flags,
+ error_code);
+ if (stream_id == 0) {
+ return;
+ }
+ auto iter = queued_frames_.find(stream_id);
+ if (frame_type != 0) {
+ --iter->second;
+ }
+ if (iter->second == 0) {
+ // TODO(birenroy): Consider passing through `error_code` here.
+ CloseStreamIfReady(frame_type, stream_id);
+ }
+}
+
OgHttp2Session::SendResult OgHttp2Session::WriteForStream(
Http2StreamId stream_id) {
auto it = stream_map_.find(stream_id);
@@ -512,7 +535,6 @@
QUICHE_LOG(ERROR) << "Sent fin; can't send trailers.";
} else {
SendTrailers(stream_id, std::move(*block_ptr));
- MaybeCloseWithRstStream(stream_id, state);
}
}
return SendResult::SEND_OK;
@@ -550,31 +572,29 @@
connection_can_write = SendResult::SEND_BLOCKED;
break;
}
- visitor_.OnFrameSent(/* DATA */ 0, stream_id, length, fin ? 0x1 : 0x0, 0);
connection_send_window_ -= length;
state.send_window -= length;
available_window = std::min({connection_send_window_, state.send_window,
static_cast<int32_t>(max_frame_payload_)});
+ if (fin) {
+ state.half_closed_local = true;
+ }
+ AfterFrameSent(/* DATA */ 0, stream_id, length, fin ? 0x1 : 0x0, 0);
+ if (!stream_map_.contains(stream_id)) {
+ // Note: the stream may have been closed if `fin` is true.
+ break;
+ }
}
if (end_data) {
- bool sent_trailers = false;
if (state.trailers != nullptr) {
auto block_ptr = std::move(state.trailers);
if (fin) {
QUICHE_LOG(ERROR) << "Sent fin; can't send trailers.";
} else {
SendTrailers(stream_id, std::move(*block_ptr));
- sent_trailers = true;
}
}
state.outbound_body = nullptr;
- if (fin || sent_trailers) {
- state.half_closed_local = true;
- if (MaybeCloseWithRstStream(stream_id, state)) {
- // No more work on the stream; it has been closed.
- break;
- }
- }
}
}
// If the stream still exists and has data to send, it should be marked as
@@ -648,11 +668,7 @@
return -501; // NGHTTP2_ERR_INVALID_ARGUMENT
}
const bool end_stream = data_source == nullptr;
- if (end_stream) {
- if (iter->second.half_closed_remote) {
- CloseStream(stream_id, Http2ErrorCode::NO_ERROR);
- }
- } else {
+ if (!end_stream) {
// Add data source to stream state
iter->second.outbound_body = std::move(data_source);
write_scheduler_.MarkStreamReady(stream_id, false);
@@ -682,7 +698,6 @@
if (state.outbound_body == nullptr) {
// Enqueue trailers immediately.
SendTrailers(stream_id, ToHeaderBlock(trailers));
- MaybeCloseWithRstStream(stream_id, state);
} else {
QUICHE_LOG_IF(ERROR, state.outbound_body->send_fin())
<< "DataFrameSource will send fin, preventing trailers!";
@@ -771,6 +786,8 @@
options_.perspective == Perspective::kClient) {
// From the client's perspective, the stream can be closed if it's already
// half_closed_local.
+ // TODO(birenroy): consider whether there are outbound frames queued for the
+ // stream.
CloseStream(stream_id, Http2ErrorCode::NO_ERROR);
}
}
@@ -828,6 +845,8 @@
return;
}
visitor_.OnRstStream(stream_id, TranslateErrorCode(error_code));
+ // TODO(birenroy): Consider whether there are outbound frames queued for the
+ // stream.
CloseStream(stream_id, TranslateErrorCode(error_code));
}
@@ -1070,23 +1089,6 @@
EnqueueFrame(std::move(frame));
}
-bool OgHttp2Session::MaybeCloseWithRstStream(Http2StreamId stream_id,
- StreamState& state) {
- if (options_.perspective == Perspective::kServer) {
- if (state.half_closed_remote) {
- CloseStream(stream_id, Http2ErrorCode::NO_ERROR);
- return true;
- } else {
- // Since the peer has not yet ended the stream, this endpoint should
- // send a RST_STREAM NO_ERROR. See RFC 7540 Section 8.1.
- EnqueueFrame(absl::make_unique<spdy::SpdyRstStreamIR>(
- stream_id, spdy::SpdyErrorCode::ERROR_CODE_NO_ERROR));
- // Sending the RST_STREAM also invokes OnCloseStream.
- }
- }
- return false;
-}
-
void OgHttp2Session::MarkDataBuffered(Http2StreamId stream_id, size_t bytes) {
connection_window_manager_.MarkDataBuffered(bytes);
auto it = stream_map_.find(stream_id);
@@ -1181,5 +1183,18 @@
}
}
+void OgHttp2Session::CloseStreamIfReady(uint8_t frame_type,
+ uint32_t stream_id) {
+ auto iter = stream_map_.find(stream_id);
+ if (iter == stream_map_.end()) {
+ return;
+ }
+ const StreamState& state = iter->second;
+ if (static_cast<FrameType>(frame_type) == FrameType::RST_STREAM ||
+ (state.half_closed_local && state.half_closed_remote)) {
+ CloseStream(stream_id, Http2ErrorCode::NO_ERROR);
+ }
+}
+
} // namespace adapter
} // namespace http2
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h
index 88cdb7d..a3a0232 100644
--- a/http2/adapter/oghttp2_session.h
+++ b/http2/adapter/oghttp2_session.h
@@ -261,6 +261,10 @@
// Serializes and sends queued frames.
SendResult SendQueuedFrames();
+ void AfterFrameSent(uint8_t frame_type, uint32_t stream_id,
+ size_t payload_length, uint8_t flags,
+ uint32_t error_code);
+
// Writes DATA frames for stream `stream_id`.
SendResult WriteForStream(Http2StreamId stream_id);
@@ -271,10 +275,6 @@
void SendTrailers(Http2StreamId stream_id, spdy::SpdyHeaderBlock trailers);
- // Encapsulates the RST_STREAM NO_ERROR behavior described in RFC 7540
- // Section 8.1. Returns true if the stream is closed.
- bool MaybeCloseWithRstStream(Http2StreamId stream_id, StreamState& state);
-
// Performs flow control accounting for data sent by the peer.
void MarkDataBuffered(Http2StreamId stream_id, size_t bytes);
@@ -302,6 +302,8 @@
void LatchErrorAndNotify(Http2ErrorCode error_code,
Http2VisitorInterface::ConnectionError error);
+ void CloseStreamIfReady(uint8_t frame_type, uint32_t stream_id);
+
// Receives events when inbound frames are parsed.
Http2VisitorInterface& visitor_;
@@ -347,6 +349,7 @@
WindowManager connection_window_manager_;
absl::flat_hash_set<Http2StreamId> streams_reset_;
+ absl::flat_hash_map<Http2StreamId, int> queued_frames_;
MetadataSequence connection_metadata_;
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc
index b4b464b..1f604ee 100644
--- a/http2/adapter/oghttp2_session_test.cc
+++ b/http2/adapter/oghttp2_session_test.cc
@@ -886,8 +886,6 @@
EXPECT_FALSE(session.want_write());
// The body source has been exhausted by the call to Send() above.
- // TODO(birenroy): Fix this strange ordering.
- EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
int trailer_result = session.SubmitTrailer(
1, ToHeaders({{"final-status", "a-ok"},
{"x-comment", "trailers sure are cool"}}));
@@ -896,6 +894,7 @@
EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5));
EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0));
+ EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
send_result = session.Send();
EXPECT_EQ(0, send_result);
@@ -977,11 +976,9 @@
EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x4, 0));
EXPECT_CALL(visitor, OnFrameSent(DATA, 1, _, 0x0, 0));
- // TODO(birenroy): Fix this strange ordering.
- EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
-
EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5));
EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0));
+ EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR));
send_result = session.Send();
EXPECT_EQ(0, send_result);
