Factor out QuicTransportStream logic so that it can be reused in WebTransport over HTTP/3. PiperOrigin-RevId: 363545740 Change-Id: I97776a85eac70c5058f6efefbae635df4b5457d3
diff --git a/quic/core/quic_stream.h b/quic/core/quic_stream.h index d89e460..822ec18 100644 --- a/quic/core/quic_stream.h +++ b/quic/core/quic_stream.h
@@ -376,6 +376,9 @@ bool fin_buffered() const { return fin_buffered_; } + // True if buffered data in send buffer is below buffered_data_threshold_. + bool CanWriteNewData() const; + protected: // Called when data of [offset, offset + data_length] is buffered in send // buffer. @@ -391,9 +394,6 @@ // a RST_STREAM has been sent. virtual void OnClose(); - // True if buffered data in send buffer is below buffered_data_threshold_. - bool CanWriteNewData() const; - // True if buffered data in send buffer is still below // buffered_data_threshold_ even after writing |length| bytes. bool CanWriteNewDataAfterData(QuicByteCount length) const;
diff --git a/quic/core/web_transport_interface.h b/quic/core/web_transport_interface.h index a458e65..75c66c5 100644 --- a/quic/core/web_transport_interface.h +++ b/quic/core/web_transport_interface.h
@@ -9,6 +9,7 @@ #define QUICHE_QUIC_CORE_WEB_TRANSPORT_INTERFACE_H_ #include <cstddef> +#include <memory> #include "absl/base/attributes.h" #include "absl/strings/string_view.h" @@ -52,6 +53,9 @@ virtual bool CanWrite() const = 0; // Indicates the number of bytes that can be read from the stream. virtual size_t ReadableBytes() const = 0; + + virtual void SetVisitor( + std::unique_ptr<WebTransportStreamVisitor> visitor) = 0; }; // Visitor that gets notified about events related to a WebTransport session.
diff --git a/quic/core/web_transport_stream_adapter.cc b/quic/core/web_transport_stream_adapter.cc new file mode 100644 index 0000000..7a5fee9 --- /dev/null +++ b/quic/core/web_transport_stream_adapter.cc
@@ -0,0 +1,125 @@ +// Copyright 2021 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quic/core/web_transport_stream_adapter.h" + +namespace quic { + +WebTransportStreamAdapter::WebTransportStreamAdapter( + QuicSession* session, + QuicStream* stream, + QuicStreamSequencer* sequencer) + : session_(session), stream_(stream), sequencer_(sequencer) {} + +size_t WebTransportStreamAdapter::Read(char* buffer, size_t buffer_size) { + iovec iov; + iov.iov_base = buffer; + iov.iov_len = buffer_size; + const size_t result = sequencer_->Readv(&iov, 1); + if (sequencer_->IsClosed()) { + MaybeNotifyFinRead(); + } + return result; +} + +size_t WebTransportStreamAdapter::Read(std::string* output) { + const size_t old_size = output->size(); + const size_t bytes_to_read = ReadableBytes(); + output->resize(old_size + bytes_to_read); + size_t bytes_read = Read(&(*output)[old_size], bytes_to_read); + QUICHE_DCHECK_EQ(bytes_to_read, bytes_read); + output->resize(old_size + bytes_read); + return bytes_read; +} + +bool WebTransportStreamAdapter::Write(absl::string_view data) { + if (!CanWrite()) { + return false; + } + + QuicUniqueBufferPtr buffer = MakeUniqueBuffer( + session_->connection()->helper()->GetStreamSendBufferAllocator(), + data.size()); + memcpy(buffer.get(), data.data(), data.size()); + QuicMemSlice memslice(std::move(buffer), data.size()); + QuicConsumedData consumed = + stream_->WriteMemSlices(QuicMemSliceSpan(&memslice), /*fin=*/false); + + if (consumed.bytes_consumed == data.size()) { + return true; + } + if (consumed.bytes_consumed == 0) { + return false; + } + // WebTransportStream::Write() is an all-or-nothing write API. To achieve + // that property, it relies on WriteMemSlices() being an all-or-nothing API. + // If WriteMemSlices() fails to provide that guarantee, we have no way to + // communicate a partial write to the caller, and thus it's safer to just + // close the connection. + QUIC_BUG(WebTransportStreamAdapter partial write) + << "WriteMemSlices() unexpectedly partially consumed the input " + "data, provided: " + << data.size() << ", written: " << consumed.bytes_consumed; + stream_->OnUnrecoverableError( + QUIC_INTERNAL_ERROR, + "WriteMemSlices() unexpectedly partially consumed the input data"); + return false; +} + +bool WebTransportStreamAdapter::SendFin() { + if (!CanWrite()) { + return false; + } + + QuicMemSlice empty; + QuicConsumedData consumed = + stream_->WriteMemSlices(QuicMemSliceSpan(&empty), /*fin=*/true); + QUICHE_DCHECK_EQ(consumed.bytes_consumed, 0u); + return consumed.fin_consumed; +} + +bool WebTransportStreamAdapter::CanWrite() const { + return stream_->CanWriteNewData() && !stream_->write_side_closed(); +} + +size_t WebTransportStreamAdapter::ReadableBytes() const { + return sequencer_->ReadableBytes(); +} + +void WebTransportStreamAdapter::OnDataAvailable() { + if (sequencer_->IsClosed()) { + MaybeNotifyFinRead(); + return; + } + + if (visitor_ == nullptr) { + return; + } + if (ReadableBytes() == 0) { + return; + } + visitor_->OnCanRead(); +} + +void WebTransportStreamAdapter::OnCanWriteNewData() { + // Ensure the origin check has been completed, as the stream can be notified + // about being writable before that. + if (!CanWrite()) { + return; + } + if (visitor_ != nullptr) { + visitor_->OnCanWrite(); + } +} + +void WebTransportStreamAdapter::MaybeNotifyFinRead() { + if (visitor_ == nullptr || fin_read_notified_) { + return; + } + fin_read_notified_ = true; + visitor_->OnFinRead(); + stream_->OnFinRead(); +} + +} // namespace quic
diff --git a/quic/core/web_transport_stream_adapter.h b/quic/core/web_transport_stream_adapter.h new file mode 100644 index 0000000..76a445c --- /dev/null +++ b/quic/core/web_transport_stream_adapter.h
@@ -0,0 +1,53 @@ +// Copyright 2021 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_CORE_WEB_TRANSPORT_STREAM_ADAPTER_H_ +#define QUICHE_QUIC_CORE_WEB_TRANSPORT_STREAM_ADAPTER_H_ + +#include "quic/core/quic_session.h" +#include "quic/core/quic_stream.h" +#include "quic/core/quic_stream_sequencer.h" +#include "quic/core/web_transport_interface.h" + +namespace quic { + +// Converts WebTransportStream API calls into QuicStream API calls. The users +// of this class can either subclass it, or wrap around it. +class QUIC_EXPORT_PRIVATE WebTransportStreamAdapter + : public WebTransportStream { + public: + WebTransportStreamAdapter(QuicSession* session, + QuicStream* stream, + QuicStreamSequencer* sequencer); + + // WebTransportStream implementation. + size_t Read(char* buffer, size_t buffer_size) override; + size_t Read(std::string* output) override; + ABSL_MUST_USE_RESULT bool Write(absl::string_view data) override; + ABSL_MUST_USE_RESULT bool SendFin() override; + bool CanWrite() const override; + size_t ReadableBytes() const override; + void SetVisitor(std::unique_ptr<WebTransportStreamVisitor> visitor) override { + visitor_ = std::move(visitor); + } + + WebTransportStreamVisitor* visitor() { return visitor_.get(); } + + // Calls that need to be passed from the corresponding QuicStream methods. + void OnDataAvailable(); + void OnCanWriteNewData(); + + private: + void MaybeNotifyFinRead(); + + QuicSession* session_; // Unowned. + QuicStream* stream_; // Unowned. + QuicStreamSequencer* sequencer_; // Unowned. + std::unique_ptr<WebTransportStreamVisitor> visitor_; + bool fin_read_notified_ = false; +}; + +} // namespace quic + +#endif // QUICHE_QUIC_CORE_WEB_TRANSPORT_STREAM_ADAPTER_H_
diff --git a/quic/quic_transport/quic_transport_integration_test.cc b/quic/quic_transport/quic_transport_integration_test.cc index 4e237c7..99bf41d 100644 --- a/quic/quic_transport/quic_transport_integration_test.cc +++ b/quic/quic_transport/quic_transport_integration_test.cc
@@ -325,7 +325,7 @@ client_->session()->AcceptIncomingUnidirectionalStream(); ASSERT_TRUE(reply != nullptr); std::string buffer; - reply->set_visitor(VisitorExpectingFin()); + reply->SetVisitor(VisitorExpectingFin()); EXPECT_GT(reply->Read(&buffer), 0u); EXPECT_EQ(buffer, "Stream Two"); @@ -339,7 +339,7 @@ [&stream_received]() { return stream_received; }, kDefaultTimeout)); reply = client_->session()->AcceptIncomingUnidirectionalStream(); ASSERT_TRUE(reply != nullptr); - reply->set_visitor(VisitorExpectingFin()); + reply->SetVisitor(VisitorExpectingFin()); EXPECT_GT(reply->Read(&buffer), 0u); EXPECT_EQ(buffer, "Stream One"); }
diff --git a/quic/quic_transport/quic_transport_stream.cc b/quic/quic_transport/quic_transport_stream.cc index 5e66616..738a535 100644 --- a/quic/quic_transport/quic_transport_stream.cc +++ b/quic/quic_transport/quic_transport_stream.cc
@@ -25,6 +25,7 @@ session->connection()->perspective(), session->IsIncomingStream(id), session->version())), + adapter_(session, this, sequencer()), session_interface_(session_interface) {} size_t QuicTransportStream::Read(char* buffer, size_t buffer_size) { @@ -32,24 +33,15 @@ return 0; } - iovec iov; - iov.iov_base = buffer; - iov.iov_len = buffer_size; - const size_t result = sequencer()->Readv(&iov, 1); - if (sequencer()->IsClosed()) { - MaybeNotifyFinRead(); - } - return result; + return adapter_.Read(buffer, buffer_size); } size_t QuicTransportStream::Read(std::string* output) { - const size_t old_size = output->size(); - const size_t bytes_to_read = ReadableBytes(); - output->resize(old_size + bytes_to_read); - size_t bytes_read = Read(&(*output)[old_size], bytes_to_read); - QUICHE_DCHECK_EQ(bytes_to_read, bytes_read); - output->resize(old_size + bytes_read); - return bytes_read; + if (!session_interface_->IsSessionReady()) { + return 0; + } + + return adapter_.Read(output); } bool QuicTransportStream::Write(absl::string_view data) { @@ -57,33 +49,7 @@ return false; } - QuicUniqueBufferPtr buffer = MakeUniqueBuffer( - session()->connection()->helper()->GetStreamSendBufferAllocator(), - data.size()); - memcpy(buffer.get(), data.data(), data.size()); - QuicMemSlice memslice(std::move(buffer), data.size()); - QuicConsumedData consumed = - WriteMemSlices(QuicMemSliceSpan(&memslice), /*fin=*/false); - - if (consumed.bytes_consumed == data.size()) { - return true; - } - if (consumed.bytes_consumed == 0) { - return false; - } - // QuicTransportStream::Write() is an all-or-nothing write API. To achieve - // that property, it relies on WriteMemSlices() being an all-or-nothing API. - // If WriteMemSlices() fails to provide that guarantee, we have no way to - // communicate a partial write to the caller, and thus it's safer to just - // close the connection. - QUIC_BUG(quic_bug_10893_1) - << "WriteMemSlices() unexpectedly partially consumed the input " - "data, provided: " - << data.size() << ", written: " << consumed.bytes_consumed; - OnUnrecoverableError( - QUIC_INTERNAL_ERROR, - "WriteMemSlices() unexpectedly partially consumed the input data"); - return false; + return adapter_.Write(data); } bool QuicTransportStream::SendFin() { @@ -91,16 +57,11 @@ return false; } - QuicMemSlice empty; - QuicConsumedData consumed = - WriteMemSlices(QuicMemSliceSpan(&empty), /*fin=*/true); - QUICHE_DCHECK_EQ(consumed.bytes_consumed, 0u); - return consumed.fin_consumed; + return adapter_.SendFin(); } bool QuicTransportStream::CanWrite() const { - return session_interface_->IsSessionReady() && CanWriteNewData() && - !write_side_closed(); + return session_interface_->IsSessionReady() && adapter_.CanWrite(); } size_t QuicTransportStream::ReadableBytes() const { @@ -108,22 +69,11 @@ return 0; } - return sequencer()->ReadableBytes(); + return adapter_.ReadableBytes(); } void QuicTransportStream::OnDataAvailable() { - if (sequencer()->IsClosed()) { - MaybeNotifyFinRead(); - return; - } - - if (visitor_ == nullptr) { - return; - } - if (ReadableBytes() == 0) { - return; - } - visitor_->OnCanRead(); + adapter_.OnDataAvailable(); } void QuicTransportStream::OnCanWriteNewData() { @@ -132,18 +82,7 @@ if (!CanWrite()) { return; } - if (visitor_ != nullptr) { - visitor_->OnCanWrite(); - } -} - -void QuicTransportStream::MaybeNotifyFinRead() { - if (visitor_ == nullptr || fin_read_notified_) { - return; - } - fin_read_notified_ = true; - visitor_->OnFinRead(); - OnFinRead(); + adapter_.OnCanWriteNewData(); } } // namespace quic
diff --git a/quic/quic_transport/quic_transport_stream.h b/quic/quic_transport/quic_transport_stream.h index 776b07f..d9c9a73 100644 --- a/quic/quic_transport/quic_transport_stream.h +++ b/quic/quic_transport/quic_transport_stream.h
@@ -14,6 +14,7 @@ #include "quic/core/quic_stream.h" #include "quic/core/quic_types.h" #include "quic/core/web_transport_interface.h" +#include "quic/core/web_transport_stream_adapter.h" #include "quic/quic_transport/quic_transport_session_interface.h" namespace quic { @@ -47,9 +48,9 @@ void OnDataAvailable() override; void OnCanWriteNewData() override; - WebTransportStreamVisitor* visitor() { return visitor_.get(); } - void set_visitor(std::unique_ptr<WebTransportStreamVisitor> visitor) { - visitor_ = std::move(visitor); + WebTransportStreamVisitor* visitor() { return adapter_.visitor(); } + void SetVisitor(std::unique_ptr<WebTransportStreamVisitor> visitor) override { + adapter_.SetVisitor(std::move(visitor)); } protected: @@ -59,9 +60,8 @@ void MaybeNotifyFinRead(); + WebTransportStreamAdapter adapter_; QuicTransportSessionInterface* session_interface_; - std::unique_ptr<WebTransportStreamVisitor> visitor_ = nullptr; - bool fin_read_notified_ = false; }; } // namespace quic
diff --git a/quic/quic_transport/quic_transport_stream_test.cc b/quic/quic_transport/quic_transport_stream_test.cc index 75446eb..364780c 100644 --- a/quic/quic_transport/quic_transport_stream_test.cc +++ b/quic/quic_transport/quic_transport_stream_test.cc
@@ -51,7 +51,7 @@ auto visitor = std::make_unique<MockStreamVisitor>(); visitor_ = visitor.get(); - stream_->set_visitor(std::move(visitor)); + stream_->SetVisitor(std::move(visitor)); } void ReceiveStreamData(absl::string_view data, QuicStreamOffset offset) {
diff --git a/quic/tools/quic_transport_simple_server_session.cc b/quic/tools/quic_transport_simple_server_session.cc index 9acf502..de9fb5c 100644 --- a/quic/tools/quic_transport_simple_server_session.cc +++ b/quic/tools/quic_transport_simple_server_session.cc
@@ -154,21 +154,21 @@ QuicTransportStream* stream) { switch (mode_) { case DISCARD: - stream->set_visitor(std::make_unique<DiscardVisitor>(stream)); + stream->SetVisitor(std::make_unique<DiscardVisitor>(stream)); break; case ECHO: switch (stream->type()) { case BIDIRECTIONAL: QUIC_DVLOG(1) << "Opening bidirectional echo stream " << stream->id(); - stream->set_visitor( + stream->SetVisitor( std::make_unique<BidirectionalEchoVisitor>(stream)); break; case READ_UNIDIRECTIONAL: QUIC_DVLOG(1) << "Started receiving data on unidirectional echo stream " << stream->id(); - stream->set_visitor( + stream->SetVisitor( std::make_unique<UnidirectionalEchoReadVisitor>(this, stream)); break; default: @@ -178,7 +178,7 @@ break; case OUTGOING_BIDIRECTIONAL: - stream->set_visitor(std::make_unique<DiscardVisitor>(stream)); + stream->SetVisitor(std::make_unique<DiscardVisitor>(stream)); ++pending_outgoing_bidirectional_streams_; MaybeCreateOutgoingBidirectionalStream(); break; @@ -252,7 +252,7 @@ ActivateStream(std::move(stream_owned)); QUIC_DVLOG(1) << "Opened echo response stream " << stream->id(); - stream->set_visitor( + stream->SetVisitor( std::make_unique<UnidirectionalEchoWriteVisitor>(stream, data)); stream->visitor()->OnCanWrite(); } @@ -267,7 +267,7 @@ QuicTransportStream* stream = stream_owned.get(); ActivateStream(std::move(stream_owned)); QUIC_DVLOG(1) << "Opened outgoing bidirectional stream " << stream->id(); - stream->set_visitor(std::make_unique<BidirectionalEchoVisitor>(stream)); + stream->SetVisitor(std::make_unique<BidirectionalEchoVisitor>(stream)); if (!stream->Write("hello")) { QUIC_DVLOG(1) << "Write failed."; }