Project import generated by Copybara. PiperOrigin-RevId: 237361882 Change-Id: I109a68f44db867b20f8c6a7732b0ce657133e52a
diff --git a/quic/core/quic_stream.cc b/quic/core/quic_stream.cc new file mode 100644 index 0000000..86207b6 --- /dev/null +++ b/quic/core/quic_stream.cc
@@ -0,0 +1,1101 @@ +// 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 "net/third_party/quiche/src/quic/core/quic_stream.h" + +#include "net/third_party/quiche/src/quic/core/quic_flow_controller.h" +#include "net/third_party/quiche/src/quic/core/quic_session.h" +#include "net/third_party/quiche/src/quic/core/quic_utils.h" +#include "net/third_party/quiche/src/quic/platform/api/quic_bug_tracker.h" +#include "net/third_party/quiche/src/quic/platform/api/quic_flag_utils.h" +#include "net/third_party/quiche/src/quic/platform/api/quic_flags.h" +#include "net/third_party/quiche/src/quic/platform/api/quic_logging.h" +#include "net/third_party/quiche/src/quic/platform/api/quic_str_cat.h" +#include "net/third_party/quiche/src/quic/platform/api/quic_string.h" + +using spdy::SpdyPriority; + +namespace quic { + +#define ENDPOINT \ + (perspective_ == Perspective::IS_SERVER ? "Server: " : "Client: ") + +namespace { + +size_t GetInitialStreamFlowControlWindowToSend(QuicSession* session) { + return session->config()->GetInitialStreamFlowControlWindowToSend(); +} + +size_t GetReceivedFlowControlWindow(QuicSession* session) { + if (session->config()->HasReceivedInitialStreamFlowControlWindowBytes()) { + return session->config()->ReceivedInitialStreamFlowControlWindowBytes(); + } + + return kMinimumFlowControlSendWindow; +} + +} // namespace + +// static +const SpdyPriority QuicStream::kDefaultPriority; + +PendingStream::PendingStream(QuicStreamId id, QuicSession* session) + : id_(id), + session_(session), + stream_bytes_read_(0), + fin_received_(false), + connection_flow_controller_(session->flow_controller()), + flow_controller_(session, + id, + /*is_connection_flow_controller*/ false, + GetReceivedFlowControlWindow(session), + GetInitialStreamFlowControlWindowToSend(session), + kStreamReceiveWindowLimit, + session_->flow_controller()->auto_tune_receive_window(), + session_->flow_controller()), + sequencer_(this) {} + +void PendingStream::OnDataAvailable() { + QUIC_BUG << "OnDataAvailable should not be called."; + CloseConnectionWithDetails(QUIC_INTERNAL_ERROR, "Unexpected data available"); +} + +void PendingStream::OnFinRead() { + QUIC_BUG << "OnFinRead should not be called."; + CloseConnectionWithDetails(QUIC_INTERNAL_ERROR, "Unexpected fin read"); +} + +void PendingStream::AddBytesConsumed(QuicByteCount bytes) { + QUIC_BUG << "AddBytesConsumed should not be called."; + CloseConnectionWithDetails(QUIC_INTERNAL_ERROR, "Unexpected bytes consumed"); +} + +void PendingStream::Reset(QuicRstStreamErrorCode error) { + session_->SendRstStream(id_, error, 0); +} + +void PendingStream::CloseConnectionWithDetails(QuicErrorCode error, + const QuicString& details) { + session_->connection()->CloseConnection( + error, details, ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET); +} + +QuicStreamId PendingStream::id() const { + return id_; +} + +const QuicSocketAddress& PendingStream::PeerAddressOfLatestPacket() const { + return session_->connection()->last_packet_source_address(); +} + +void PendingStream::OnStreamFrame(const QuicStreamFrame& frame) { + DCHECK_EQ(frame.stream_id, id_); + DCHECK_NE(0u, frame.offset); + + 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 + << "Receive stream frame reaches max stream length. frame offset " + << frame.offset << " length " << frame.data_length; + CloseConnectionWithDetails( + QUIC_STREAM_LENGTH_OVERFLOW, + "Peer sends more data than allowed on this stream."); + return; + } + + if (frame.fin) { + fin_received_ = true; + } + + // This count includes duplicate data received. + size_t 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()) { + CloseConnectionWithDetails( + QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA, + "Flow control violation after increasing offset"); + return; + } + } + + sequencer_.OnStreamFrame(frame); +} + +void PendingStream::OnRstStreamFrame(const QuicRstStreamFrame& frame) { + DCHECK_EQ(frame.stream_id, id_); + + if (frame.byte_offset > kMaxStreamLength) { + // Peer are not suppose to write bytes more than maxium allowed. + CloseConnectionWithDetails(QUIC_STREAM_LENGTH_OVERFLOW, + "Reset frame stream offset overflow."); + return; + } + MaybeIncreaseHighestReceivedOffset(frame.byte_offset); + if (flow_controller_.FlowControlViolation() || + connection_flow_controller_->FlowControlViolation()) { + CloseConnectionWithDetails( + QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA, + "Flow control violation after increasing offset"); + return; + } +} + +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; +} + +QuicStream::QuicStream(PendingStream pending, StreamType type) + : QuicStream(pending.id_, + pending.session_, + std::move(pending.sequencer_), + /*is_static=*/false, + type, + pending.stream_bytes_read_, + pending.fin_received_, + std::move(pending.flow_controller_), + pending.connection_flow_controller_) { + sequencer_.set_stream(this); +} + +QuicStream::QuicStream(QuicStreamId id, + QuicSession* session, + bool is_static, + StreamType type) + : QuicStream(id, + session, + QuicStreamSequencer(this), + is_static, + type, + 0, + false, + QuicFlowController( + session, + id, + /*is_connection_flow_controller*/ false, + GetReceivedFlowControlWindow(session), + GetInitialStreamFlowControlWindowToSend(session), + kStreamReceiveWindowLimit, + session->flow_controller()->auto_tune_receive_window(), + session->flow_controller()), + session->flow_controller()) {} + +QuicStream::QuicStream(QuicStreamId id, + QuicSession* session, + QuicStreamSequencer sequencer, + bool is_static, + StreamType type, + uint64_t stream_bytes_read, + bool fin_received, + QuicFlowController flow_controller, + QuicFlowController* connection_flow_controller) + : sequencer_(std::move(sequencer)), + id_(id), + session_(session), + priority_(kDefaultPriority), + stream_bytes_read_(stream_bytes_read), + stream_error_(QUIC_STREAM_NO_ERROR), + connection_error_(QUIC_NO_ERROR), + read_side_closed_(false), + write_side_closed_(false), + fin_buffered_(false), + fin_sent_(false), + fin_outstanding_(false), + fin_lost_(false), + fin_received_(fin_received), + rst_sent_(false), + rst_received_(false), + perspective_(session_->perspective()), + 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()), + type_(session->connection()->transport_version() == QUIC_VERSION_99 + ? QuicUtils::GetStreamType(id_, + perspective_, + session->IsIncomingStream(id_)) + : type) { + if (type_ == WRITE_UNIDIRECTIONAL) { + set_fin_received(true); + CloseReadSide(); + } else if (type_ == READ_UNIDIRECTIONAL) { + set_fin_sent(true); + CloseWriteSide(); + } + SetFromConfig(); + session_->RegisterStreamPriority(id, is_static_, priority_); +} + +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 (session_ != nullptr) { + session_->UnregisterStreamPriority(id(), is_static_); + } +} + +void QuicStream::SetFromConfig() {} + +void QuicStream::OnStreamFrame(const QuicStreamFrame& frame) { + DCHECK_EQ(frame.stream_id, id_); + + DCHECK(!(read_side_closed_ && write_side_closed_)); + + if (type_ == WRITE_UNIDIRECTIONAL) { + CloseConnectionWithDetails( + 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 << "Receive stream frame on stream " << id_ + << " reaches max stream length. frame offset " << frame.offset + << " length " << frame.data_length << ". " + << sequencer_.DebugString(); + CloseConnectionWithDetails( + QUIC_STREAM_LENGTH_OVERFLOW, + QuicStrCat("Peer sends more data than allowed on stream ", id_, + ". frame: offset = ", frame.offset, ", length = ", + frame.data_length, ". ", sequencer_.DebugString())); + return; + } + if (frame.fin) { + fin_received_ = true; + if (fin_sent_) { + session_->StreamDraining(id_); + } + } + + 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. + size_t 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()) { + CloseConnectionWithDetails( + QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA, + "Flow control violation after increasing offset"); + return; + } + } + + sequencer_.OnStreamFrame(frame); +} + +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. + CloseConnectionWithDetails(QUIC_STREAM_LENGTH_OVERFLOW, + "Reset frame stream offset overflow."); + return; + } + MaybeIncreaseHighestReceivedOffset(frame.byte_offset); + if (flow_controller_.FlowControlViolation() || + connection_flow_controller_->FlowControlViolation()) { + CloseConnectionWithDetails( + QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA, + "Flow control violation after increasing offset"); + return; + } + + stream_error_ = frame.error_code; + // Google QUIC closes both sides of the stream in response to a + // RESET_STREAM, IETF QUIC closes only the read side. + if (transport_version() != QUIC_VERSION_99) { + CloseWriteSide(); + } + CloseReadSide(); +} + +void QuicStream::OnConnectionClosed(QuicErrorCode error, + ConnectionCloseSource /*source*/) { + if (read_side_closed_ && write_side_closed_) { + return; + } + if (error != QUIC_NO_ERROR) { + stream_error_ = QUIC_STREAM_CONNECTION_ERROR; + connection_error_ = error; + } + + CloseWriteSide(); + CloseReadSide(); +} + +void QuicStream::OnFinRead() { + 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::Reset(QuicRstStreamErrorCode error) { + stream_error_ = error; + // Sending a RstStream results in calling CloseStream. + session()->SendRstStream(id(), error, stream_bytes_written()); + rst_sent_ = true; +} + +void QuicStream::CloseConnectionWithDetails(QuicErrorCode error, + const QuicString& details) { + session()->connection()->CloseConnection( + error, details, ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET); +} + +SpdyPriority QuicStream::priority() const { + return priority_; +} + +void QuicStream::SetPriority(SpdyPriority priority) { + priority_ = priority; + session_->UpdateStreamPriority(id(), priority); +} + +void QuicStream::WriteOrBufferData( + QuicStringPiece data, + bool fin, + QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener) { + if (data.empty() && !fin) { + QUIC_BUG << "data.empty() && !fin"; + return; + } + + if (fin_buffered_) { + QUIC_BUG << "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) { + CloseConnectionWithDetails( + QUIC_TRY_TO_WRITE_DATA_ON_READ_UNIDIRECTIONAL_STREAM, + "Try to send data on read unidirectional stream"); + } + return; + } + + QuicConsumedData consumed_data(0, false); + 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 << "Write too many data via stream " << id_; + CloseConnectionWithDetails( + QUIC_STREAM_LENGTH_OVERFLOW, + QuicStrCat("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(); + } +} + +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(); + } + 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_.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::WritevData(const struct iovec* iov, + int iov_count, + bool fin) { + if (write_side_closed_) { + QUIC_DLOG(ERROR) << ENDPOINT << "Stream " << id() + << "attempting to write when the write side is closed"; + if (type_ == READ_UNIDIRECTIONAL) { + CloseConnectionWithDetails( + QUIC_TRY_TO_WRITE_DATA_ON_READ_UNIDIRECTIONAL_STREAM, + "Try to send data on read unidirectional stream"); + } + return QuicConsumedData(0, false); + } + + // How much data was provided. + size_t write_length = 0; + if (iov != nullptr) { + for (int i = 0; i < iov_count; ++i) { + write_length += iov[i].iov_len; + } + } + + QuicConsumedData consumed_data(0, false); + if (fin_buffered_) { + QUIC_BUG << "Fin already buffered"; + return consumed_data; + } + + if (kMaxStreamLength - send_buffer_.stream_offset() < write_length) { + QUIC_BUG << "Write too many data via stream " << id_; + CloseConnectionWithDetails( + QUIC_STREAM_LENGTH_OVERFLOW, + QuicStrCat("Write too many data via stream ", id_)); + return consumed_data; + } + + bool had_buffered_data = HasBufferedData(); + if (CanWriteNewData()) { + // Save all data if buffered data size is below low water mark. + consumed_data.bytes_consumed = write_length; + if (consumed_data.bytes_consumed > 0) { + QuicStreamOffset offset = send_buffer_.stream_offset(); + send_buffer_.SaveStreamData(iov, iov_count, 0, write_length); + OnDataBuffered(offset, write_length, nullptr); + } + } + consumed_data.fin_consumed = + consumed_data.bytes_consumed == write_length && fin; + fin_buffered_ = consumed_data.fin_consumed; + + if (!had_buffered_data && (HasBufferedData() || fin_buffered_)) { + // Write data if there is no buffered data before. + WriteBufferedData(); + } + + return consumed_data; +} + +QuicConsumedData QuicStream::WriteMemSlices(QuicMemSliceSpan span, bool fin) { + QuicConsumedData consumed_data(0, false); + if (span.empty() && !fin) { + QUIC_BUG << "span.empty() && !fin"; + return consumed_data; + } + + if (fin_buffered_) { + QUIC_BUG << "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) { + CloseConnectionWithDetails( + 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 = + span.SaveMemSlicesInSendBuffer(&send_buffer_); + if (offset > send_buffer_.stream_offset() || + kMaxStreamLength < send_buffer_.stream_offset()) { + QUIC_BUG << "Write too many data via stream " << id_; + CloseConnectionWithDetails( + QUIC_STREAM_LENGTH_OVERFLOW, + QuicStrCat("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(); + } + + 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_); +} + +QuicConsumedData QuicStream::WritevDataInner(size_t write_length, + QuicStreamOffset offset, + bool fin) { + StreamSendingState state = fin ? FIN : NO_FIN; + if (fin && add_random_padding_after_fin_) { + state = FIN_AND_PADDING; + } + return session()->WritevData(this, id(), write_length, offset, state); +} + +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_->CloseStream(id()); + } +} + +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_->CloseStream(id()); + } +} + +bool QuicStream::HasBufferedData() const { + DCHECK_GE(send_buffer_.stream_offset(), stream_bytes_written()); + return send_buffer_.stream_offset() > stream_bytes_written(); +} + +QuicTransportVersion QuicStream::transport_version() const { + return session_->connection()->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(); +} + +const QuicSocketAddress& QuicStream::PeerAddressOfLatestPacket() const { + return session_->connection()->last_packet_source_address(); +} + +void QuicStream::OnClose() { + CloseReadSide(); + CloseWriteSide(); + + if (!fin_sent_ && !rst_sent_) { + // 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. + QUIC_DLOG(INFO) << ENDPOINT << "Sending RST_STREAM in OnClose: " << id(); + session_->SendRstStream(id(), QUIC_RST_ACKNOWLEDGEMENT, + stream_bytes_written()); + session_->OnStreamDoneWaitingForAcks(id_); + rst_sent_ = true; + } + + if (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 (flow_controller_.UpdateSendWindowOffset(frame.byte_offset)) { + // Let session unblock this stream. + session_->MarkConnectionLevelWriteBlocked(id_); + } +} + +bool QuicStream::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. + 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) { + flow_controller_.AddBytesSent(bytes); + if (stream_contributes_to_connection_flow_control_) { + connection_flow_controller_->AddBytesSent(bytes); + } +} + +void QuicStream::AddBytesConsumed(QuicByteCount bytes) { + // 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); + } +} + +void QuicStream::UpdateSendWindowOffset(QuicStreamOffset new_window) { + if (flow_controller_.UpdateSendWindowOffset(new_window)) { + // Let session unblock this stream. + session_->MarkConnectionLevelWriteBlocked(id_); + } +} + +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, + 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)) { + CloseConnectionWithDetails(QUIC_INTERNAL_ERROR, + "Trying to ack unsent data."); + return false; + } + if (!fin_sent_ && fin_acked) { + CloseConnectionWithDetails(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()) { + session_->OnStreamDoneWaitingForAcks(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) { + 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 = session()->WritevData(this, id_, retransmission_length, + retransmission_offset, + can_bundle_fin ? FIN : NO_FIN); + 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 = session()->WritevData(this, id_, 0, stream_bytes_written(), FIN); + if (!consumed.fin_consumed) { + return false; + } + } + return true; +} + +bool QuicStream::IsWaitingForAcks() const { + return (!rst_sent_ || stream_error_ == QUIC_STREAM_NO_ERROR) && + (send_buffer_.stream_bytes_outstanding() || fin_outstanding_); +} + +size_t QuicStream::ReadableBytes() const { + return sequencer_.ReadableBytes(); +} + +bool QuicStream::WriteStreamData(QuicStreamOffset offset, + QuicByteCount data_length, + QuicDataWriter* writer) { + 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() { + DCHECK(!write_side_closed_ && (HasBufferedData() || fin_buffered_)); + + if (session_->ShouldYield(id())) { + session_->MarkConnectionLevelWriteBlocked(id()); + return; + } + + // Size of buffered data. + size_t 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 = flow_controller_.SendWindowSize(); + 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 = static_cast<size_t>(send_window); + QUIC_DVLOG(1) << "stream " << id() << " shortens write length to " + << write_length << " due to flow control"; + } + if (session_->session_decides_what_to_write()) { + session_->SetTransmissionType(NOT_RETRANSMISSION); + } + QuicConsumedData consumed_data = + WritevDataInner(write_length, stream_bytes_written(), fin); + + 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) { + fin_sent_ = true; + fin_outstanding_ = true; + if (fin_received_) { + session_->StreamDraining(id_); + } + 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 { + 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(size_t 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 = + session()->WritevData(this, id_, 0, stream_bytes_written(), FIN); + 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 = + session()->WritevData(this, id_, pending.length, pending.offset, + can_bundle_fin ? FIN : NO_FIN); + 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 << "Cannot set TTL of a static stream."; + return false; + } + if (deadline_.IsInitialized()) { + QUIC_DLOG(WARNING) << "Deadline has already been set."; + return false; + } + if (!session()->session_decides_what_to_write()) { + QUIC_DLOG(WARNING) << "This session does not support stream TTL yet."; + 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; + } + DCHECK(session()->session_decides_what_to_write()); + 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); +} + +void QuicStream::SendStopSending(uint16_t code) { + if (transport_version() != QUIC_VERSION_99) { + // If the connection is not version 99, do nothing. + // Do not QUIC_BUG or anything; the application really does not need to know + // what version the connection is in. + return; + } + session_->SendStopSending(code, id_); +} + +void QuicStream::OnStopSending(uint16_t code) {} + +} // namespace quic