Google Git
Sign in
quiche / quiche / b763f609510b92807659955eaef41227c73bf941 / . / quic / core / quic_stream.cc
blob: 6237dae2f62bf5045b85cb63b925a7e70681ab2a [file] [log] [blame]
// Copyright (c) 2012 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/quic_stream.h"
#include <limits>
#include <string>
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"
#include "quic/core/quic_error_codes.h"
#include "quic/core/quic_flow_controller.h"
#include "quic/core/quic_session.h"
#include "quic/core/quic_types.h"
#include "quic/core/quic_utils.h"
#include "quic/core/quic_versions.h"
#include "quic/platform/api/quic_bug_tracker.h"
#include "quic/platform/api/quic_flag_utils.h"
#include "quic/platform/api/quic_flags.h"
#include "quic/platform/api/quic_logging.h"
#include "quic/platform/api/quic_mem_slice.h"
#include "common/platform/api/quiche_logging.h"
using spdy::SpdyPriority;
namespace quic {
#define ENDPOINT \
(perspective_ == Perspective::IS_SERVER ? "Server: " : "Client: ")
namespace {
QuicByteCount DefaultFlowControlWindow(ParsedQuicVersion version) {
if (!version.AllowsLowFlowControlLimits()) {
return kDefaultFlowControlSendWindow;
}
return 0;
}
QuicByteCount GetInitialStreamFlowControlWindowToSend(QuicSession* session,
QuicStreamId stream_id) {
ParsedQuicVersion version = session->connection()->version();
if (version.handshake_protocol != PROTOCOL_TLS1_3) {
return session->config()->GetInitialStreamFlowControlWindowToSend();
}
// Unidirectional streams (v99 only).
if (VersionHasIetfQuicFrames(version.transport_version) &&
!QuicUtils::IsBidirectionalStreamId(stream_id, version)) {
return session->config()
->GetInitialMaxStreamDataBytesUnidirectionalToSend();
}
if (QuicUtils::IsOutgoingStreamId(version, stream_id,
session->perspective())) {
return session->config()
->GetInitialMaxStreamDataBytesOutgoingBidirectionalToSend();
}
return session->config()
->GetInitialMaxStreamDataBytesIncomingBidirectionalToSend();
}
QuicByteCount GetReceivedFlowControlWindow(QuicSession* session,
QuicStreamId stream_id) {
ParsedQuicVersion version = session->connection()->version();
if (version.handshake_protocol != PROTOCOL_TLS1_3) {
if (session->config()->HasReceivedInitialStreamFlowControlWindowBytes()) {
return session->config()->ReceivedInitialStreamFlowControlWindowBytes();
}
return DefaultFlowControlWindow(version);
}
// Unidirectional streams (v99 only).
if (VersionHasIetfQuicFrames(version.transport_version) &&
!QuicUtils::IsBidirectionalStreamId(stream_id, version)) {
if (session->config()
->HasReceivedInitialMaxStreamDataBytesUnidirectional()) {
return session->config()
->ReceivedInitialMaxStreamDataBytesUnidirectional();
}
return DefaultFlowControlWindow(version);
}
if (QuicUtils::IsOutgoingStreamId(version, stream_id,
session->perspective())) {
if (session->config()
->HasReceivedInitialMaxStreamDataBytesOutgoingBidirectional()) {
return session->config()
->ReceivedInitialMaxStreamDataBytesOutgoingBidirectional();
}
return DefaultFlowControlWindow(version);
}
if (session->config()
->HasReceivedInitialMaxStreamDataBytesIncomingBidirectional()) {
return session->config()
->ReceivedInitialMaxStreamDataBytesIncomingBidirectional();
}
return DefaultFlowControlWindow(version);
}
} // namespace
// static
const SpdyPriority QuicStream::kDefaultPriority;
// static
const int QuicStream::kDefaultUrgency;
PendingStream::PendingStream(QuicStreamId id, QuicSession* session)
: id_(id),
version_(session->version()),
stream_delegate_(session),
stream_bytes_read_(0),
fin_received_(false),
is_bidirectional_(QuicUtils::GetStreamType(id, session->perspective(),
/*peer_initiated = */ true,
session->version()) ==
BIDIRECTIONAL),
connection_flow_controller_(session->flow_controller()),
flow_controller_(session, id,
/*is_connection_flow_controller*/ false,
GetReceivedFlowControlWindow(session, id),
GetInitialStreamFlowControlWindowToSend(session, id),
kStreamReceiveWindowLimit,
session->flow_controller()->auto_tune_receive_window(),
session->flow_controller()),
sequencer_(this) {}
void PendingStream::OnDataAvailable() {
// Data should be kept in the sequencer so that
// QuicSession::ProcessPendingStream() can read it.
}
void PendingStream::OnFinRead() { QUICHE_DCHECK(sequencer_.IsClosed()); }
void PendingStream::AddBytesConsumed(QuicByteCount bytes) {
// It will be called when the metadata of the stream is consumed.
flow_controller_.AddBytesConsumed(bytes);
connection_flow_controller_->AddBytesConsumed(bytes);
}
void PendingStream::ResetWithError(QuicResetStreamError /*error*/) {
// Currently PendingStream is only read-unidirectional. It shouldn't send
// Reset.
QUIC_NOTREACHED();
}
void PendingStream::OnUnrecoverableError(QuicErrorCode error,
const std::string& details) {
stream_delegate_->OnStreamError(error, details);
}
void PendingStream::OnUnrecoverableError(QuicErrorCode error,
QuicIetfTransportErrorCodes ietf_error,
const std::string& details) {
stream_delegate_->OnStreamError(error, ietf_error, details);
}
QuicStreamId PendingStream::id() const { return id_; }
ParsedQuicVersion PendingStream::version() const { return version_; }
void PendingStream::OnStreamFrame(const QuicStreamFrame& frame) {
QUICHE_DCHECK_EQ(frame.stream_id, id_);
bool is_stream_too_long =
(frame.offset > kMaxStreamLength) ||
(kMaxStreamLength - frame.offset < frame.data_length);
if (is_stream_too_long) {
// Close connection if stream becomes too long.
QUIC_PEER_BUG(quic_peer_bug_12570_1)
<< "Receive stream frame reaches max stream length. frame offset "
<< frame.offset << " length " << frame.data_length;
OnUnrecoverableError(QUIC_STREAM_LENGTH_OVERFLOW,
"Peer sends more data than allowed on this stream.");
return;
}
if (frame.offset + frame.data_length > sequencer_.close_offset()) {
OnUnrecoverableError(
QUIC_STREAM_DATA_BEYOND_CLOSE_OFFSET,
absl::StrCat(
"Stream ", id_,
" received data with offset: ", frame.offset + frame.data_length,
", which is beyond close offset: ", sequencer()->close_offset()));
return;
}
if (frame.fin) {
fin_received_ = true;
}
// This count includes duplicate data received.
QuicByteCount frame_payload_size = frame.data_length;
stream_bytes_read_ += frame_payload_size;
// Flow control is interested in tracking highest received offset.
// Only interested in received frames that carry data.
if (frame_payload_size > 0 &&
MaybeIncreaseHighestReceivedOffset(frame.offset + frame_payload_size)) {
// As the highest received offset has changed, check to see if this is a
// violation of flow control.
if (flow_controller_.FlowControlViolation() ||
connection_flow_controller_->FlowControlViolation()) {
OnUnrecoverableError(QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA,
"Flow control violation after increasing offset");
return;
}
}
sequencer_.OnStreamFrame(frame);
}
void PendingStream::OnRstStreamFrame(const QuicRstStreamFrame& frame) {
QUICHE_DCHECK_EQ(frame.stream_id, id_);
if (frame.byte_offset > kMaxStreamLength) {
// Peer are not suppose to write bytes more than maxium allowed.
OnUnrecoverableError(QUIC_STREAM_LENGTH_OVERFLOW,
"Reset frame stream offset overflow.");
return;
}
const QuicStreamOffset kMaxOffset =
std::numeric_limits<QuicStreamOffset>::max();
if (sequencer()->close_offset() != kMaxOffset &&
frame.byte_offset != sequencer()->close_offset()) {
OnUnrecoverableError(
QUIC_STREAM_MULTIPLE_OFFSET,
absl::StrCat("Stream ", id_,
" received new final offset: ", frame.byte_offset,
", which is different from close offset: ",
sequencer()->close_offset()));
return;
}
MaybeIncreaseHighestReceivedOffset(frame.byte_offset);
if (flow_controller_.FlowControlViolation() ||
connection_flow_controller_->FlowControlViolation()) {
OnUnrecoverableError(QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA,
"Flow control violation after increasing offset");
return;
}
}
void PendingStream::OnWindowUpdateFrame(const QuicWindowUpdateFrame& frame) {
QUICHE_DCHECK(is_bidirectional_);
flow_controller_.UpdateSendWindowOffset(frame.max_data);
}
bool PendingStream::MaybeIncreaseHighestReceivedOffset(
QuicStreamOffset new_offset) {
uint64_t increment =
new_offset - flow_controller_.highest_received_byte_offset();
if (!flow_controller_.UpdateHighestReceivedOffset(new_offset)) {
return false;
}
// If |new_offset| increased the stream flow controller's highest received
// offset, increase the connection flow controller's value by the incremental
// difference.
connection_flow_controller_->UpdateHighestReceivedOffset(
connection_flow_controller_->highest_received_byte_offset() + increment);
return true;
}
void PendingStream::OnStopSending(
QuicResetStreamError stop_sending_error_code) {
if (!stop_sending_error_code_) {
stop_sending_error_code_ = stop_sending_error_code;
}
}
void PendingStream::MarkConsumed(QuicByteCount num_bytes) {
sequencer_.MarkConsumed(num_bytes);
}
void PendingStream::StopReading() {
QUIC_DVLOG(1) << "Stop reading from pending stream " << id();
sequencer_.StopReading();
}
QuicStream::QuicStream(PendingStream* pending, QuicSession* session,
bool is_static)
: QuicStream(pending->id_, session, std::move(pending->sequencer_),
is_static,
QuicUtils::GetStreamType(pending->id_, session->perspective(),
/*peer_initiated = */ true,
session->version()),
pending->stream_bytes_read_, pending->fin_received_,
std::move(pending->flow_controller_),
pending->connection_flow_controller_) {
QUICHE_DCHECK(session->version().HasIetfQuicFrames());
sequencer_.set_stream(this);
}
namespace {
absl::optional<QuicFlowController> FlowController(QuicStreamId id,
QuicSession* session,
StreamType type) {
if (type == CRYPTO) {
// The only QuicStream with a StreamType of CRYPTO is QuicCryptoStream, when
// it is using crypto frames instead of stream frames. The QuicCryptoStream
// doesn't have any flow control in that case, so we don't create a
// QuicFlowController for it.
return absl::nullopt;
}
return QuicFlowController(
session, id,
/*is_connection_flow_controller*/ false,
GetReceivedFlowControlWindow(session, id),
GetInitialStreamFlowControlWindowToSend(session, id),
kStreamReceiveWindowLimit,
session->flow_controller()->auto_tune_receive_window(),
session->flow_controller());
}
} // namespace
QuicStream::QuicStream(QuicStreamId id, QuicSession* session, bool is_static,
StreamType type)
: QuicStream(id, session, QuicStreamSequencer(this), is_static, type, 0,
false, FlowController(id, session, type),
session->flow_controller()) {}
QuicStream::QuicStream(QuicStreamId id, QuicSession* session,
QuicStreamSequencer sequencer, bool is_static,
StreamType type, uint64_t stream_bytes_read,
bool fin_received,
absl::optional<QuicFlowController> flow_controller,
QuicFlowController* connection_flow_controller)
: sequencer_(std::move(sequencer)),
id_(id),
session_(session),
stream_delegate_(session),
precedence_(CalculateDefaultPriority(session)),
stream_bytes_read_(stream_bytes_read),
stream_error_(QuicResetStreamError::NoError()),
connection_error_(QUIC_NO_ERROR),
read_side_closed_(false),
write_side_closed_(false),
write_side_data_recvd_state_notified_(false),
fin_buffered_(false),
fin_sent_(false),
fin_outstanding_(false),
fin_lost_(false),
fin_received_(fin_received),
rst_sent_(false),
rst_received_(false),
stop_sending_sent_(false),
flow_controller_(std::move(flow_controller)),
connection_flow_controller_(connection_flow_controller),
stream_contributes_to_connection_flow_control_(true),
busy_counter_(0),
add_random_padding_after_fin_(false),
send_buffer_(
session->connection()->helper()->GetStreamSendBufferAllocator()),
buffered_data_threshold_(GetQuicFlag(FLAGS_quic_buffered_data_threshold)),
is_static_(is_static),
deadline_(QuicTime::Zero()),
was_draining_(false),
type_(VersionHasIetfQuicFrames(session->transport_version()) &&
type != CRYPTO
? QuicUtils::GetStreamType(id_, session->perspective(),
session->IsIncomingStream(id_),
session->version())
: type),
creation_time_(session->connection()->clock()->ApproximateNow()),
perspective_(session->perspective()) {
if (type_ == WRITE_UNIDIRECTIONAL) {
fin_received_ = true;
CloseReadSide();
} else if (type_ == READ_UNIDIRECTIONAL) {
fin_sent_ = true;
CloseWriteSide();
}
if (type_ != CRYPTO) {
stream_delegate_->RegisterStreamPriority(id, is_static_, precedence_);
}
}
QuicStream::~QuicStream() {
if (session_ != nullptr && IsWaitingForAcks()) {
QUIC_DVLOG(1)
<< ENDPOINT << "Stream " << id_
<< " gets destroyed while waiting for acks. stream_bytes_outstanding = "
<< send_buffer_.stream_bytes_outstanding()
<< ", fin_outstanding: " << fin_outstanding_;
}
if (stream_delegate_ != nullptr && type_ != CRYPTO) {
stream_delegate_->UnregisterStreamPriority(id(), is_static_);
}
}
void QuicStream::OnStreamFrame(const QuicStreamFrame& frame) {
QUICHE_DCHECK_EQ(frame.stream_id, id_);
QUICHE_DCHECK(!(read_side_closed_ && write_side_closed_));
if (frame.fin && is_static_) {
OnUnrecoverableError(QUIC_INVALID_STREAM_ID,
"Attempt to close a static stream");
return;
}
if (type_ == WRITE_UNIDIRECTIONAL) {
OnUnrecoverableError(QUIC_DATA_RECEIVED_ON_WRITE_UNIDIRECTIONAL_STREAM,
"Data received on write unidirectional stream");
return;
}
bool is_stream_too_long =
(frame.offset > kMaxStreamLength) ||
(kMaxStreamLength - frame.offset < frame.data_length);
if (is_stream_too_long) {
// Close connection if stream becomes too long.
QUIC_PEER_BUG(quic_peer_bug_10586_1)
<< "Receive stream frame on stream " << id_
<< " reaches max stream length. frame offset " << frame.offset
<< " length " << frame.data_length << ". " << sequencer_.DebugString();
OnUnrecoverableError(
QUIC_STREAM_LENGTH_OVERFLOW,
absl::StrCat("Peer sends more data than allowed on stream ", id_,
". frame: offset = ", frame.offset, ", length = ",
frame.data_length, ". ", sequencer_.DebugString()));
return;
}
if (frame.offset + frame.data_length > sequencer_.close_offset()) {
OnUnrecoverableError(
QUIC_STREAM_DATA_BEYOND_CLOSE_OFFSET,
absl::StrCat(
"Stream ", id_,
" received data with offset: ", frame.offset + frame.data_length,
", which is beyond close offset: ", sequencer_.close_offset()));
return;
}
if (frame.fin && !fin_received_) {
fin_received_ = true;
if (fin_sent_) {
QUICHE_DCHECK(!was_draining_);
session_->StreamDraining(id_,
/*unidirectional=*/type_ != BIDIRECTIONAL);
was_draining_ = true;
}
}
if (read_side_closed_) {
QUIC_DLOG(INFO)
<< ENDPOINT << "Stream " << frame.stream_id
<< " is closed for reading. Ignoring newly received stream data.";
// The subclass does not want to read data: blackhole the data.
return;
}
// This count includes duplicate data received.
QuicByteCount frame_payload_size = frame.data_length;
stream_bytes_read_ += frame_payload_size;
// Flow control is interested in tracking highest received offset.
// Only interested in received frames that carry data.
if (frame_payload_size > 0 &&
MaybeIncreaseHighestReceivedOffset(frame.offset + frame_payload_size)) {
// As the highest received offset has changed, check to see if this is a
// violation of flow control.
QUIC_BUG_IF(quic_bug_12570_2, !flow_controller_.has_value())
<< ENDPOINT << "OnStreamFrame called on stream without flow control";
if ((flow_controller_.has_value() &&
flow_controller_->FlowControlViolation()) ||
connection_flow_controller_->FlowControlViolation()) {
OnUnrecoverableError(QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA,
"Flow control violation after increasing offset");
return;
}
}
sequencer_.OnStreamFrame(frame);
}
bool QuicStream::OnStopSending(QuicResetStreamError error) {
// Do not reset the stream if all data has been sent and acknowledged.
if (write_side_closed() && !IsWaitingForAcks()) {
QUIC_DVLOG(1) << ENDPOINT
<< "Ignoring STOP_SENDING for a write closed stream, id: "
<< id_;
return false;
}
if (is_static_) {
QUIC_DVLOG(1) << ENDPOINT
<< "Received STOP_SENDING for a static stream, id: " << id_
<< " Closing connection";
OnUnrecoverableError(QUIC_INVALID_STREAM_ID,
"Received STOP_SENDING for a static stream");
return false;
}
stream_error_ = error;
MaybeSendRstStream(error);
return true;
}
int QuicStream::num_frames_received() const {
return sequencer_.num_frames_received();
}
int QuicStream::num_duplicate_frames_received() const {
return sequencer_.num_duplicate_frames_received();
}
void QuicStream::OnStreamReset(const QuicRstStreamFrame& frame) {
rst_received_ = true;
if (frame.byte_offset > kMaxStreamLength) {
// Peer are not suppose to write bytes more than maxium allowed.
OnUnrecoverableError(QUIC_STREAM_LENGTH_OVERFLOW,
"Reset frame stream offset overflow.");
return;
}
const QuicStreamOffset kMaxOffset =
std::numeric_limits<QuicStreamOffset>::max();
if (sequencer()->close_offset() != kMaxOffset &&
frame.byte_offset != sequencer()->close_offset()) {
OnUnrecoverableError(
QUIC_STREAM_MULTIPLE_OFFSET,
absl::StrCat("Stream ", id_,
" received new final offset: ", frame.byte_offset,
", which is different from close offset: ",
sequencer_.close_offset()));
return;
}
MaybeIncreaseHighestReceivedOffset(frame.byte_offset);
QUIC_BUG_IF(quic_bug_12570_3, !flow_controller_.has_value())
<< ENDPOINT << "OnStreamReset called on stream without flow control";
if ((flow_controller_.has_value() &&
flow_controller_->FlowControlViolation()) ||
connection_flow_controller_->FlowControlViolation()) {
OnUnrecoverableError(QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA,
"Flow control violation after increasing offset");
return;
}
stream_error_ = frame.error();
// Google QUIC closes both sides of the stream in response to a
// RESET_STREAM, IETF QUIC closes only the read side.
if (!VersionHasIetfQuicFrames(transport_version())) {
CloseWriteSide();
}
CloseReadSide();
}
void QuicStream::OnConnectionClosed(QuicErrorCode error,
ConnectionCloseSource /*source*/) {
if (read_side_closed_ && write_side_closed_) {
return;
}
if (error != QUIC_NO_ERROR) {
stream_error_ =
QuicResetStreamError::FromInternal(QUIC_STREAM_CONNECTION_ERROR);
connection_error_ = error;
}
CloseWriteSide();
CloseReadSide();
}
void QuicStream::OnFinRead() {
QUICHE_DCHECK(sequencer_.IsClosed());
// OnFinRead can be called due to a FIN flag in a headers block, so there may
// have been no OnStreamFrame call with a FIN in the frame.
fin_received_ = true;
// If fin_sent_ is true, then CloseWriteSide has already been called, and the
// stream will be destroyed by CloseReadSide, so don't need to call
// StreamDraining.
CloseReadSide();
}
void QuicStream::SetFinSent() {
QUICHE_DCHECK(!VersionUsesHttp3(transport_version()));
fin_sent_ = true;
}
void QuicStream::Reset(QuicRstStreamErrorCode error) {
ResetWithError(QuicResetStreamError::FromInternal(error));
}
void QuicStream::ResetWithError(QuicResetStreamError error) {
stream_error_ = error;
QuicConnection::ScopedPacketFlusher flusher(session()->connection());
MaybeSendStopSending(error);
MaybeSendRstStream(error);
if (read_side_closed_ && write_side_closed_ && !IsWaitingForAcks()) {
session()->MaybeCloseZombieStream(id_);
}
}
void QuicStream::ResetWriteSide(QuicResetStreamError error) {
stream_error_ = error;
MaybeSendRstStream(error);
if (read_side_closed_ && write_side_closed_ && !IsWaitingForAcks()) {
session()->MaybeCloseZombieStream(id_);
}
}
void QuicStream::SendStopSending(QuicResetStreamError error) {
stream_error_ = error;
MaybeSendStopSending(error);
if (read_side_closed_ && write_side_closed_ && !IsWaitingForAcks()) {
session()->MaybeCloseZombieStream(id_);
}
}
void QuicStream::OnUnrecoverableError(QuicErrorCode error,
const std::string& details) {
stream_delegate_->OnStreamError(error, details);
}
void QuicStream::OnUnrecoverableError(QuicErrorCode error,
QuicIetfTransportErrorCodes ietf_error,
const std::string& details) {
stream_delegate_->OnStreamError(error, ietf_error, details);
}
const spdy::SpdyStreamPrecedence& QuicStream::precedence() const {
return precedence_;
}
void QuicStream::SetPriority(const spdy::SpdyStreamPrecedence& precedence) {
precedence_ = precedence;
MaybeSendPriorityUpdateFrame();
stream_delegate_->UpdateStreamPriority(id(), precedence);
}
void QuicStream::WriteOrBufferData(
absl::string_view data, bool fin,
QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener) {
QUIC_BUG_IF(quic_bug_12570_4,
QuicUtils::IsCryptoStreamId(transport_version(), id_))
<< ENDPOINT
<< "WriteOrBufferData is used to send application data, use "
"WriteOrBufferDataAtLevel to send crypto data.";
return WriteOrBufferDataAtLevel(
data, fin, session()->GetEncryptionLevelToSendApplicationData(),
ack_listener);
}
void QuicStream::WriteOrBufferDataAtLevel(
absl::string_view data, bool fin, EncryptionLevel level,
QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener) {
if (data.empty() && !fin) {
QUIC_BUG(quic_bug_10586_2) << "data.empty() && !fin";
return;
}
if (fin_buffered_) {
QUIC_BUG(quic_bug_10586_3) << "Fin already buffered";
return;
}
if (write_side_closed_) {
QUIC_DLOG(ERROR) << ENDPOINT
<< "Attempt to write when the write side is closed";
if (type_ == READ_UNIDIRECTIONAL) {
OnUnrecoverableError(QUIC_TRY_TO_WRITE_DATA_ON_READ_UNIDIRECTIONAL_STREAM,
"Try to send data on read unidirectional stream");
}
return;
}
fin_buffered_ = fin;
bool had_buffered_data = HasBufferedData();
// Do not respect buffered data upper limit as WriteOrBufferData guarantees
// all data to be consumed.
if (data.length() > 0) {
struct iovec iov(QuicUtils::MakeIovec(data));
QuicStreamOffset offset = send_buffer_.stream_offset();
if (kMaxStreamLength - offset < data.length()) {
QUIC_BUG(quic_bug_10586_4) << "Write too many data via stream " << id_;
OnUnrecoverableError(
QUIC_STREAM_LENGTH_OVERFLOW,
absl::StrCat("Write too many data via stream ", id_));
return;
}
send_buffer_.SaveStreamData(&iov, 1, 0, data.length());
OnDataBuffered(offset, data.length(), ack_listener);
}
if (!had_buffered_data && (HasBufferedData() || fin_buffered_)) {
// Write data if there is no buffered data before.
WriteBufferedData(level);
}
}
void QuicStream::OnCanWrite() {
if (HasDeadlinePassed()) {
OnDeadlinePassed();
return;
}
if (HasPendingRetransmission()) {
WritePendingRetransmission();
// Exit early to allow other streams to write pending retransmissions if
// any.
return;
}
if (write_side_closed_) {
QUIC_DLOG(ERROR)
<< ENDPOINT << "Stream " << id()
<< " attempting to write new data when the write side is closed";
return;
}
if (HasBufferedData() || (fin_buffered_ && !fin_sent_)) {
WriteBufferedData(session()->GetEncryptionLevelToSendApplicationData());
}
if (!fin_buffered_ && !fin_sent_ && CanWriteNewData()) {
// Notify upper layer to write new data when buffered data size is below
// low water mark.
OnCanWriteNewData();
}
}
void QuicStream::MaybeSendBlocked() {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_5)
<< ENDPOINT << "MaybeSendBlocked called on stream without flow control";
return;
}
if (flow_controller_->ShouldSendBlocked()) {
session_->SendBlocked(id_);
}
if (!stream_contributes_to_connection_flow_control_) {
return;
}
if (connection_flow_controller_->ShouldSendBlocked()) {
session_->SendBlocked(QuicUtils::GetInvalidStreamId(transport_version()));
}
// If the stream is blocked by connection-level flow control but not by
// stream-level flow control, add the stream to the write blocked list so that
// the stream will be given a chance to write when a connection-level
// WINDOW_UPDATE arrives.
if (connection_flow_controller_->IsBlocked() &&
!flow_controller_->IsBlocked()) {
session_->MarkConnectionLevelWriteBlocked(id());
}
}
QuicConsumedData QuicStream::WriteMemSlice(QuicMemSlice span, bool fin) {
return WriteMemSlices(absl::MakeSpan(&span, 1), fin);
}
QuicConsumedData QuicStream::WriteMemSlices(absl::Span<QuicMemSlice> span,
bool fin) {
QuicConsumedData consumed_data(0, false);
if (span.empty() && !fin) {
QUIC_BUG(quic_bug_10586_6) << "span.empty() && !fin";
return consumed_data;
}
if (fin_buffered_) {
QUIC_BUG(quic_bug_10586_7) << "Fin already buffered";
return consumed_data;
}
if (write_side_closed_) {
QUIC_DLOG(ERROR) << ENDPOINT << "Stream " << id()
<< " attempting to write when the write side is closed";
if (type_ == READ_UNIDIRECTIONAL) {
OnUnrecoverableError(QUIC_TRY_TO_WRITE_DATA_ON_READ_UNIDIRECTIONAL_STREAM,
"Try to send data on read unidirectional stream");
}
return consumed_data;
}
bool had_buffered_data = HasBufferedData();
if (CanWriteNewData() || span.empty()) {
consumed_data.fin_consumed = fin;
if (!span.empty()) {
// Buffer all data if buffered data size is below limit.
QuicStreamOffset offset = send_buffer_.stream_offset();
consumed_data.bytes_consumed = send_buffer_.SaveMemSliceSpan(span);
if (offset > send_buffer_.stream_offset() ||
kMaxStreamLength < send_buffer_.stream_offset()) {
QUIC_BUG(quic_bug_10586_8) << "Write too many data via stream " << id_;
OnUnrecoverableError(
QUIC_STREAM_LENGTH_OVERFLOW,
absl::StrCat("Write too many data via stream ", id_));
return consumed_data;
}
OnDataBuffered(offset, consumed_data.bytes_consumed, nullptr);
}
}
fin_buffered_ = consumed_data.fin_consumed;
if (!had_buffered_data && (HasBufferedData() || fin_buffered_)) {
// Write data if there is no buffered data before.
WriteBufferedData(session()->GetEncryptionLevelToSendApplicationData());
}
return consumed_data;
}
bool QuicStream::HasPendingRetransmission() const {
return send_buffer_.HasPendingRetransmission() || fin_lost_;
}
bool QuicStream::IsStreamFrameOutstanding(QuicStreamOffset offset,
QuicByteCount data_length,
bool fin) const {
return send_buffer_.IsStreamDataOutstanding(offset, data_length) ||
(fin && fin_outstanding_);
}
void QuicStream::CloseReadSide() {
if (read_side_closed_) {
return;
}
QUIC_DVLOG(1) << ENDPOINT << "Done reading from stream " << id();
read_side_closed_ = true;
sequencer_.ReleaseBuffer();
if (write_side_closed_) {
QUIC_DVLOG(1) << ENDPOINT << "Closing stream " << id();
session_->OnStreamClosed(id());
OnClose();
}
}
void QuicStream::CloseWriteSide() {
if (write_side_closed_) {
return;
}
QUIC_DVLOG(1) << ENDPOINT << "Done writing to stream " << id();
write_side_closed_ = true;
if (read_side_closed_) {
QUIC_DVLOG(1) << ENDPOINT << "Closing stream " << id();
session_->OnStreamClosed(id());
OnClose();
}
}
void QuicStream::MaybeSendStopSending(QuicResetStreamError error) {
if (stop_sending_sent_) {
return;
}
if (!session()->version().UsesHttp3() && !error.ok()) {
// In gQUIC, RST with error closes both read and write side.
return;
}
if (session()->version().UsesHttp3()) {
session()->MaybeSendStopSendingFrame(id(), error);
} else {
QUICHE_DCHECK_EQ(QUIC_STREAM_NO_ERROR, error.internal_code());
session()->MaybeSendRstStreamFrame(id(), QuicResetStreamError::NoError(),
stream_bytes_written());
}
stop_sending_sent_ = true;
CloseReadSide();
}
void QuicStream::MaybeSendRstStream(QuicResetStreamError error) {
if (rst_sent_) {
return;
}
if (!session()->version().UsesHttp3()) {
QUIC_BUG_IF(quic_bug_12570_5, error.ok());
stop_sending_sent_ = true;
CloseReadSide();
}
session()->MaybeSendRstStreamFrame(id(), error, stream_bytes_written());
rst_sent_ = true;
CloseWriteSide();
}
bool QuicStream::HasBufferedData() const {
QUICHE_DCHECK_GE(send_buffer_.stream_offset(), stream_bytes_written());
return send_buffer_.stream_offset() > stream_bytes_written();
}
ParsedQuicVersion QuicStream::version() const { return session_->version(); }
QuicTransportVersion QuicStream::transport_version() const {
return session_->transport_version();
}
HandshakeProtocol QuicStream::handshake_protocol() const {
return session_->connection()->version().handshake_protocol;
}
void QuicStream::StopReading() {
QUIC_DVLOG(1) << ENDPOINT << "Stop reading from stream " << id();
sequencer_.StopReading();
}
void QuicStream::OnClose() {
QUICHE_DCHECK(read_side_closed_ && write_side_closed_);
if (!fin_sent_ && !rst_sent_) {
QUIC_BUG_IF(quic_bug_12570_6, session()->connection()->connected() &&
session()->version().UsesHttp3())
<< "The stream should've already sent RST in response to "
"STOP_SENDING";
// For flow control accounting, tell the peer how many bytes have been
// written on this stream before termination. Done here if needed, using a
// RST_STREAM frame.
MaybeSendRstStream(QUIC_RST_ACKNOWLEDGEMENT);
session_->MaybeCloseZombieStream(id_);
}
if (!flow_controller_.has_value() ||
flow_controller_->FlowControlViolation() ||
connection_flow_controller_->FlowControlViolation()) {
return;
}
// The stream is being closed and will not process any further incoming bytes.
// As there may be more bytes in flight, to ensure that both endpoints have
// the same connection level flow control state, mark all unreceived or
// buffered bytes as consumed.
QuicByteCount bytes_to_consume =
flow_controller_->highest_received_byte_offset() -
flow_controller_->bytes_consumed();
AddBytesConsumed(bytes_to_consume);
}
void QuicStream::OnWindowUpdateFrame(const QuicWindowUpdateFrame& frame) {
if (type_ == READ_UNIDIRECTIONAL) {
OnUnrecoverableError(
QUIC_WINDOW_UPDATE_RECEIVED_ON_READ_UNIDIRECTIONAL_STREAM,
"WindowUpdateFrame received on READ_UNIDIRECTIONAL stream.");
return;
}
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_9)
<< ENDPOINT
<< "OnWindowUpdateFrame called on stream without flow control";
return;
}
if (flow_controller_->UpdateSendWindowOffset(frame.max_data)) {
// Let session unblock this stream.
session_->MarkConnectionLevelWriteBlocked(id_);
}
}
bool QuicStream::MaybeIncreaseHighestReceivedOffset(
QuicStreamOffset new_offset) {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_10)
<< ENDPOINT
<< "MaybeIncreaseHighestReceivedOffset called on stream without "
"flow control";
return false;
}
uint64_t increment =
new_offset - flow_controller_->highest_received_byte_offset();
if (!flow_controller_->UpdateHighestReceivedOffset(new_offset)) {
return false;
}
// If |new_offset| increased the stream flow controller's highest received
// offset, increase the connection flow controller's value by the incremental
// difference.
if (stream_contributes_to_connection_flow_control_) {
connection_flow_controller_->UpdateHighestReceivedOffset(
connection_flow_controller_->highest_received_byte_offset() +
increment);
}
return true;
}
void QuicStream::AddBytesSent(QuicByteCount bytes) {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_11)
<< ENDPOINT << "AddBytesSent called on stream without flow control";
return;
}
flow_controller_->AddBytesSent(bytes);
if (stream_contributes_to_connection_flow_control_) {
connection_flow_controller_->AddBytesSent(bytes);
}
}
void QuicStream::AddBytesConsumed(QuicByteCount bytes) {
if (type_ == CRYPTO) {
// A stream with type CRYPTO has no flow control, so there's nothing this
// function needs to do. This function still gets called by the
// QuicStreamSequencers used by QuicCryptoStream.
return;
}
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_12570_7)
<< ENDPOINT
<< "AddBytesConsumed called on non-crypto stream without flow control";
return;
}
// Only adjust stream level flow controller if still reading.
if (!read_side_closed_) {
flow_controller_->AddBytesConsumed(bytes);
}
if (stream_contributes_to_connection_flow_control_) {
connection_flow_controller_->AddBytesConsumed(bytes);
}
}
bool QuicStream::MaybeConfigSendWindowOffset(QuicStreamOffset new_offset,
bool was_zero_rtt_rejected) {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_12)
<< ENDPOINT
<< "ConfigSendWindowOffset called on stream without flow control";
return false;
}
// The validation code below is for QUIC with TLS only.
if (new_offset < flow_controller_->send_window_offset()) {
QUICHE_DCHECK(session()->version().UsesTls());
if (was_zero_rtt_rejected && new_offset < flow_controller_->bytes_sent()) {
// The client is given flow control window lower than what's written in
// 0-RTT. This QUIC implementation is unable to retransmit them.
QUIC_BUG_IF(quic_bug_12570_8, perspective_ == Perspective::IS_SERVER)
<< "Server streams' flow control should never be configured twice.";
OnUnrecoverableError(
QUIC_ZERO_RTT_UNRETRANSMITTABLE,
absl::StrCat(
"Server rejected 0-RTT, aborting because new stream max data ",
new_offset, " for stream ", id_, " is less than currently used: ",
flow_controller_->bytes_sent()));
return false;
} else if (session()->version().AllowsLowFlowControlLimits()) {
// In IETF QUIC, if the client receives flow control limit lower than what
// was resumed from 0-RTT, depending on 0-RTT status, it's either the
// peer's fault or our implementation's fault.
QUIC_BUG_IF(quic_bug_12570_9, perspective_ == Perspective::IS_SERVER)
<< "Server streams' flow control should never be configured twice.";
OnUnrecoverableError(
was_zero_rtt_rejected ? QUIC_ZERO_RTT_REJECTION_LIMIT_REDUCED
: QUIC_ZERO_RTT_RESUMPTION_LIMIT_REDUCED,
absl::StrCat(
was_zero_rtt_rejected ? "Server rejected 0-RTT, aborting because "
: "",
"new stream max data ", new_offset, " decreases current limit: ",
flow_controller_->send_window_offset()));
return false;
}
}
if (flow_controller_->UpdateSendWindowOffset(new_offset)) {
// Let session unblock this stream.
session_->MarkConnectionLevelWriteBlocked(id_);
}
return true;
}
void QuicStream::AddRandomPaddingAfterFin() {
add_random_padding_after_fin_ = true;
}
bool QuicStream::OnStreamFrameAcked(QuicStreamOffset offset,
QuicByteCount data_length, bool fin_acked,
QuicTime::Delta /*ack_delay_time*/,
QuicTime /*receive_timestamp*/,
QuicByteCount* newly_acked_length) {
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_ << " Acking "
<< "[" << offset << ", " << offset + data_length << "]"
<< " fin = " << fin_acked;
*newly_acked_length = 0;
if (!send_buffer_.OnStreamDataAcked(offset, data_length,
newly_acked_length)) {
OnUnrecoverableError(QUIC_INTERNAL_ERROR, "Trying to ack unsent data.");
return false;
}
if (!fin_sent_ && fin_acked) {
OnUnrecoverableError(QUIC_INTERNAL_ERROR, "Trying to ack unsent fin.");
return false;
}
// Indicates whether ack listener's OnPacketAcked should be called.
const bool new_data_acked =
*newly_acked_length > 0 || (fin_acked && fin_outstanding_);
if (fin_acked) {
fin_outstanding_ = false;
fin_lost_ = false;
}
if (!IsWaitingForAcks() && write_side_closed_ &&
!write_side_data_recvd_state_notified_) {
OnWriteSideInDataRecvdState();
write_side_data_recvd_state_notified_ = true;
}
if (!IsWaitingForAcks() && read_side_closed_ && write_side_closed_) {
session_->MaybeCloseZombieStream(id_);
}
return new_data_acked;
}
void QuicStream::OnStreamFrameRetransmitted(QuicStreamOffset offset,
QuicByteCount data_length,
bool fin_retransmitted) {
send_buffer_.OnStreamDataRetransmitted(offset, data_length);
if (fin_retransmitted) {
fin_lost_ = false;
}
}
void QuicStream::OnStreamFrameLost(QuicStreamOffset offset,
QuicByteCount data_length, bool fin_lost) {
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_ << " Losting "
<< "[" << offset << ", " << offset + data_length << "]"
<< " fin = " << fin_lost;
if (data_length > 0) {
send_buffer_.OnStreamDataLost(offset, data_length);
}
if (fin_lost && fin_outstanding_) {
fin_lost_ = true;
}
}
bool QuicStream::RetransmitStreamData(QuicStreamOffset offset,
QuicByteCount data_length, bool fin,
TransmissionType type) {
QUICHE_DCHECK(type == PTO_RETRANSMISSION || type == RTO_RETRANSMISSION ||
type == TLP_RETRANSMISSION || type == PROBING_RETRANSMISSION);
if (HasDeadlinePassed()) {
OnDeadlinePassed();
return true;
}
QuicIntervalSet<QuicStreamOffset> retransmission(offset,
offset + data_length);
retransmission.Difference(bytes_acked());
bool retransmit_fin = fin && fin_outstanding_;
if (retransmission.Empty() && !retransmit_fin) {
return true;
}
QuicConsumedData consumed(0, false);
for (const auto& interval : retransmission) {
QuicStreamOffset retransmission_offset = interval.min();
QuicByteCount retransmission_length = interval.max() - interval.min();
const bool can_bundle_fin =
retransmit_fin && (retransmission_offset + retransmission_length ==
stream_bytes_written());
consumed = stream_delegate_->WritevData(
id_, retransmission_length, retransmission_offset,
can_bundle_fin ? FIN : NO_FIN, type,
session()->GetEncryptionLevelToSendApplicationData());
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_
<< " is forced to retransmit stream data ["
<< retransmission_offset << ", "
<< retransmission_offset + retransmission_length
<< ") and fin: " << can_bundle_fin
<< ", consumed: " << consumed;
OnStreamFrameRetransmitted(retransmission_offset, consumed.bytes_consumed,
consumed.fin_consumed);
if (can_bundle_fin) {
retransmit_fin = !consumed.fin_consumed;
}
if (consumed.bytes_consumed < retransmission_length ||
(can_bundle_fin && !consumed.fin_consumed)) {
// Connection is write blocked.
return false;
}
}
if (retransmit_fin) {
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_
<< " retransmits fin only frame.";
consumed = stream_delegate_->WritevData(
id_, 0, stream_bytes_written(), FIN, type,
session()->GetEncryptionLevelToSendApplicationData());
if (!consumed.fin_consumed) {
return false;
}
}
return true;
}
bool QuicStream::IsWaitingForAcks() const {
return (!rst_sent_ || stream_error_.ok()) &&
(send_buffer_.stream_bytes_outstanding() || fin_outstanding_);
}
QuicByteCount QuicStream::ReadableBytes() const {
return sequencer_.ReadableBytes();
}
bool QuicStream::WriteStreamData(QuicStreamOffset offset,
QuicByteCount data_length,
QuicDataWriter* writer) {
QUICHE_DCHECK_LT(0u, data_length);
QUIC_DVLOG(2) << ENDPOINT << "Write stream " << id_ << " data from offset "
<< offset << " length " << data_length;
return send_buffer_.WriteStreamData(offset, data_length, writer);
}
void QuicStream::WriteBufferedData(EncryptionLevel level) {
QUICHE_DCHECK(!write_side_closed_ && (HasBufferedData() || fin_buffered_));
if (session_->ShouldYield(id())) {
session_->MarkConnectionLevelWriteBlocked(id());
return;
}
// Size of buffered data.
QuicByteCount write_length = BufferedDataBytes();
// A FIN with zero data payload should not be flow control blocked.
bool fin_with_zero_data = (fin_buffered_ && write_length == 0);
bool fin = fin_buffered_;
// How much data flow control permits to be written.
QuicByteCount send_window;
if (flow_controller_.has_value()) {
send_window = flow_controller_->SendWindowSize();
} else {
send_window = std::numeric_limits<QuicByteCount>::max();
QUIC_BUG(quic_bug_10586_13)
<< ENDPOINT
<< "WriteBufferedData called on stream without flow control";
}
if (stream_contributes_to_connection_flow_control_) {
send_window =
std::min(send_window, connection_flow_controller_->SendWindowSize());
}
if (send_window == 0 && !fin_with_zero_data) {
// Quick return if nothing can be sent.
MaybeSendBlocked();
return;
}
if (write_length > send_window) {
// Don't send the FIN unless all the data will be sent.
fin = false;
// Writing more data would be a violation of flow control.
write_length = send_window;
QUIC_DVLOG(1) << "stream " << id() << " shortens write length to "
<< write_length << " due to flow control";
}
StreamSendingState state = fin ? FIN : NO_FIN;
if (fin && add_random_padding_after_fin_) {
state = FIN_AND_PADDING;
}
QuicConsumedData consumed_data =
stream_delegate_->WritevData(id(), write_length, stream_bytes_written(),
state, NOT_RETRANSMISSION, level);
OnStreamDataConsumed(consumed_data.bytes_consumed);
AddBytesSent(consumed_data.bytes_consumed);
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_ << " sends "
<< stream_bytes_written() << " bytes "
<< " and has buffered data " << BufferedDataBytes() << " bytes."
<< " fin is sent: " << consumed_data.fin_consumed
<< " fin is buffered: " << fin_buffered_;
// The write may have generated a write error causing this stream to be
// closed. If so, simply return without marking the stream write blocked.
if (write_side_closed_) {
return;
}
if (consumed_data.bytes_consumed == write_length) {
if (!fin_with_zero_data) {
MaybeSendBlocked();
}
if (fin && consumed_data.fin_consumed) {
QUICHE_DCHECK(!fin_sent_);
fin_sent_ = true;
fin_outstanding_ = true;
if (fin_received_) {
QUICHE_DCHECK(!was_draining_);
session_->StreamDraining(id_,
/*unidirectional=*/type_ != BIDIRECTIONAL);
was_draining_ = true;
}
CloseWriteSide();
} else if (fin && !consumed_data.fin_consumed) {
session_->MarkConnectionLevelWriteBlocked(id());
}
} else {
session_->MarkConnectionLevelWriteBlocked(id());
}
if (consumed_data.bytes_consumed > 0 || consumed_data.fin_consumed) {
busy_counter_ = 0;
}
}
uint64_t QuicStream::BufferedDataBytes() const {
QUICHE_DCHECK_GE(send_buffer_.stream_offset(), stream_bytes_written());
return send_buffer_.stream_offset() - stream_bytes_written();
}
bool QuicStream::CanWriteNewData() const {
return BufferedDataBytes() < buffered_data_threshold_;
}
bool QuicStream::CanWriteNewDataAfterData(QuicByteCount length) const {
return (BufferedDataBytes() + length) < buffered_data_threshold_;
}
uint64_t QuicStream::stream_bytes_written() const {
return send_buffer_.stream_bytes_written();
}
const QuicIntervalSet<QuicStreamOffset>& QuicStream::bytes_acked() const {
return send_buffer_.bytes_acked();
}
void QuicStream::OnStreamDataConsumed(QuicByteCount bytes_consumed) {
send_buffer_.OnStreamDataConsumed(bytes_consumed);
}
void QuicStream::WritePendingRetransmission() {
while (HasPendingRetransmission()) {
QuicConsumedData consumed(0, false);
if (!send_buffer_.HasPendingRetransmission()) {
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_
<< " retransmits fin only frame.";
consumed = stream_delegate_->WritevData(
id_, 0, stream_bytes_written(), FIN, LOSS_RETRANSMISSION,
session()->GetEncryptionLevelToSendApplicationData());
fin_lost_ = !consumed.fin_consumed;
if (fin_lost_) {
// Connection is write blocked.
return;
}
} else {
StreamPendingRetransmission pending =
send_buffer_.NextPendingRetransmission();
// Determine whether the lost fin can be bundled with the data.
const bool can_bundle_fin =
fin_lost_ &&
(pending.offset + pending.length == stream_bytes_written());
consumed = stream_delegate_->WritevData(
id_, pending.length, pending.offset, can_bundle_fin ? FIN : NO_FIN,
LOSS_RETRANSMISSION,
session()->GetEncryptionLevelToSendApplicationData());
QUIC_DVLOG(1) << ENDPOINT << "stream " << id_
<< " tries to retransmit stream data [" << pending.offset
<< ", " << pending.offset + pending.length
<< ") and fin: " << can_bundle_fin
<< ", consumed: " << consumed;
OnStreamFrameRetransmitted(pending.offset, consumed.bytes_consumed,
consumed.fin_consumed);
if (consumed.bytes_consumed < pending.length ||
(can_bundle_fin && !consumed.fin_consumed)) {
// Connection is write blocked.
return;
}
}
}
}
bool QuicStream::MaybeSetTtl(QuicTime::Delta ttl) {
if (is_static_) {
QUIC_BUG(quic_bug_10586_14) << "Cannot set TTL of a static stream.";
return false;
}
if (deadline_.IsInitialized()) {
QUIC_DLOG(WARNING) << "Deadline has already been set.";
return false;
}
QuicTime now = session()->connection()->clock()->ApproximateNow();
deadline_ = now + ttl;
return true;
}
bool QuicStream::HasDeadlinePassed() const {
if (!deadline_.IsInitialized()) {
// No deadline has been set.
return false;
}
QuicTime now = session()->connection()->clock()->ApproximateNow();
if (now < deadline_) {
return false;
}
// TTL expired.
QUIC_DVLOG(1) << "stream " << id() << " deadline has passed";
return true;
}
void QuicStream::OnDeadlinePassed() { Reset(QUIC_STREAM_TTL_EXPIRED); }
bool QuicStream::IsFlowControlBlocked() const {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_15)
<< "Trying to access non-existent flow controller.";
return false;
}
return flow_controller_->IsBlocked();
}
QuicStreamOffset QuicStream::highest_received_byte_offset() const {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_16)
<< "Trying to access non-existent flow controller.";
return 0;
}
return flow_controller_->highest_received_byte_offset();
}
void QuicStream::UpdateReceiveWindowSize(QuicStreamOffset size) {
if (!flow_controller_.has_value()) {
QUIC_BUG(quic_bug_10586_17)
<< "Trying to access non-existent flow controller.";
return;
}
flow_controller_->UpdateReceiveWindowSize(size);
}
// static
spdy::SpdyStreamPrecedence QuicStream::CalculateDefaultPriority(
const QuicSession* session) {
return spdy::SpdyStreamPrecedence(
VersionUsesHttp3(session->transport_version())
? kDefaultUrgency
: QuicStream::kDefaultPriority);
}
absl::optional<QuicByteCount> QuicStream::GetSendWindow() const {
return flow_controller_.has_value()
? absl::optional<QuicByteCount>(flow_controller_->SendWindowSize())
: absl::nullopt;
}
absl::optional<QuicByteCount> QuicStream::GetReceiveWindow() const {
return flow_controller_.has_value()
? absl::optional<QuicByteCount>(
flow_controller_->receive_window_size())
: absl::nullopt;
}
void QuicStream::OnStreamCreatedFromPendingStream() {
sequencer()->SetUnblocked();
}
} // namespace quic