diff --git a/core/functional_tests/http2server/service.cpp b/core/functional_tests/http2server/service.cpp index 8fd5dd2da7ec..248d67dfcd44 100644 --- a/core/functional_tests/http2server/service.cpp +++ b/core/functional_tests/http2server/service.cpp @@ -53,13 +53,15 @@ class HandlerHttp2Stream final : public server::handlers::HttpHandlerBase { const auto& count_str = req.GetArg("count"); const std::size_t count = std::stoi(count_str); UASSERT(count != 0); + const auto& delay_str = req.GetArg("delay_ms"); + const std::chrono::milliseconds delay{delay_str.empty() ? 2 : std::stoi(delay_str)}; stream.SetStatusCode(200); stream.SetEndOfHeaders(); for (std::size_t i = 0; i < count - 1; i++) { std::string part{body_part}; stream.PushBodyChunk(std::move(part), {}); // Some pause... - engine::SleepFor(std::chrono::milliseconds{2}); + engine::SleepFor(delay); } std::string part{body_part}; stream.PushBodyChunk(std::move(part), {}); diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index 101273dc0404..885bb53ab5b0 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -1,11 +1,23 @@ import asyncio -import pytest +import h2.connection +import h2.events +import h2.settings + +import utils DEFAULT_PATH = '/http2server-stream' -@pytest.mark.skip(reason='TAXICOMMON-10258') +def _stream_headers(query: str) -> list: + return [ + (':method', 'GET'), + (':path', f'{DEFAULT_PATH}?{query}'), + (':scheme', 'http'), + (':authority', 'localhost'), + ] + + async def test_body_stream(http2_client, service_client, dynamic_config): part = 'part' count = 100 @@ -25,7 +37,6 @@ async def _stream_request(client, req_per_client): assert data == r.text -@pytest.mark.skip(reason='TAXICOMMON-10258') async def test_body_stream_small_pieces( http2_client, service_client, @@ -34,7 +45,6 @@ async def test_body_stream_small_pieces( await _stream_request(http2_client, 1) -@pytest.mark.skip(reason='TAXICOMMON-10258') async def test_body_stream_concurrent( http2_client, service_client, @@ -44,3 +54,179 @@ async def test_body_stream_concurrent( req_per_client = 10 tasks = [_stream_request(http2_client, req_per_client) for _ in range(clients_count)] await asyncio.gather(*tasks) + + +async def test_body_stream_concurrent_unique_bodies( + http2_client, + service_client, + dynamic_config, +): + # Each stream echoes a payload whose every 8-byte block encodes the + # stream number and the block position. Unlike identical payloads, this + # detects bytes leaking between concurrently multiplexed streams as well + # as chunk reordering within one stream. + async def echo_stream(i): + data = ''.join(f'{i:02d}:{j:04d};' for j in range(128)) # 1 KiB + r = await http2_client.get(DEFAULT_PATH, params={'type': 'ne'}, data=data) + assert 200 == r.status_code + assert data == r.text + + await asyncio.gather(*[echo_stream(i) for i in range(20)]) + + +async def test_body_stream_no_head_of_line_blocking( + http2_client, + service_client, + dynamic_config, +): + # A slow streamed response (~5s) on one stream must not delay other + # requests multiplexed on the same connection. If it did, each "fast" + # request below would complete only after the slow stream finishes and + # trip its timeout. + part = 'x' + count = 50 + slow = asyncio.create_task( + http2_client.get( + DEFAULT_PATH, + params={ + 'type': 'eq', + 'body_part': part, + 'count': count, + 'delay_ms': 100, + }, + timeout=30.0, + ), + ) + try: + for _ in range(5): + r = await asyncio.wait_for( + http2_client.get( + '/http2server', + params={'type': 'echo-body'}, + data='ping', + ), + timeout=2.0, + ) + assert 200 == r.status_code + assert 'ping' == r.text + finally: + r = await slow + assert 200 == r.status_code + assert part * count == r.text + + +async def test_reset_mid_stream_keeps_connection_usable( + create_connection, + service_client, +): + async with create_connection() as (sock, conn): + # A slow stream: the handler will keep producing for ~3s after the + # client resets the stream; those events must be dropped, not tear + # down the connection or the process. + stream_id = conn.get_next_available_stream_id() + conn.send_headers( + stream_id, + _stream_headers('type=eq&body_part=part&count=30&delay_ms=100'), + end_stream=True, + ) + await sock.sendall(conn.data_to_send()) + + events = [] + while not any(isinstance(event, h2.events.DataReceived) for event in events): + events += await utils.send_and_receive(sock, conn) + + conn.reset_stream(stream_id, error_code=0x8) # CANCEL + await sock.sendall(conn.data_to_send()) + + # The same connection must still serve requests, concurrently with + # the handler of the reset stream still pushing body parts. + echo_stream_id = conn.get_next_available_stream_id() + conn.send_headers( + echo_stream_id, + [ + (':method', 'GET'), + (':path', '/http2server?type=echo-header'), + (':scheme', 'http'), + (':authority', 'localhost'), + ('echo-header', 'still-alive'), + ], + end_stream=True, + ) + await sock.sendall(conn.data_to_send()) + + events = await utils.receive_until_stream_ended(sock, conn) + assert b'still-alive' == utils.response_data(events) + + +async def test_h2c_upgrade_with_streamed_response(create_socket, service_client): + # The first request of an h2c upgrade is parsed as HTTP/1.1, so the + # streamed response has no HTTP/2 producer; it must degrade to a + # buffered send instead of hanging on a forever-deferred provider. + async with create_socket() as sock: + conn = h2.connection.H2Connection() + settings_header = conn.initiate_upgrade_connection().decode('ascii') + request = ( + f'GET {DEFAULT_PATH}?type=eq&body_part=part&count=10 HTTP/1.1\r\n' + 'Host: localhost\r\n' + 'Connection: Upgrade, HTTP2-Settings\r\n' + 'Upgrade: h2c\r\n' + f'HTTP2-Settings: {settings_header}\r\n' + '\r\n' + ) + await sock.sendall(request.encode('ascii')) + + receive = b'' + while utils.HTTP1_HEADERS_END not in receive: + receive += await sock.recv(utils.RECEIVE_SIZE) + headers, _, http2_data = receive.partition(utils.HTTP1_HEADERS_END) + assert headers.startswith(b'HTTP/1.1 101 Switching Protocols') + + events = conn.receive_data(http2_data) if http2_data else [] + await sock.sendall(conn.data_to_send()) + while not any(isinstance(event, h2.events.StreamEnded) for event in events): + events += await utils.send_and_receive(sock, conn) + + assert b'part' * 10 == utils.response_data(events) + + +async def test_flow_control_backpressure(create_connection, service_client): + # With a tiny stream window the server may only produce as fast as the + # client opens the window with WINDOW_UPDATEs; the deferred provider must + # resume each time instead of stalling or flooding. + window = 1024 + part = 'x' * 1024 + count = 100 # total body is 100 KiB, also exceeds the connection window + + async with create_connection() as (sock, conn): + conn.update_settings( + {h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: window}, + ) + stream_id = conn.get_next_available_stream_id() + conn.send_headers( + stream_id, + _stream_headers(f'type=eq&body_part={part}&count={count}&delay_ms=0'), + end_stream=True, + ) + await sock.sendall(conn.data_to_send()) + + body = b'' + ended = False + while not ended: + receive = await sock.recv(utils.RECEIVE_SIZE) + if not receive: + raise RuntimeError('Socket connection was closed by the other side') + for event in conn.receive_data(receive): + if isinstance(event, h2.events.DataReceived): + body += event.data + conn.acknowledge_received_data( + event.flow_controlled_length, + event.stream_id, + ) + elif isinstance(event, h2.events.StreamEnded): + ended = True + data = conn.data_to_send() + if data: + await sock.sendall(data) + + assert len(body) == len(part) * count + assert part.encode() * count == body diff --git a/core/src/server/http/http2_session.cpp b/core/src/server/http/http2_session.cpp index e1c0d8273b45..3b54ab530235 100644 --- a/core/src/server/http/http2_session.cpp +++ b/core/src/server/http/http2_session.cpp @@ -58,7 +58,6 @@ Http2Session::Http2Session( streaming_consumer_(streaming_queue_->GetConsumer()) { UASSERT(streaming_queue_); - UASSERT(streaming_event_.IsAutoReset()); nghttp2_session_callbacks* callbacks{nullptr}; UINVARIANT(nghttp2_session_callbacks_new(&callbacks) == 0, "Failed to init callbacks for HTTP/2.0"); @@ -367,21 +366,25 @@ void Http2Session::WriteWhileWant() { engine::SingleConsumerEvent& Http2Session::GetStreamingEvent() { return streaming_event_; } -void Http2Session::HandleStreamingEvents() { - impl::Http2StreamEvent event; - while (streaming_consumer_.PopNoblock(event)) { - UASSERT(event.stream_id != -1); - auto& stream = GetStreamChecked(Stream::Id{event.stream_id}); - if (stream.IsDeferred()) { - const auto res = nghttp2_session_resume_data(session_.get(), static_cast(stream.GetId())); - ThrowIfErr(res, "Error while resume_data"); - stream.SetDeferred(false); - } - stream.PushChunk(std::move(event.body_part)); - stream.SetEnd(event.is_end); - event = {}; +bool Http2Session::PopStreamingEventNoblock(impl::Http2StreamEvent& event) { + return streaming_consumer_.PopNoblock(event); +} + +void Http2Session::ApplyStreamingEvent(impl::Http2StreamEvent&& event) { + UASSERT(event.stream_id != -1); + auto* stream = static_cast(nghttp2_session_get_stream_user_data(session_.get(), event.stream_id)); + if (stream == nullptr) { + // The stream is already closed (e.g. reset by the client) while the + // handler was still producing body parts. Drop the event. + return; + } + if (stream->IsDeferred()) { + const auto res = nghttp2_session_resume_data(session_.get(), event.stream_id); + ThrowIfErr(res, "Error while resume_data"); + stream->SetDeferred(false); } - WriteWhileWant(); + stream->PushChunk(std::move(event.body_part)); + stream->SetEnd(event.is_end); } } // namespace server::http diff --git a/core/src/server/http/http2_session.hpp b/core/src/server/http/http2_session.hpp index 3f8ba7deffc6..517f2ecb85c3 100644 --- a/core/src/server/http/http2_session.hpp +++ b/core/src/server/http/http2_session.hpp @@ -62,7 +62,13 @@ class Http2Session final : public request::RequestParser { engine::SingleConsumerEvent& GetStreamingEvent(); void WriteWhileWant(); - void HandleStreamingEvents(); + + // Returns false if there are no pending streaming events. + [[nodiscard]] bool PopStreamingEventNoblock(impl::Http2StreamEvent& event); + + // Applies a body-streaming event to its stream. Events for streams that + // are already closed (e.g. reset by the client) are dropped. + void ApplyStreamingEvent(impl::Http2StreamEvent&& event); bool ConnectionIsOk() const; @@ -131,7 +137,9 @@ class Http2Session final : public request::RequestParser { engine::io::RwBase* socket_; std::shared_ptr streaming_queue_{nullptr}; - engine::SingleConsumerEvent streaming_event_; + // No-auto-reset: the event is awaited through WaitAny in the connection + // loop, which resets it manually before draining the queue. + engine::SingleConsumerEvent streaming_event_{engine::SingleConsumerEvent::NoAutoReset{}}; impl::Http2StreamEventQueue::Consumer streaming_consumer_; }; diff --git a/core/src/server/http/http2_session_test.cpp b/core/src/server/http/http2_session_test.cpp index c4a40812cae3..353c1cb882d7 100644 --- a/core/src/server/http/http2_session_test.cpp +++ b/core/src/server/http/http2_session_test.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include @@ -382,6 +383,83 @@ UTEST_F(Http2SessionTest, ForCurl) { EXPECT_EQ(request->GetMethod(), HttpMethod::kPost); } +UTEST(Http2SessionStreaming, EventQueueIsFifoPerStreamAndSignals) { + const auto queue = impl::Http2StreamEventQueue::Create(); + engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset()}; + impl::Http2StreamEventProducer producer{*queue, event}; + auto consumer = queue->GetConsumer(); + + producer.PushEvent({1, "first"}); + producer.PushEvent({1, "second"}); + producer.CloseStream(1); + EXPECT_TRUE(event.IsReady()); + + impl::Http2StreamEvent popped; + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.stream_id, 1); + EXPECT_EQ(popped.body_part, "first"); + EXPECT_FALSE(popped.is_end); + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.body_part, "second"); + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_TRUE(popped.is_end); + EXPECT_FALSE(consumer.PopNoblock(popped)); +} + +UTEST(Http2SessionStreaming, StreamingEventIsWaitAnyCompatible) { + auto parser = CreateTestParser([](std::shared_ptr&&) {}, USERVER_NAMESPACE::http::HttpVersion::k2); + auto& session = dynamic_cast(*parser); + + auto& event = session.GetStreamingEvent(); + EXPECT_FALSE(event.IsAutoReset()); + // Auto-reset events UINVARIANT-abort here; the connection loop appends + // this token to its WaitAnyContext. + EXPECT_FALSE(event.GetAwaitableToken().IsEmpty()); +} + +UTEST(Http2SessionStreaming, EventForUnknownStreamIsDropped) { + auto parser = CreateTestParser([](std::shared_ptr&&) {}, USERVER_NAMESPACE::http::HttpVersion::k2); + auto& session = dynamic_cast(*parser); + + impl::Http2StreamEvent event; + EXPECT_FALSE(session.PopStreamingEventNoblock(event)); + + // A handler may still be producing body parts after the client reset the + // stream; such events must be dropped, not tear down the connection. + impl::Http2StreamEvent late{42, "late chunk", true}; + EXPECT_NO_THROW(session.ApplyStreamingEvent(std::move(late))); +} + +UTEST(Http2SessionStreaming, SetStreamBodyPicksHttp2Producer) { + const auto queue = impl::Http2StreamEventQueue::Create(); + engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset()}; + + request::ResponseDataAccounter accounter; + const auto request = HttpRequestBuilder{accounter} + .SetMethod(HttpMethod::kGet) + .SetHttpMajor(2) + .SetHttpMinor(0) + .SetUrl("/") + .SetResponseStreamId(1) + .SetStreamProducer(impl::Http2StreamEventProducer{*queue, event}) + .Build(); + auto& response = request->GetHttpResponse(); + + // Used to UINVARIANT-abort the whole process for HTTP/2 responses. + response.SetStreamBody(); + EXPECT_TRUE(response.IsBodyStreamed()); + + auto producer = response.GetBodyProducer(); + ASSERT_TRUE(std::holds_alternative(producer)); + + auto consumer = queue->GetConsumer(); + std::get(producer).PushEvent({1, "chunk"}); + impl::Http2StreamEvent popped; + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.body_part, "chunk"); + EXPECT_TRUE(event.IsReady()); +} + } // namespace server::http USERVER_NAMESPACE_END diff --git a/core/src/server/http/http2_writer.cpp b/core/src/server/http/http2_writer.cpp index 5bf9eedc6e32..3320bd0f75d1 100644 --- a/core/src/server/http/http2_writer.cpp +++ b/core/src/server/http/http2_writer.cpp @@ -105,6 +105,20 @@ class Http2ResponseWriter final { void WriteHttpResponse() { auto data = response_.ExtractData(); + bool buffered_h1_stream = false; + if (response_.IsBodyStreamed() && response_.body_stream_.has_value()) { + // The handler streamed into the HTTP/1.1 queue because the stream + // id was assigned only now, at send time (h2c upgrade). The + // handler has already finished, so buffer the parts and send a + // regular response. + std::string body_part; + while (response_.body_stream_->Pop(body_part)) { + data.append(body_part); + } + response_.body_stream_.reset(); + buffered_h1_stream = true; + } + auto headers = GetHeaders(); const bool is_body_forbidden = IsBodyForbiddenForStatus(response_.status_); @@ -116,7 +130,7 @@ class Http2ResponseWriter final { const auto stream_id = response_.GetStreamId().value(); auto& stream = http2_session_.GetStreamChecked(Stream::Id{stream_id}); - stream.SetStreaming(response_.IsBodyStreamed() && data.empty()); + stream.SetStreaming(response_.IsBodyStreamed() && !buffered_h1_stream && data.empty()); std::size_t bytes = headers.GetSize(); nghttp2_data_provider* provider{nullptr}; diff --git a/core/src/server/http/http_response.cpp b/core/src/server/http/http_response.cpp index 817366707767..18498f007db7 100644 --- a/core/src/server/http/http_response.cpp +++ b/core/src/server/http/http_response.cpp @@ -462,7 +462,6 @@ void SetThrottleReason(http::HttpResponse& http_response, std::string log_reason void HttpResponse::SetStreamBody() { UASSERT(body_stream_producer_.index() == 0); if (GetStreamId().has_value()) { - UINVARIANT(false, "Streaming in HTTP/2.0 is not supported currently."); body_stream_producer_.emplace(GetStreamProducer()); } else { UASSERT(!body_stream_); diff --git a/core/src/server/http/http_response_body_stream.cpp b/core/src/server/http/http_response_body_stream.cpp index de2f5c3d6cb4..e7ded0820ade 100644 --- a/core/src/server/http/http_response_body_stream.cpp +++ b/core/src/server/http/http_response_body_stream.cpp @@ -24,7 +24,14 @@ ResponseBodyStream::~ResponseBodyStream() { void ResponseBodyStream::PushBodyChunk(std::string&& chunk, engine::Deadline deadline) { UASSERT_MSG(headers_ended_, "SetEndOfHeaders() was not called before PushBodyChunk()"); - UASSERT_MSG(http_response_.GetData().empty(), "PushBodyChunk() was called after SetBody()"); + // Only check before the first chunk is announced: after that the + // connection coroutine may be sending the response concurrently and reads + // of the response data would race with it. SetBody() after the first + // chunk is already asserted in SetBody() itself. + UASSERT_MSG( + headers_end_sent_ || http_response_.GetData().empty(), + "PushBodyChunk() was called after SetBody()" + ); if (headers_ended_ && !headers_end_sent_) { http_response_.SetHeadersEnd(); diff --git a/core/src/server/net/http2_connection.cpp b/core/src/server/net/http2_connection.cpp index a34566d05411..01039077af77 100644 --- a/core/src/server/net/http2_connection.cpp +++ b/core/src/server/net/http2_connection.cpp @@ -27,13 +27,17 @@ constexpr std::string_view kHttp2Preface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; constexpr std::string_view kPrefaceBegin = kHttp2Preface.substr(0, kMinLenPrefaceToDetect); constexpr std::uint64_t kSocketId = std::numeric_limits::max(); +constexpr std::uint64_t kStreamingId = std::numeric_limits::max() - 1; -enum class WakeupKind { kSocketReadable, kTaskComputedResponse }; +enum class WakeupKind { kSocketReadable, kStreamingReady, kTaskComputedResponse }; WakeupKind GetWakeupKind(std::uint64_t id) { if (id == kSocketId) { return WakeupKind::kSocketReadable; } + if (id == kStreamingId) { + return WakeupKind::kStreamingReady; + } return WakeupKind::kTaskComputedResponse; } @@ -92,6 +96,7 @@ void Http2Connection::ListenForRequests() { engine::WaitAnyContext wait_any{}; wait_any.Append(kSocketId, GetSocket().GetReadableBase()); + wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); while (!engine::current_task::ShouldCancel()) { StartAllRequestTasks(wait_any); @@ -115,12 +120,23 @@ void Http2Connection::ListenForRequests() { } wait_any.Append(kSocketId, GetSocket().GetReadableBase()); break; + case WakeupKind::kStreamingReady: + // The completed awaitable was dropped out of `wait_any`, so the + // no-auto-reset event has no active awaiter and may be reset + // here. Resetting before the drain keeps a signal arriving + // mid-drain for the next round. `Reset()` is not allowed while + // the event is appended (active awaiter), which is why the + // drain in `OnRequestTaskFinished` leaves the signal alone. + parser_->GetStreamingEvent().Reset(); + HandleStreamingEvents(); + wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); + break; case WakeupKind::kTaskComputedResponse: OnRequestTaskFinished(*ready_id); break; } - UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 1); + UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 2); } } @@ -140,15 +156,69 @@ Http2Connection::RequestTaskContext Http2Connection::StartRequestTask(std::share stats_.active_request_count.Add(1); - return {.task = ConnectionBase::StartRequestTask(request_ptr), .request = std::move(request_ptr)}; + auto task = ConnectionBase::StartRequestTask(request_ptr); + + // `SetStreamBody()` is called synchronously in `StartRequestTask` before + // the handler task is spawned, so `IsBodyStreamed()` is reliable here. + // Requests without a stream id (h2c upgrade) keep the buffered send path. + const auto& response = request_ptr->GetHttpResponse(); + if (response.IsBodyStreamed() && response.GetStreamId().has_value()) { + streamed_requests_.emplace(*response.GetStreamId(), StreamedRequestContext{request_ptr, false}); + } + + return {.task = std::move(task), .request = std::move(request_ptr)}; } void Http2Connection::OnRequestTaskFinished(std::uint64_t event_id) noexcept { - SendResponse(*handler_tasks_[event_id].request); + auto& request = *handler_tasks_[event_id].request; + const auto stream_id = request.GetHttpResponse().GetStreamId(); + if (stream_id.has_value() && streamed_requests_.find(*stream_id) != streamed_requests_.end()) { + // Drain the remaining body parts. `ResponseBodyStream` always pushes a + // final event before the handler task completes, so this also submits + // the response if no streaming event was processed for it yet. + try { + HandleStreamingEvents(); + } catch (const std::exception& ex) { + LOG_ERROR() << "Error while sending streamed body parts: " << ex; + request.GetHttpResponse().SetSendFailed(std::chrono::steady_clock::now()); + } + SubmitStreamedResponseIfPending(*stream_id); + FinalizeResponse(request); + streamed_requests_.erase(*stream_id); + } else { + SendResponse(request); + } handler_tasks_.erase(event_id); } +void Http2Connection::HandleStreamingEvents() { + http::impl::Http2StreamEvent event; + while (parser_->PopStreamingEventNoblock(event)) { + // The first event for a stream means its headers are complete + // (`SetHeadersEnd()` precedes the first `PushBodyChunk()`), so the + // response with its deferred body provider is submitted here. + SubmitStreamedResponseIfPending(event.stream_id); + parser_->ApplyStreamingEvent(std::move(event)); + event = {}; + } + parser_->WriteWhileWant(); +} + +void Http2Connection::SubmitStreamedResponseIfPending(std::int32_t stream_id) noexcept { + const auto it = streamed_requests_.find(stream_id); + if (it == streamed_requests_.end() || it->second.submit_attempted) { + return; + } + it->second.submit_attempted = true; + SubmitResponse(*it->second.request); +} + void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { + SubmitResponse(request); + FinalizeResponse(request); +} + +void Http2Connection::SubmitResponse(http::HttpRequest& request) noexcept { auto& response = request.GetHttpResponse(); UASSERT(!response.IsSent()); request.SetStartSendResponseTime(); @@ -175,6 +245,9 @@ void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { } else { response.SetSendFailed(std::chrono::steady_clock::now()); } +} + +void Http2Connection::FinalizeResponse(http::HttpRequest& request) noexcept { request.SetFinishSendResponseTime(); stats_.active_request_count.Subtract(1); ++stats_.requests_processed_count; diff --git a/core/src/server/net/http2_connection.hpp b/core/src/server/net/http2_connection.hpp index e6c2d5addad0..87158e99f716 100644 --- a/core/src/server/net/http2_connection.hpp +++ b/core/src/server/net/http2_connection.hpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -59,11 +60,23 @@ class Http2Connection final : public ConnectionBase { HttpRequestPtr request; }; + // A request whose handler streams the response body (`IsBodyStreamed()`). + // Its response is submitted upon the first streaming event instead of on + // handler task completion. + struct StreamedRequestContext final { + HttpRequestPtr request; + bool submit_attempted{false}; + }; + void ListenForRequests(); RequestTaskContext StartRequestTask(std::shared_ptr&& request_ptr) noexcept; void StartAllRequestTasks(engine::WaitAnyContext& wait_any); void OnRequestTaskFinished(std::uint64_t event_id) noexcept; + void HandleStreamingEvents(); + void SubmitStreamedResponseIfPending(std::int32_t stream_id) noexcept; void SendResponse(http::HttpRequest& request) noexcept; + void SubmitResponse(http::HttpRequest& request) noexcept; + void FinalizeResponse(http::HttpRequest& request) noexcept; std::unique_ptr MakeParser(); void EnsureHttp2(); @@ -82,6 +95,7 @@ class Http2Connection final : public ConnectionBase { engine::io::Sockaddr remote_address_; std::unique_ptr parser_; utils::SlotMap handler_tasks_; + std::unordered_map streamed_requests_; }; } // namespace server::net