Google Git
Sign in
quiche / quiche / refs/heads/roll/20190423c / . / quic / core / quic_session.cc
blob: d780961d0970d7efc4fc3aa0d3b2389df5c8b884 [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 "net/third_party/quiche/src/quic/core/quic_session.h"
#include <cstdint>
#include <string>
#include <utility>
#include "net/third_party/quiche/src/quic/core/quic_connection.h"
#include "net/third_party/quiche/src/quic/core/quic_flow_controller.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_map_util.h"
#include "net/third_party/quiche/src/quic/platform/api/quic_server_stats.h"
#include "net/third_party/quiche/src/quic/platform/api/quic_stack_trace.h"
#include "net/third_party/quiche/src/quic/platform/api/quic_str_cat.h"
using spdy::SpdyPriority;
namespace quic {
namespace {
class ClosedStreamsCleanUpDelegate : public QuicAlarm::Delegate {
public:
explicit ClosedStreamsCleanUpDelegate(QuicSession* session)
: session_(session) {}
ClosedStreamsCleanUpDelegate(const ClosedStreamsCleanUpDelegate&) = delete;
ClosedStreamsCleanUpDelegate& operator=(const ClosedStreamsCleanUpDelegate&) =
delete;
void OnAlarm() override { session_->CleanUpClosedStreams(); }
private:
QuicSession* session_;
};
} // namespace
#define ENDPOINT \
(perspective() == Perspective::IS_SERVER ? "Server: " : "Client: ")
QuicSession::QuicSession(QuicConnection* connection,
Visitor* owner,
const QuicConfig& config,
const ParsedQuicVersionVector& supported_versions)
: connection_(connection),
visitor_(owner),
write_blocked_streams_(),
config_(config),
stream_id_manager_(this,
kDefaultMaxStreamsPerConnection,
config_.GetMaxIncomingDynamicStreamsToSend()),
v99_streamid_manager_(this,
kDefaultMaxStreamsPerConnection,
config_.GetMaxIncomingDynamicStreamsToSend()),
num_dynamic_incoming_streams_(0),
num_draining_incoming_streams_(0),
num_outgoing_static_streams_(0),
num_incoming_static_streams_(0),
num_locally_closed_incoming_streams_highest_offset_(0),
error_(QUIC_NO_ERROR),
flow_controller_(
this,
QuicUtils::GetInvalidStreamId(connection->transport_version()),
/*is_connection_flow_controller*/ true,
kMinimumFlowControlSendWindow,
config_.GetInitialSessionFlowControlWindowToSend(),
kSessionReceiveWindowLimit,
perspective() == Perspective::IS_SERVER,
nullptr),
currently_writing_stream_id_(0),
largest_static_stream_id_(0),
is_handshake_confirmed_(false),
goaway_sent_(false),
goaway_received_(false),
control_frame_manager_(this),
last_message_id_(0),
closed_streams_clean_up_alarm_(nullptr),
supported_versions_(supported_versions) {
closed_streams_clean_up_alarm_ =
QuicWrapUnique<QuicAlarm>(connection_->alarm_factory()->CreateAlarm(
new ClosedStreamsCleanUpDelegate(this)));
}
void QuicSession::Initialize() {
connection_->set_visitor(this);
connection_->SetSessionNotifier(this);
connection_->SetDataProducer(this);
connection_->SetFromConfig(config_);
DCHECK_EQ(QuicUtils::GetCryptoStreamId(connection_->transport_version()),
GetMutableCryptoStream()->id());
RegisterStaticStream(
QuicUtils::GetCryptoStreamId(connection_->transport_version()),
GetMutableCryptoStream());
}
QuicSession::~QuicSession() {
QUIC_LOG_IF(WARNING, !zombie_streams_.empty()) << "Still have zombie streams";
}
void QuicSession::RegisterStaticStream(QuicStreamId id, QuicStream* stream) {
static_stream_map_[id] = stream;
QUIC_BUG_IF(id >
largest_static_stream_id_ +
QuicUtils::StreamIdDelta(connection_->transport_version()))
<< ENDPOINT << "Static stream registered out of order: " << id
<< " vs: " << largest_static_stream_id_;
largest_static_stream_id_ = std::max(id, largest_static_stream_id_);
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.RegisterStaticStream(id);
}
}
void QuicSession::RegisterStaticStreamNew(std::unique_ptr<QuicStream> stream) {
DCHECK(GetQuicReloadableFlag(quic_eliminate_static_stream_map));
QuicStreamId stream_id = stream->id();
dynamic_stream_map_[stream_id] = std::move(stream);
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.RegisterStaticStream(stream_id);
}
if (IsIncomingStream(stream_id)) {
++num_incoming_static_streams_;
} else {
++num_outgoing_static_streams_;
}
}
void QuicSession::OnStreamFrame(const QuicStreamFrame& frame) {
// TODO(rch) deal with the error case of stream id 0.
QuicStreamId stream_id = frame.stream_id;
if (stream_id ==
QuicUtils::GetInvalidStreamId(connection()->transport_version())) {
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Received data for an invalid stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
if (frame.fin && QuicContainsKey(static_stream_map_, stream_id)) {
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Attempt to close a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
StreamHandler handler = GetOrCreateStreamImpl(stream_id, frame.offset != 0);
if (handler.is_pending) {
handler.pending->OnStreamFrame(frame);
return;
}
if (!handler.stream) {
// The stream no longer exists, but we may still be interested in the
// final stream byte offset sent by the peer. A frame with a FIN can give
// us this offset.
if (frame.fin) {
QuicStreamOffset final_byte_offset = frame.offset + frame.data_length;
OnFinalByteOffsetReceived(stream_id, final_byte_offset);
}
return;
}
if (GetQuicReloadableFlag(quic_eliminate_static_stream_map) && frame.fin &&
handler.stream->is_static()) {
QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map, 1, 9);
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Attempt to close a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
handler.stream->OnStreamFrame(frame);
}
void QuicSession::OnCryptoFrame(const QuicCryptoFrame& frame) {
GetMutableCryptoStream()->OnCryptoFrame(frame);
}
bool QuicSession::OnStopSendingFrame(const QuicStopSendingFrame& frame) {
// We are not version 99. In theory, if not in version 99 then the framer
// could not call OnStopSending... This is just a check that is good when
// both a new protocol and a new implementation of that protocol are both
// being developed.
DCHECK_EQ(QUIC_VERSION_99, connection_->transport_version());
QuicStreamId stream_id = frame.stream_id;
// If Stream ID is invalid then close the connection.
if (stream_id ==
QuicUtils::GetInvalidStreamId(connection()->transport_version())) {
QUIC_DVLOG(1) << ENDPOINT
<< "Received STOP_SENDING with invalid stream_id: "
<< stream_id << " Closing connection";
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Received STOP_SENDING for an invalid stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return false;
}
// Ignore STOP_SENDING for static streams.
// TODO(fkastenholz): IETF Quic does not have static streams and does not
// make exceptions for them with respect to processing things like
// STOP_SENDING.
if (QuicContainsKey(static_stream_map_, stream_id)) {
QUIC_DVLOG(1) << ENDPOINT
<< "Received STOP_SENDING for a static stream, id: "
<< stream_id << " Closing connection";
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Received STOP_SENDING for a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return false;
}
if (visitor_) {
visitor_->OnStopSendingReceived(frame);
}
// If stream is closed, ignore the frame
if (IsClosedStream(stream_id)) {
QUIC_DVLOG(1)
<< ENDPOINT
<< "Received STOP_SENDING for closed or non-existent stream, id: "
<< stream_id << " Ignoring.";
return true; // Continue processing the packet.
}
// If stream is non-existent, close the connection
DynamicStreamMap::iterator it = dynamic_stream_map_.find(stream_id);
if (it == dynamic_stream_map_.end()) {
QUIC_DVLOG(1) << ENDPOINT
<< "Received STOP_SENDING for non-existent stream, id: "
<< stream_id << " Closing connection";
connection()->CloseConnection(
IETF_QUIC_PROTOCOL_VIOLATION,
"Received STOP_SENDING for a non-existent stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return false;
}
// Get the QuicStream for this stream. Ignore the STOP_SENDING
// if the QuicStream pointer is NULL
// QUESTION(fkastenholz): IS THIS THE RIGHT THING TO DO? (that is, this would
// happen IFF there was an entry in the map, but the pointer is null. sounds
// more like a deep programming error rather than a simple protocol problem).
QuicStream* stream = it->second.get();
if (stream == nullptr) {
QUIC_BUG << ENDPOINT
<< "Received STOP_SENDING for NULL QuicStream, stream_id: "
<< stream_id << ". Ignoring.";
return true;
}
if (GetQuicReloadableFlag(quic_eliminate_static_stream_map) &&
stream->is_static()) {
QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map, 2, 9);
QUIC_DVLOG(1) << ENDPOINT
<< "Received STOP_SENDING for a static stream, id: "
<< stream_id << " Closing connection";
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Received STOP_SENDING for a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return false;
}
stream->OnStopSending(frame.application_error_code);
stream->set_stream_error(
static_cast<QuicRstStreamErrorCode>(frame.application_error_code));
SendRstStreamInner(
stream->id(),
static_cast<quic::QuicRstStreamErrorCode>(frame.application_error_code),
stream->stream_bytes_written(),
/*close_write_side_only=*/true);
return true;
}
void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
QuicStreamId stream_id = frame.stream_id;
if (stream_id ==
QuicUtils::GetInvalidStreamId(connection()->transport_version())) {
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Received data for an invalid stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
if (QuicContainsKey(static_stream_map_, stream_id)) {
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Attempt to reset a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
if (visitor_) {
visitor_->OnRstStreamReceived(frame);
}
// may_buffer is true here to allow subclasses to buffer streams until the
// first byte of payload arrives which would allow sessions to delay
// creation of the stream until the type is known.
StreamHandler handler = GetOrCreateStreamImpl(stream_id, /*may_buffer=*/true);
if (handler.is_pending) {
handler.pending->OnRstStreamFrame(frame);
ClosePendingStream(stream_id);
return;
}
if (!handler.stream) {
HandleRstOnValidNonexistentStream(frame);
return; // Errors are handled by GetOrCreateStream.
}
if (GetQuicReloadableFlag(quic_eliminate_static_stream_map) &&
handler.stream->is_static()) {
QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map, 3, 9);
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Attempt to reset a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
handler.stream->OnStreamReset(frame);
}
void QuicSession::OnGoAway(const QuicGoAwayFrame& frame) {
goaway_received_ = true;
}
void QuicSession::OnMessageReceived(QuicStringPiece message) {
QUIC_DVLOG(1) << ENDPOINT << "Received message, length: " << message.length()
<< ", " << message;
}
// static
void QuicSession::RecordConnectionCloseAtServer(QuicErrorCode error,
ConnectionCloseSource source) {
if (error != QUIC_NO_ERROR) {
if (source == ConnectionCloseSource::FROM_SELF) {
QUIC_SERVER_HISTOGRAM_ENUM(
"quic_server_connection_close_errors", error, QUIC_LAST_ERROR,
"QuicErrorCode for server-closed connections.");
} else {
QUIC_SERVER_HISTOGRAM_ENUM(
"quic_client_connection_close_errors", error, QUIC_LAST_ERROR,
"QuicErrorCode for client-closed connections.");
}
}
}
void QuicSession::OnConnectionClosed(QuicErrorCode error,
const std::string& error_details,
ConnectionCloseSource source) {
DCHECK(!connection_->connected());
if (perspective() == Perspective::IS_SERVER) {
RecordConnectionCloseAtServer(error, source);
}
if (error_ == QUIC_NO_ERROR) {
error_ = error;
}
if (!GetQuicReloadableFlag(quic_eliminate_static_stream_map)) {
while (!dynamic_stream_map_.empty()) {
DynamicStreamMap::iterator it = dynamic_stream_map_.begin();
QuicStreamId id = it->first;
it->second->OnConnectionClosed(error, source);
// The stream should call CloseStream as part of OnConnectionClosed.
if (dynamic_stream_map_.find(id) != dynamic_stream_map_.end()) {
QUIC_BUG << ENDPOINT << "Stream " << id
<< " failed to close under OnConnectionClosed";
CloseStream(id);
}
}
} else {
QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map, 4, 9);
// Copy all non static streams in a new map for the ease of deleting.
QuicSmallMap<QuicStreamId, QuicStream*, 10> non_static_streams;
for (const auto& it : dynamic_stream_map_) {
if (!it.second->is_static()) {
non_static_streams[it.first] = it.second.get();
}
}
for (const auto& it : non_static_streams) {
QuicStreamId id = it.first;
it.second->OnConnectionClosed(error, source);
if (dynamic_stream_map_.find(id) != dynamic_stream_map_.end()) {
QUIC_BUG << ENDPOINT << "Stream " << id
<< " failed to close under OnConnectionClosed";
CloseStream(id);
}
}
}
// Cleanup zombie stream map on connection close.
while (!zombie_streams_.empty()) {
ZombieStreamMap::iterator it = zombie_streams_.begin();
closed_streams_.push_back(std::move(it->second));
zombie_streams_.erase(it);
}
closed_streams_clean_up_alarm_->Cancel();
if (visitor_) {
visitor_->OnConnectionClosed(connection_->connection_id(), error,
error_details, source);
}
}
void QuicSession::OnWriteBlocked() {
if (!connection_->connected()) {
return;
}
if (visitor_) {
visitor_->OnWriteBlocked(connection_);
}
}
void QuicSession::OnSuccessfulVersionNegotiation(
const ParsedQuicVersion& version) {
GetMutableCryptoStream()->OnSuccessfulVersionNegotiation(version);
}
void QuicSession::OnConnectivityProbeReceived(
const QuicSocketAddress& self_address,
const QuicSocketAddress& peer_address) {
if (perspective() == Perspective::IS_SERVER) {
// Server only sends back a connectivity probe after received a
// connectivity probe from a new peer address.
connection_->SendConnectivityProbingResponsePacket(peer_address);
}
}
void QuicSession::OnPathDegrading() {}
bool QuicSession::AllowSelfAddressChange() const {
return false;
}
void QuicSession::OnForwardProgressConfirmed() {}
void QuicSession::OnWindowUpdateFrame(const QuicWindowUpdateFrame& frame) {
// Stream may be closed by the time we receive a WINDOW_UPDATE, so we can't
// assume that it still exists.
QuicStreamId stream_id = frame.stream_id;
if (stream_id ==
QuicUtils::GetInvalidStreamId(connection_->transport_version())) {
// This is a window update that applies to the connection, rather than an
// individual stream.
QUIC_DLOG(INFO) << ENDPOINT
<< "Received connection level flow control window "
"update with byte offset: "
<< frame.byte_offset;
flow_controller_.UpdateSendWindowOffset(frame.byte_offset);
return;
}
QuicStream* stream = GetOrCreateStream(stream_id);
if (stream != nullptr) {
stream->OnWindowUpdateFrame(frame);
}
}
void QuicSession::OnBlockedFrame(const QuicBlockedFrame& frame) {
// TODO(rjshade): Compare our flow control receive windows for specified
// streams: if we have a large window then maybe something
// had gone wrong with the flow control accounting.
QUIC_DLOG(INFO) << ENDPOINT << "Received BLOCKED frame with stream id: "
<< frame.stream_id;
}
bool QuicSession::CheckStreamNotBusyLooping(QuicStream* stream,
uint64_t previous_bytes_written,
bool previous_fin_sent) {
if ( // Stream should not be closed.
!stream->write_side_closed() &&
// Not connection flow control blocked.
!flow_controller_.IsBlocked() &&
// Detect lack of forward progress.
previous_bytes_written == stream->stream_bytes_written() &&
previous_fin_sent == stream->fin_sent()) {
stream->set_busy_counter(stream->busy_counter() + 1);
QUIC_DVLOG(1) << "Suspected busy loop on stream id " << stream->id()
<< " stream_bytes_written " << stream->stream_bytes_written()
<< " fin " << stream->fin_sent() << " count "
<< stream->busy_counter();
// Wait a few iterations before firing, the exact count is
// arbitrary, more than a few to cover a few test-only false
// positives.
if (stream->busy_counter() > 20) {
QUIC_LOG(ERROR) << "Detected busy loop on stream id " << stream->id()
<< " stream_bytes_written "
<< stream->stream_bytes_written() << " fin "
<< stream->fin_sent();
return false;
}
} else {
stream->set_busy_counter(0);
}
return true;
}
bool QuicSession::CheckStreamWriteBlocked(QuicStream* stream) const {
if (!stream->write_side_closed() && stream->HasBufferedData() &&
!stream->flow_controller()->IsBlocked() &&
!write_blocked_streams_.IsStreamBlocked(stream->id())) {
QUIC_DLOG(ERROR) << "stream " << stream->id() << " has buffered "
<< stream->BufferedDataBytes()
<< " bytes, and is not flow control blocked, "
"but it is not in the write block list.";
return false;
}
return true;
}
void QuicSession::OnCanWrite() {
if (!RetransmitLostData()) {
// Cannot finish retransmitting lost data, connection is write blocked.
QUIC_DVLOG(1) << ENDPOINT
<< "Cannot finish retransmitting lost data, connection is "
"write blocked.";
return;
}
if (session_decides_what_to_write()) {
SetTransmissionType(NOT_RETRANSMISSION);
}
// We limit the number of writes to the number of pending streams. If more
// streams become pending, WillingAndAbleToWrite will be true, which will
// cause the connection to request resumption before yielding to other
// connections.
// If we are connection level flow control blocked, then only allow the
// crypto and headers streams to try writing as all other streams will be
// blocked.
size_t num_writes = flow_controller_.IsBlocked()
? write_blocked_streams_.NumBlockedSpecialStreams()
: write_blocked_streams_.NumBlockedStreams();
if (num_writes == 0 && !control_frame_manager_.WillingToWrite()) {
return;
}
QuicConnection::ScopedPacketFlusher flusher(
connection_, QuicConnection::SEND_ACK_IF_QUEUED);
if (control_frame_manager_.WillingToWrite()) {
control_frame_manager_.OnCanWrite();
}
for (size_t i = 0; i < num_writes; ++i) {
if (!(write_blocked_streams_.HasWriteBlockedSpecialStream() ||
write_blocked_streams_.HasWriteBlockedDataStreams())) {
// Writing one stream removed another!? Something's broken.
QUIC_BUG << "WriteBlockedStream is missing";
connection_->CloseConnection(QUIC_INTERNAL_ERROR,
"WriteBlockedStream is missing",
ConnectionCloseBehavior::SILENT_CLOSE);
return;
}
if (!connection_->CanWriteStreamData()) {
return;
}
currently_writing_stream_id_ = write_blocked_streams_.PopFront();
QuicStream* stream = GetOrCreateStream(currently_writing_stream_id_);
if (stream != nullptr && !stream->flow_controller()->IsBlocked()) {
// If the stream can't write all bytes it'll re-add itself to the blocked
// list.
uint64_t previous_bytes_written = stream->stream_bytes_written();
bool previous_fin_sent = stream->fin_sent();
QUIC_DVLOG(1) << "stream " << stream->id() << " bytes_written "
<< previous_bytes_written << " fin " << previous_fin_sent;
stream->OnCanWrite();
DCHECK(CheckStreamWriteBlocked(stream));
DCHECK(CheckStreamNotBusyLooping(stream, previous_bytes_written,
previous_fin_sent));
}
currently_writing_stream_id_ = 0;
}
}
bool QuicSession::WillingAndAbleToWrite() const {
// Schedule a write when:
// 1) control frame manager has pending or new control frames, or
// 2) any stream has pending retransmissions, or
// 3) If the crypto or headers streams are blocked, or
// 4) connection is not flow control blocked and there are write blocked
// streams.
return control_frame_manager_.WillingToWrite() ||
!streams_with_pending_retransmission_.empty() ||
write_blocked_streams_.HasWriteBlockedSpecialStream() ||
(!flow_controller_.IsBlocked() &&
write_blocked_streams_.HasWriteBlockedDataStreams());
}
bool QuicSession::HasPendingHandshake() const {
return QuicContainsKey(
streams_with_pending_retransmission_,
QuicUtils::GetCryptoStreamId(connection_->transport_version())) ||
write_blocked_streams_.IsStreamBlocked(
QuicUtils::GetCryptoStreamId(connection_->transport_version()));
}
uint64_t QuicSession::GetNumOpenDynamicStreams() const {
return dynamic_stream_map_.size() - draining_streams_.size() +
locally_closed_streams_highest_offset_.size() -
num_incoming_static_streams_ - num_outgoing_static_streams_;
}
void QuicSession::ProcessUdpPacket(const QuicSocketAddress& self_address,
const QuicSocketAddress& peer_address,
const QuicReceivedPacket& packet) {
connection_->ProcessUdpPacket(self_address, peer_address, packet);
}
QuicConsumedData QuicSession::WritevData(QuicStream* stream,
QuicStreamId id,
size_t write_length,
QuicStreamOffset offset,
StreamSendingState state) {
// This check is an attempt to deal with potential memory corruption
// in which |id| ends up set to 1 (the crypto stream id). If this happen
// it might end up resulting in unencrypted stream data being sent.
// While this is impossible to avoid given sufficient corruption, this
// seems like a reasonable mitigation.
if (id == QuicUtils::GetCryptoStreamId(connection_->transport_version()) &&
stream != GetMutableCryptoStream()) {
QUIC_BUG << "Stream id mismatch";
connection_->CloseConnection(
QUIC_INTERNAL_ERROR,
"Non-crypto stream attempted to write data as crypto stream.",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return QuicConsumedData(0, false);
}
if (!IsEncryptionEstablished() &&
id != QuicUtils::GetCryptoStreamId(connection_->transport_version())) {
// Do not let streams write without encryption. The calling stream will end
// up write blocked until OnCanWrite is next called.
return QuicConsumedData(0, false);
}
QuicConsumedData data =
connection_->SendStreamData(id, write_length, offset, state);
if (offset >= stream->stream_bytes_written()) {
// This is new stream data.
write_blocked_streams_.UpdateBytesForStream(id, data.bytes_consumed);
}
return data;
}
bool QuicSession::WriteControlFrame(const QuicFrame& frame) {
return connection_->SendControlFrame(frame);
}
void QuicSession::SendRstStream(QuicStreamId id,
QuicRstStreamErrorCode error,
QuicStreamOffset bytes_written) {
SendRstStreamInner(id, error, bytes_written, /*close_write_side_only=*/false);
}
void QuicSession::SendRstStreamInner(QuicStreamId id,
QuicRstStreamErrorCode error,
QuicStreamOffset bytes_written,
bool close_write_side_only) {
if (connection()->connected()) {
// Only send if still connected.
if (close_write_side_only) {
DCHECK_EQ(QUIC_VERSION_99, connection_->transport_version());
// Send a RST_STREAM frame.
control_frame_manager_.WriteOrBufferRstStream(id, error, bytes_written);
} else {
// Send a RST_STREAM frame plus, if version 99, an IETF
// QUIC STOP_SENDING frame. Both sre sent to emulate
// the two-way close that Google QUIC's RST_STREAM does.
if (connection_->transport_version() == QUIC_VERSION_99) {
QuicConnection::ScopedPacketFlusher flusher(
connection(), QuicConnection::SEND_ACK_IF_QUEUED);
control_frame_manager_.WriteOrBufferRstStream(id, error, bytes_written);
control_frame_manager_.WriteOrBufferStopSending(error, id);
} else {
control_frame_manager_.WriteOrBufferRstStream(id, error, bytes_written);
}
}
connection_->OnStreamReset(id, error);
}
if (error != QUIC_STREAM_NO_ERROR && QuicContainsKey(zombie_streams_, id)) {
OnStreamDoneWaitingForAcks(id);
return;
}
if (!close_write_side_only) {
CloseStreamInner(id, true);
return;
}
DCHECK_EQ(QUIC_VERSION_99, connection_->transport_version());
DynamicStreamMap::iterator it = dynamic_stream_map_.find(id);
if (it != dynamic_stream_map_.end()) {
if (GetQuicReloadableFlag(quic_eliminate_static_stream_map) &&
it->second->is_static()) {
QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map, 5, 9);
QUIC_DVLOG(1) << ENDPOINT
<< "Try to send rst for a static stream, id: " << id
<< " Closing connection";
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Sending rst for a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
QuicStream* stream = it->second.get();
if (stream) {
stream->set_rst_sent(true);
stream->CloseWriteSide();
}
}
}
void QuicSession::SendGoAway(QuicErrorCode error_code,
const std::string& reason) {
// GOAWAY frame is not supported in v99.
DCHECK_NE(QUIC_VERSION_99, connection_->transport_version());
if (goaway_sent_) {
return;
}
goaway_sent_ = true;
control_frame_manager_.WriteOrBufferGoAway(
error_code, stream_id_manager_.largest_peer_created_stream_id(), reason);
}
void QuicSession::SendBlocked(QuicStreamId id) {
control_frame_manager_.WriteOrBufferBlocked(id);
}
void QuicSession::SendWindowUpdate(QuicStreamId id,
QuicStreamOffset byte_offset) {
control_frame_manager_.WriteOrBufferWindowUpdate(id, byte_offset);
}
void QuicSession::SendMaxStreams(QuicStreamCount stream_count,
bool unidirectional) {
control_frame_manager_.WriteOrBufferMaxStreams(stream_count, unidirectional);
}
void QuicSession::SendStreamsBlocked(QuicStreamCount stream_count,
bool unidirectional) {
control_frame_manager_.WriteOrBufferStreamsBlocked(stream_count,
unidirectional);
}
void QuicSession::CloseStream(QuicStreamId stream_id) {
CloseStreamInner(stream_id, false);
}
void QuicSession::InsertLocallyClosedStreamsHighestOffset(
const QuicStreamId id,
QuicStreamOffset offset) {
locally_closed_streams_highest_offset_[id] = offset;
if (IsIncomingStream(id)) {
++num_locally_closed_incoming_streams_highest_offset_;
}
}
void QuicSession::CloseStreamInner(QuicStreamId stream_id, bool locally_reset) {
QUIC_DVLOG(1) << ENDPOINT << "Closing stream " << stream_id;
DynamicStreamMap::iterator it = dynamic_stream_map_.find(stream_id);
if (it == dynamic_stream_map_.end()) {
// When CloseStreamInner has been called recursively (via
// QuicStream::OnClose), the stream will already have been deleted
// from stream_map_, so return immediately.
QUIC_DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id;
return;
}
QuicStream* stream = it->second.get();
if (GetQuicReloadableFlag(quic_eliminate_static_stream_map) &&
stream->is_static()) {
QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map, 6, 9);
QUIC_DVLOG(1) << ENDPOINT
<< "Try to close a static stream, id: " << stream_id
<< " Closing connection";
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Try to close a static stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
// Tell the stream that a RST has been sent.
if (locally_reset) {
stream->set_rst_sent(true);
}
if (stream->IsWaitingForAcks()) {
zombie_streams_[stream->id()] = std::move(it->second);
} else {
closed_streams_.push_back(std::move(it->second));
// Do not retransmit data of a closed stream.
streams_with_pending_retransmission_.erase(stream_id);
if (!closed_streams_clean_up_alarm_->IsSet()) {
closed_streams_clean_up_alarm_->Set(
connection_->clock()->ApproximateNow());
}
}
// If we haven't received a FIN or RST for this stream, we need to keep track
// of the how many bytes the stream's flow controller believes it has
// received, for accurate connection level flow control accounting.
const bool had_fin_or_rst = stream->HasFinalReceivedByteOffset();
if (!had_fin_or_rst) {
InsertLocallyClosedStreamsHighestOffset(
stream_id, stream->flow_controller()->highest_received_byte_offset());
}
dynamic_stream_map_.erase(it);
if (IsIncomingStream(stream_id)) {
--num_dynamic_incoming_streams_;
}
const bool stream_was_draining =
draining_streams_.find(stream_id) != draining_streams_.end();
if (stream_was_draining) {
if (IsIncomingStream(stream_id)) {
--num_draining_incoming_streams_;
}
draining_streams_.erase(stream_id);
} else if (connection_->transport_version() == QUIC_VERSION_99) {
// Stream was not draining, but we did have a fin or rst, so we can now
// free the stream ID if version 99.
if (had_fin_or_rst) {
v99_streamid_manager_.OnStreamClosed(stream_id);
}
}
stream->OnClose();
if (!stream_was_draining && !IsIncomingStream(stream_id) && had_fin_or_rst &&
connection_->transport_version() != QUIC_VERSION_99) {
// Streams that first became draining already called OnCanCreate...
// This covers the case where the stream went directly to being closed.
OnCanCreateNewOutgoingStream();
}
}
void QuicSession::ClosePendingStream(QuicStreamId stream_id) {
QUIC_DVLOG(1) << ENDPOINT << "Closing stream " << stream_id;
if (pending_stream_map_.find(stream_id) == pending_stream_map_.end()) {
QUIC_BUG << ENDPOINT << "Stream is already closed: " << stream_id;
return;
}
SendRstStream(stream_id, QUIC_RST_ACKNOWLEDGEMENT, 0);
// The pending stream may have been deleted and removed during SendRstStream.
// Remove the stream from pending stream map iff it is still in the map.
if (pending_stream_map_.find(stream_id) != pending_stream_map_.end()) {
pending_stream_map_.erase(stream_id);
}
--num_dynamic_incoming_streams_;
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.OnStreamClosed(stream_id);
}
OnCanCreateNewOutgoingStream();
}
void QuicSession::OnFinalByteOffsetReceived(
QuicStreamId stream_id,
QuicStreamOffset final_byte_offset) {
auto it = locally_closed_streams_highest_offset_.find(stream_id);
if (it == locally_closed_streams_highest_offset_.end()) {
return;
}
QUIC_DVLOG(1) << ENDPOINT << "Received final byte offset "
<< final_byte_offset << " for stream " << stream_id;
QuicByteCount offset_diff = final_byte_offset - it->second;
if (flow_controller_.UpdateHighestReceivedOffset(
flow_controller_.highest_received_byte_offset() + offset_diff)) {
// If the final offset violates flow control, close the connection now.
if (flow_controller_.FlowControlViolation()) {
connection_->CloseConnection(
QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA,
"Connection level flow control violation",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
}
flow_controller_.AddBytesConsumed(offset_diff);
locally_closed_streams_highest_offset_.erase(it);
if (IsIncomingStream(stream_id)) {
--num_locally_closed_incoming_streams_highest_offset_;
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.OnStreamClosed(stream_id);
}
} else if (connection_->transport_version() != QUIC_VERSION_99) {
OnCanCreateNewOutgoingStream();
}
}
bool QuicSession::IsEncryptionEstablished() const {
// Once the handshake is confirmed, it never becomes un-confirmed.
if (is_handshake_confirmed_) {
return true;
}
return GetCryptoStream()->encryption_established();
}
bool QuicSession::IsCryptoHandshakeConfirmed() const {
return GetCryptoStream()->handshake_confirmed();
}
void QuicSession::OnConfigNegotiated() {
connection_->SetFromConfig(config_);
uint32_t max_streams = 0;
if (config_.HasReceivedMaxIncomingDynamicStreams()) {
max_streams = config_.ReceivedMaxIncomingDynamicStreams();
}
QUIC_DVLOG(1) << "Setting max_open_outgoing_streams_ to " << max_streams;
if (connection_->transport_version() == QUIC_VERSION_99) {
// TODO: When transport negotiation knows about bi- and uni- directional
// streams, this should be modified to indicate which one to the manager.
// Currently, BOTH are set to the same value.
// TODO(fkastenholz): AdjustMax is cognizant of the number of static streams
// and sets the maximum to be max_streams + number_of_statics. This should
// eventually be removed from IETF QUIC. -- Replace the call with
// ConfigureMaxOpen...
v99_streamid_manager_.AdjustMaxOpenOutgoingStreams(max_streams);
} else {
stream_id_manager_.set_max_open_outgoing_streams(max_streams);
}
if (perspective() == Perspective::IS_SERVER) {
if (config_.HasReceivedConnectionOptions()) {
// The following variations change the initial receive flow control
// window sizes.
if (ContainsQuicTag(config_.ReceivedConnectionOptions(), kIFW6)) {
AdjustInitialFlowControlWindows(64 * 1024);
}
if (ContainsQuicTag(config_.ReceivedConnectionOptions(), kIFW7)) {
AdjustInitialFlowControlWindows(128 * 1024);
}
if (ContainsQuicTag(config_.ReceivedConnectionOptions(), kIFW8)) {
AdjustInitialFlowControlWindows(256 * 1024);
}
if (ContainsQuicTag(config_.ReceivedConnectionOptions(), kIFW9)) {
AdjustInitialFlowControlWindows(512 * 1024);
}
if (ContainsQuicTag(config_.ReceivedConnectionOptions(), kIFWA)) {
AdjustInitialFlowControlWindows(1024 * 1024);
}
}
config_.SetStatelessResetTokenToSend(GetStatelessResetToken());
}
// A small number of additional incoming streams beyond the limit should be
// allowed. This helps avoid early connection termination when FIN/RSTs for
// old streams are lost or arrive out of order.
// Use a minimum number of additional streams, or a percentage increase,
// whichever is larger.
uint32_t max_incoming_streams_to_send =
config_.GetMaxIncomingDynamicStreamsToSend();
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.SetMaxOpenIncomingStreams(
max_incoming_streams_to_send);
} else {
uint32_t max_incoming_streams =
std::max(max_incoming_streams_to_send + kMaxStreamsMinimumIncrement,
static_cast<uint32_t>(max_incoming_streams_to_send *
kMaxStreamsMultiplier));
stream_id_manager_.set_max_open_incoming_streams(max_incoming_streams);
}
if (config_.HasReceivedInitialStreamFlowControlWindowBytes()) {
// Streams which were created before the SHLO was received (0-RTT
// requests) are now informed of the peer's initial flow control window.
OnNewStreamFlowControlWindow(
config_.ReceivedInitialStreamFlowControlWindowBytes());
}
if (config_.HasReceivedInitialSessionFlowControlWindowBytes()) {
OnNewSessionFlowControlWindow(
config_.ReceivedInitialSessionFlowControlWindowBytes());
}
}
void QuicSession::AdjustInitialFlowControlWindows(size_t stream_window) {
const float session_window_multiplier =
config_.GetInitialStreamFlowControlWindowToSend()
? static_cast<float>(
config_.GetInitialSessionFlowControlWindowToSend()) /
config_.GetInitialStreamFlowControlWindowToSend()
: 1.5;
QUIC_DVLOG(1) << ENDPOINT << "Set stream receive window to " << stream_window;
config_.SetInitialStreamFlowControlWindowToSend(stream_window);
size_t session_window = session_window_multiplier * stream_window;
QUIC_DVLOG(1) << ENDPOINT << "Set session receive window to "
<< session_window;
config_.SetInitialSessionFlowControlWindowToSend(session_window);
flow_controller_.UpdateReceiveWindowSize(session_window);
// Inform all existing streams about the new window.
for (auto const& kv : static_stream_map_) {
kv.second->flow_controller()->UpdateReceiveWindowSize(stream_window);
}
for (auto const& kv : dynamic_stream_map_) {
kv.second->flow_controller()->UpdateReceiveWindowSize(stream_window);
}
}
void QuicSession::HandleFrameOnNonexistentOutgoingStream(
QuicStreamId stream_id) {
DCHECK(!IsClosedStream(stream_id));
// Received a frame for a locally-created stream that is not currently
// active. This is an error.
connection()->CloseConnection(
QUIC_INVALID_STREAM_ID, "Data for nonexistent stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
}
void QuicSession::HandleRstOnValidNonexistentStream(
const QuicRstStreamFrame& frame) {
// If the stream is neither originally in active streams nor created in
// GetOrCreateDynamicStream(), it could be a closed stream in which case its
// final received byte offset need to be updated.
if (IsClosedStream(frame.stream_id)) {
// The RST frame contains the final byte offset for the stream: we can now
// update the connection level flow controller if needed.
OnFinalByteOffsetReceived(frame.stream_id, frame.byte_offset);
}
}
void QuicSession::OnNewStreamFlowControlWindow(QuicStreamOffset new_window) {
if (new_window < kMinimumFlowControlSendWindow) {
QUIC_LOG_FIRST_N(ERROR, 1)
<< "Peer sent us an invalid stream flow control send window: "
<< new_window << ", below default: " << kMinimumFlowControlSendWindow;
if (connection_->connected()) {
connection_->CloseConnection(
QUIC_FLOW_CONTROL_INVALID_WINDOW, "New stream window too low",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
}
return;
}
// Inform all existing streams about the new window.
for (auto const& kv : static_stream_map_) {
kv.second->UpdateSendWindowOffset(new_window);
}
for (auto const& kv : dynamic_stream_map_) {
kv.second->UpdateSendWindowOffset(new_window);
}
}
void QuicSession::OnNewSessionFlowControlWindow(QuicStreamOffset new_window) {
if (new_window < kMinimumFlowControlSendWindow) {
QUIC_LOG_FIRST_N(ERROR, 1)
<< "Peer sent us an invalid session flow control send window: "
<< new_window << ", below default: " << kMinimumFlowControlSendWindow;
if (connection_->connected()) {
connection_->CloseConnection(
QUIC_FLOW_CONTROL_INVALID_WINDOW, "New connection window too low",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
}
return;
}
flow_controller_.UpdateSendWindowOffset(new_window);
}
void QuicSession::OnCryptoHandshakeEvent(CryptoHandshakeEvent event) {
switch (event) {
// TODO(satyamshekhar): Move the logic of setting the encrypter/decrypter
// to QuicSession since it is the glue.
case ENCRYPTION_FIRST_ESTABLISHED:
// Given any streams blocked by encryption a chance to write.
OnCanWrite();
break;
case ENCRYPTION_REESTABLISHED:
// Retransmit originally packets that were sent, since they can't be
// decrypted by the peer.
connection_->RetransmitUnackedPackets(ALL_INITIAL_RETRANSMISSION);
// Given any streams blocked by encryption a chance to write.
OnCanWrite();
break;
case HANDSHAKE_CONFIRMED:
QUIC_BUG_IF(!config_.negotiated())
<< ENDPOINT << "Handshake confirmed without parameter negotiation.";
// Discard originally encrypted packets, since they can't be decrypted by
// the peer.
NeuterUnencryptedData();
is_handshake_confirmed_ = true;
break;
default:
QUIC_LOG(ERROR) << ENDPOINT << "Got unknown handshake event: " << event;
}
}
void QuicSession::OnCryptoHandshakeMessageSent(
const CryptoHandshakeMessage& /*message*/) {}
void QuicSession::OnCryptoHandshakeMessageReceived(
const CryptoHandshakeMessage& /*message*/) {}
void QuicSession::RegisterStreamPriority(QuicStreamId id,
bool is_static,
SpdyPriority priority) {
write_blocked_streams()->RegisterStream(id, is_static, priority);
}
void QuicSession::UnregisterStreamPriority(QuicStreamId id, bool is_static) {
write_blocked_streams()->UnregisterStream(id, is_static);
}
void QuicSession::UpdateStreamPriority(QuicStreamId id,
SpdyPriority new_priority) {
write_blocked_streams()->UpdateStreamPriority(id, new_priority);
}
QuicConfig* QuicSession::config() {
return &config_;
}
void QuicSession::ActivateStream(std::unique_ptr<QuicStream> stream) {
DCHECK(!stream->is_static());
QuicStreamId stream_id = stream->id();
QUIC_DVLOG(1) << ENDPOINT << "num_streams: " << dynamic_stream_map_.size()
<< ". activating " << stream_id;
DCHECK(!QuicContainsKey(dynamic_stream_map_, stream_id));
DCHECK(!QuicContainsKey(static_stream_map_, stream_id));
dynamic_stream_map_[stream_id] = std::move(stream);
if (IsIncomingStream(stream_id)) {
++num_dynamic_incoming_streams_;
}
}
QuicStreamId QuicSession::GetNextOutgoingBidirectionalStreamId() {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.GetNextOutgoingBidirectionalStreamId();
}
return stream_id_manager_.GetNextOutgoingStreamId();
}
QuicStreamId QuicSession::GetNextOutgoingUnidirectionalStreamId() {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.GetNextOutgoingUnidirectionalStreamId();
}
return stream_id_manager_.GetNextOutgoingStreamId();
}
bool QuicSession::CanOpenNextOutgoingBidirectionalStream() {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.CanOpenNextOutgoingBidirectionalStream();
}
return stream_id_manager_.CanOpenNextOutgoingStream(
GetNumOpenOutgoingStreams());
}
bool QuicSession::CanOpenNextOutgoingUnidirectionalStream() {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.CanOpenNextOutgoingUnidirectionalStream();
}
return stream_id_manager_.CanOpenNextOutgoingStream(
GetNumOpenOutgoingStreams());
}
QuicStream* QuicSession::GetOrCreateStream(const QuicStreamId stream_id) {
StreamHandler handler =
GetOrCreateStreamImpl(stream_id, /*may_buffer=*/false);
DCHECK(!handler.is_pending);
return handler.stream;
}
QuicSession::StreamHandler QuicSession::GetOrCreateStreamImpl(
QuicStreamId stream_id,
bool may_buffer) {
StaticStreamMap::iterator it = static_stream_map_.find(stream_id);
if (it != static_stream_map_.end()) {
return StreamHandler(it->second);
}
return GetOrCreateDynamicStreamImpl(stream_id, may_buffer);
}
void QuicSession::StreamDraining(QuicStreamId stream_id) {
DCHECK(QuicContainsKey(dynamic_stream_map_, stream_id));
if (!QuicContainsKey(draining_streams_, stream_id)) {
draining_streams_.insert(stream_id);
if (IsIncomingStream(stream_id)) {
++num_draining_incoming_streams_;
}
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.OnStreamClosed(stream_id);
}
}
if (!IsIncomingStream(stream_id)) {
// Inform application that a stream is available.
OnCanCreateNewOutgoingStream();
}
}
bool QuicSession::MaybeIncreaseLargestPeerStreamId(
const QuicStreamId stream_id) {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.MaybeIncreaseLargestPeerStreamId(stream_id);
}
return stream_id_manager_.MaybeIncreaseLargestPeerStreamId(stream_id);
}
bool QuicSession::ShouldYield(QuicStreamId stream_id) {
if (stream_id == currently_writing_stream_id_) {
return false;
}
return write_blocked_streams()->ShouldYield(stream_id);
}
QuicStream* QuicSession::GetOrCreateDynamicStream(
const QuicStreamId stream_id) {
StreamHandler handler =
GetOrCreateDynamicStreamImpl(stream_id, /*may_buffer=*/false);
DCHECK(!handler.is_pending);
return handler.stream;
}
QuicSession::StreamHandler QuicSession::GetOrCreateDynamicStreamImpl(
QuicStreamId stream_id,
bool may_buffer) {
DCHECK(!QuicContainsKey(static_stream_map_, stream_id))
<< "Attempt to call GetOrCreateDynamicStream for a static stream";
DynamicStreamMap::iterator it = dynamic_stream_map_.find(stream_id);
if (it != dynamic_stream_map_.end()) {
return StreamHandler(it->second.get());
}
if (IsClosedStream(stream_id)) {
return StreamHandler();
}
if (!IsIncomingStream(stream_id)) {
HandleFrameOnNonexistentOutgoingStream(stream_id);
return StreamHandler();
}
auto pending_it = pending_stream_map_.find(stream_id);
if (pending_it != pending_stream_map_.end()) {
DCHECK_EQ(QUIC_VERSION_99, connection_->transport_version());
if (may_buffer) {
return StreamHandler(pending_it->second.get());
}
// The stream limit accounting has already been taken care of
// when the PendingStream was created, so there is no need to
// do so here. Now we can create the actual stream from the
// PendingStream.
StreamHandler handler(CreateIncomingStream(std::move(*pending_it->second)));
pending_stream_map_.erase(pending_it);
return handler;
}
// TODO(fkastenholz): If we are creating a new stream and we have
// sent a goaway, we should ignore the stream creation. Need to
// add code to A) test if goaway was sent ("if (goaway_sent_)") and
// B) reject stream creation ("return nullptr")
if (!MaybeIncreaseLargestPeerStreamId(stream_id)) {
return StreamHandler();
}
if (connection_->transport_version() != QUIC_VERSION_99) {
// TODO(fayang): Let LegacyQuicStreamIdManager count open streams and make
// CanOpenIncomingStream interface cosistent with that of v99.
if (!stream_id_manager_.CanOpenIncomingStream(
GetNumOpenIncomingStreams())) {
// Refuse to open the stream.
SendRstStream(stream_id, QUIC_REFUSED_STREAM, 0);
return StreamHandler();
}
}
if (connection_->transport_version() == QUIC_VERSION_99 && may_buffer &&
ShouldBufferIncomingStream(stream_id)) {
++num_dynamic_incoming_streams_;
// Since STREAM frames may arrive out of order, delay creating the
// stream object until the first byte arrives. Buffer the frames and
// handle flow control accounting in the PendingStream.
auto pending = QuicMakeUnique<PendingStream>(stream_id, this);
StreamHandler handler(pending.get());
pending_stream_map_[stream_id] = std::move(pending);
return handler;
}
return StreamHandler(CreateIncomingStream(stream_id));
}
void QuicSession::set_largest_peer_created_stream_id(
QuicStreamId largest_peer_created_stream_id) {
if (connection_->transport_version() == QUIC_VERSION_99) {
v99_streamid_manager_.SetLargestPeerCreatedStreamId(
largest_peer_created_stream_id);
return;
}
stream_id_manager_.set_largest_peer_created_stream_id(
largest_peer_created_stream_id);
}
bool QuicSession::IsClosedStream(QuicStreamId id) {
DCHECK_NE(QuicUtils::GetInvalidStreamId(connection_->transport_version()),
id);
if (IsOpenStream(id)) {
// Stream is active
return false;
}
if (connection_->transport_version() == QUIC_VERSION_99) {
return !v99_streamid_manager_.IsAvailableStream(id);
}
return !stream_id_manager_.IsAvailableStream(id);
}
bool QuicSession::IsOpenStream(QuicStreamId id) {
DCHECK_NE(QuicUtils::GetInvalidStreamId(connection_->transport_version()),
id);
if (QuicContainsKey(static_stream_map_, id) ||
QuicContainsKey(dynamic_stream_map_, id) ||
QuicContainsKey(pending_stream_map_, id)) {
// Stream is active
return true;
}
return false;
}
size_t QuicSession::GetNumOpenIncomingStreams() const {
return num_dynamic_incoming_streams_ - num_draining_incoming_streams_ +
num_locally_closed_incoming_streams_highest_offset_;
}
size_t QuicSession::GetNumOpenOutgoingStreams() const {
DCHECK_GE(GetNumDynamicOutgoingStreams() +
GetNumLocallyClosedOutgoingStreamsHighestOffset(),
GetNumDrainingOutgoingStreams());
return GetNumDynamicOutgoingStreams() +
GetNumLocallyClosedOutgoingStreamsHighestOffset() -
GetNumDrainingOutgoingStreams();
}
size_t QuicSession::GetNumActiveStreams() const {
return dynamic_stream_map_.size() - draining_streams_.size() -
num_incoming_static_streams_ - num_outgoing_static_streams_;
}
size_t QuicSession::GetNumDrainingStreams() const {
return draining_streams_.size();
}
void QuicSession::MarkConnectionLevelWriteBlocked(QuicStreamId id) {
if (GetOrCreateStream(id) == nullptr) {
QUIC_BUG << "Marking unknown stream " << id << " blocked.";
QUIC_LOG_FIRST_N(ERROR, 2) << QuicStackTrace();
}
write_blocked_streams_.AddStream(id);
}
bool QuicSession::HasDataToWrite() const {
return write_blocked_streams_.HasWriteBlockedSpecialStream() ||
write_blocked_streams_.HasWriteBlockedDataStreams() ||
connection_->HasQueuedData() ||
!streams_with_pending_retransmission_.empty() ||
control_frame_manager_.WillingToWrite();
}
void QuicSession::OnAckNeedsRetransmittableFrame() {
flow_controller_.SendWindowUpdate();
}
void QuicSession::SendPing() {
control_frame_manager_.WritePing();
}
size_t QuicSession::GetNumDynamicOutgoingStreams() const {
DCHECK_GE(static_cast<size_t>(dynamic_stream_map_.size() +
pending_stream_map_.size()),
num_dynamic_incoming_streams_ + num_outgoing_static_streams_ +
num_incoming_static_streams_);
return dynamic_stream_map_.size() + pending_stream_map_.size() -
num_dynamic_incoming_streams_ - num_outgoing_static_streams_ -
num_incoming_static_streams_;
}
size_t QuicSession::GetNumDrainingOutgoingStreams() const {
DCHECK_GE(draining_streams_.size(), num_draining_incoming_streams_);
return draining_streams_.size() - num_draining_incoming_streams_;
}
size_t QuicSession::GetNumLocallyClosedOutgoingStreamsHighestOffset() const {
DCHECK_GE(locally_closed_streams_highest_offset_.size(),
num_locally_closed_incoming_streams_highest_offset_);
return locally_closed_streams_highest_offset_.size() -
num_locally_closed_incoming_streams_highest_offset_;
}
bool QuicSession::IsConnectionFlowControlBlocked() const {
return flow_controller_.IsBlocked();
}
bool QuicSession::IsStreamFlowControlBlocked() {
for (auto const& kv : static_stream_map_) {
if (kv.second->flow_controller()->IsBlocked()) {
return true;
}
}
for (auto const& kv : dynamic_stream_map_) {
if (kv.second->flow_controller()->IsBlocked()) {
return true;
}
}
return false;
}
size_t QuicSession::MaxAvailableBidirectionalStreams() const {
if (connection()->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.GetMaxAllowdIncomingBidirectionalStreams();
}
return stream_id_manager_.MaxAvailableStreams();
}
size_t QuicSession::MaxAvailableUnidirectionalStreams() const {
if (connection()->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.GetMaxAllowdIncomingUnidirectionalStreams();
}
return stream_id_manager_.MaxAvailableStreams();
}
bool QuicSession::IsIncomingStream(QuicStreamId id) const {
if (connection()->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.IsIncomingStream(id);
}
return stream_id_manager_.IsIncomingStream(id);
}
void QuicSession::OnStreamDoneWaitingForAcks(QuicStreamId id) {
auto it = zombie_streams_.find(id);
if (it == zombie_streams_.end()) {
return;
}
closed_streams_.push_back(std::move(it->second));
if (!closed_streams_clean_up_alarm_->IsSet()) {
closed_streams_clean_up_alarm_->Set(connection_->clock()->ApproximateNow());
}
zombie_streams_.erase(it);
// Do not retransmit data of a closed stream.
streams_with_pending_retransmission_.erase(id);
}
QuicStream* QuicSession::GetStream(QuicStreamId id) const {
if (id <= largest_static_stream_id_) {
auto static_stream = static_stream_map_.find(id);
if (static_stream != static_stream_map_.end()) {
return static_stream->second;
}
}
auto active_stream = dynamic_stream_map_.find(id);
if (active_stream != dynamic_stream_map_.end()) {
return active_stream->second.get();
}
auto zombie_stream = zombie_streams_.find(id);
if (zombie_stream != zombie_streams_.end()) {
return zombie_stream->second.get();
}
return nullptr;
}
bool QuicSession::OnFrameAcked(const QuicFrame& frame,
QuicTime::Delta ack_delay_time) {
if (frame.type == MESSAGE_FRAME) {
OnMessageAcked(frame.message_frame->message_id);
return true;
}
if (frame.type == CRYPTO_FRAME) {
return GetMutableCryptoStream()->OnCryptoFrameAcked(*frame.crypto_frame,
ack_delay_time);
}
if (frame.type != STREAM_FRAME) {
return control_frame_manager_.OnControlFrameAcked(frame);
}
bool new_stream_data_acked = false;
QuicStream* stream = GetStream(frame.stream_frame.stream_id);
// Stream can already be reset when sent frame gets acked.
if (stream != nullptr) {
QuicByteCount newly_acked_length = 0;
new_stream_data_acked = stream->OnStreamFrameAcked(
frame.stream_frame.offset, frame.stream_frame.data_length,
frame.stream_frame.fin, ack_delay_time, &newly_acked_length);
if (!stream->HasPendingRetransmission()) {
streams_with_pending_retransmission_.erase(stream->id());
}
}
return new_stream_data_acked;
}
void QuicSession::OnStreamFrameRetransmitted(const QuicStreamFrame& frame) {
QuicStream* stream = GetStream(frame.stream_id);
if (stream == nullptr) {
QUIC_BUG << "Stream: " << frame.stream_id << " is closed when " << frame
<< " is retransmitted.";
connection()->CloseConnection(
QUIC_INTERNAL_ERROR, "Attempt to retransmit frame of a closed stream",
ConnectionCloseBehavior::SEND_CONNECTION_CLOSE_PACKET);
return;
}
stream->OnStreamFrameRetransmitted(frame.offset, frame.data_length,
frame.fin);
}
void QuicSession::OnFrameLost(const QuicFrame& frame) {
if (frame.type == MESSAGE_FRAME) {
OnMessageLost(frame.message_frame->message_id);
return;
}
if (frame.type == CRYPTO_FRAME) {
GetMutableCryptoStream()->OnCryptoFrameLost(frame.crypto_frame);
return;
}
if (frame.type != STREAM_FRAME) {
control_frame_manager_.OnControlFrameLost(frame);
return;
}
QuicStream* stream = GetStream(frame.stream_frame.stream_id);
if (stream == nullptr) {
return;
}
stream->OnStreamFrameLost(frame.stream_frame.offset,
frame.stream_frame.data_length,
frame.stream_frame.fin);
if (stream->HasPendingRetransmission() &&
!QuicContainsKey(streams_with_pending_retransmission_,
frame.stream_frame.stream_id)) {
streams_with_pending_retransmission_.insert(
std::make_pair(frame.stream_frame.stream_id, true));
}
}
void QuicSession::RetransmitFrames(const QuicFrames& frames,
TransmissionType type) {
QuicConnection::ScopedPacketFlusher retransmission_flusher(
connection_, QuicConnection::NO_ACK);
SetTransmissionType(type);
for (const QuicFrame& frame : frames) {
if (frame.type == MESSAGE_FRAME) {
// Do not retransmit MESSAGE frames.
continue;
}
if (frame.type == CRYPTO_FRAME) {
GetMutableCryptoStream()->RetransmitData(frame.crypto_frame);
continue;
}
if (frame.type != STREAM_FRAME) {
if (!control_frame_manager_.RetransmitControlFrame(frame)) {
break;
}
continue;
}
QuicStream* stream = GetStream(frame.stream_frame.stream_id);
if (stream != nullptr &&
!stream->RetransmitStreamData(frame.stream_frame.offset,
frame.stream_frame.data_length,
frame.stream_frame.fin)) {
break;
}
}
}
bool QuicSession::IsFrameOutstanding(const QuicFrame& frame) const {
if (frame.type == MESSAGE_FRAME) {
return false;
}
if (frame.type == CRYPTO_FRAME) {
return GetCryptoStream()->IsFrameOutstanding(
frame.crypto_frame->level, frame.crypto_frame->offset,
frame.crypto_frame->data_length);
}
if (frame.type != STREAM_FRAME) {
return control_frame_manager_.IsControlFrameOutstanding(frame);
}
QuicStream* stream = GetStream(frame.stream_frame.stream_id);
return stream != nullptr &&
stream->IsStreamFrameOutstanding(frame.stream_frame.offset,
frame.stream_frame.data_length,
frame.stream_frame.fin);
}
bool QuicSession::HasUnackedCryptoData() const {
const QuicCryptoStream* crypto_stream = GetCryptoStream();
if (crypto_stream->IsWaitingForAcks()) {
return true;
}
if (GetQuicReloadableFlag(quic_fix_has_pending_crypto_data) &&
crypto_stream->HasBufferedData()) {
QUIC_RELOADABLE_FLAG_COUNT(quic_fix_has_pending_crypto_data);
return true;
}
return false;
}
WriteStreamDataResult QuicSession::WriteStreamData(QuicStreamId id,
QuicStreamOffset offset,
QuicByteCount data_length,
QuicDataWriter* writer) {
QuicStream* stream = GetStream(id);
if (stream == nullptr) {
// This causes the connection to be closed because of failed to serialize
// packet.
QUIC_BUG << "Stream " << id << " does not exist when trying to write data."
<< " version:" << connection_->transport_version();
return STREAM_MISSING;
}
if (stream->WriteStreamData(offset, data_length, writer)) {
return WRITE_SUCCESS;
}
return WRITE_FAILED;
}
bool QuicSession::WriteCryptoData(EncryptionLevel level,
QuicStreamOffset offset,
QuicByteCount data_length,
QuicDataWriter* writer) {
return GetMutableCryptoStream()->WriteCryptoFrame(level, offset, data_length,
writer);
}
QuicUint128 QuicSession::GetStatelessResetToken() const {
return QuicUtils::GenerateStatelessResetToken(connection_->connection_id());
}
bool QuicSession::RetransmitLostData() {
QuicConnection::ScopedPacketFlusher retransmission_flusher(
connection_, QuicConnection::SEND_ACK_IF_QUEUED);
// Retransmit crypto data first.
bool uses_crypto_frames =
QuicVersionUsesCryptoFrames(connection_->transport_version());
QuicCryptoStream* crypto_stream = GetMutableCryptoStream();
if (uses_crypto_frames && crypto_stream->HasPendingCryptoRetransmission()) {
SetTransmissionType(HANDSHAKE_RETRANSMISSION);
crypto_stream->WritePendingCryptoRetransmission();
}
// Retransmit crypto data in stream 1 frames (version < 47).
if (!uses_crypto_frames &&
QuicContainsKey(
streams_with_pending_retransmission_,
QuicUtils::GetCryptoStreamId(connection_->transport_version()))) {
SetTransmissionType(HANDSHAKE_RETRANSMISSION);
// Retransmit crypto data first.
QuicStream* crypto_stream = GetStream(
QuicUtils::GetCryptoStreamId(connection_->transport_version()));
crypto_stream->OnCanWrite();
DCHECK(CheckStreamWriteBlocked(crypto_stream));
if (crypto_stream->HasPendingRetransmission()) {
// Connection is write blocked.
return false;
} else {
streams_with_pending_retransmission_.erase(
QuicUtils::GetCryptoStreamId(connection_->transport_version()));
}
}
if (control_frame_manager_.HasPendingRetransmission()) {
SetTransmissionType(LOSS_RETRANSMISSION);
control_frame_manager_.OnCanWrite();
if (control_frame_manager_.HasPendingRetransmission()) {
return false;
}
}
while (!streams_with_pending_retransmission_.empty()) {
if (!connection_->CanWriteStreamData()) {
break;
}
// Retransmit lost data on headers and data streams.
const QuicStreamId id = streams_with_pending_retransmission_.begin()->first;
QuicStream* stream = GetStream(id);
if (stream != nullptr) {
SetTransmissionType(LOSS_RETRANSMISSION);
stream->OnCanWrite();
DCHECK(CheckStreamWriteBlocked(stream));
if (stream->HasPendingRetransmission()) {
// Connection is write blocked.
break;
} else if (!streams_with_pending_retransmission_.empty() &&
streams_with_pending_retransmission_.begin()->first == id) {
// Retransmit lost data may cause connection close. If this stream
// has not yet sent fin, a RST_STREAM will be sent and it will be
// removed from streams_with_pending_retransmission_.
streams_with_pending_retransmission_.pop_front();
}
} else {
QUIC_BUG << "Try to retransmit data of a closed stream";
streams_with_pending_retransmission_.pop_front();
}
}
return streams_with_pending_retransmission_.empty();
}
void QuicSession::NeuterUnencryptedData() {
if (connection_->session_decides_what_to_write()) {
QuicCryptoStream* crypto_stream = GetMutableCryptoStream();
crypto_stream->NeuterUnencryptedStreamData();
if (!crypto_stream->HasPendingRetransmission()) {
streams_with_pending_retransmission_.erase(
QuicUtils::GetCryptoStreamId(connection_->transport_version()));
}
}
connection_->NeuterUnencryptedPackets();
}
void QuicSession::SetTransmissionType(TransmissionType type) {
connection_->SetTransmissionType(type);
}
MessageResult QuicSession::SendMessage(QuicMemSliceSpan message) {
if (!IsEncryptionEstablished()) {
return {MESSAGE_STATUS_ENCRYPTION_NOT_ESTABLISHED, 0};
}
MessageStatus result =
connection_->SendMessage(last_message_id_ + 1, message);
if (result == MESSAGE_STATUS_SUCCESS) {
return {result, ++last_message_id_};
}
return {result, 0};
}
void QuicSession::OnMessageAcked(QuicMessageId message_id) {
QUIC_DVLOG(1) << ENDPOINT << "message " << message_id << " gets acked.";
}
void QuicSession::OnMessageLost(QuicMessageId message_id) {
QUIC_DVLOG(1) << ENDPOINT << "message " << message_id
<< " is considered lost";
}
void QuicSession::CleanUpClosedStreams() {
closed_streams_.clear();
}
bool QuicSession::session_decides_what_to_write() const {
return connection_->session_decides_what_to_write();
}
QuicPacketLength QuicSession::GetCurrentLargestMessagePayload() const {
return connection_->GetCurrentLargestMessagePayload();
}
QuicPacketLength QuicSession::GetGuaranteedLargestMessagePayload() const {
return connection_->GetGuaranteedLargestMessagePayload();
}
void QuicSession::SendStopSending(uint16_t code, QuicStreamId stream_id) {
control_frame_manager_.WriteOrBufferStopSending(code, stream_id);
}
void QuicSession::OnCanCreateNewOutgoingStream() {}
QuicStreamId QuicSession::next_outgoing_bidirectional_stream_id() const {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.next_outgoing_bidirectional_stream_id();
}
return stream_id_manager_.next_outgoing_stream_id();
}
QuicStreamId QuicSession::next_outgoing_unidirectional_stream_id() const {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.next_outgoing_unidirectional_stream_id();
}
return stream_id_manager_.next_outgoing_stream_id();
}
bool QuicSession::OnMaxStreamsFrame(const QuicMaxStreamsFrame& frame) {
return v99_streamid_manager_.OnMaxStreamsFrame(frame);
}
bool QuicSession::OnStreamsBlockedFrame(const QuicStreamsBlockedFrame& frame) {
return v99_streamid_manager_.OnStreamsBlockedFrame(frame);
}
size_t QuicSession::max_open_incoming_bidirectional_streams() const {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.GetMaxAllowdIncomingBidirectionalStreams();
}
return stream_id_manager_.max_open_incoming_streams();
}
size_t QuicSession::max_open_incoming_unidirectional_streams() const {
if (connection_->transport_version() == QUIC_VERSION_99) {
return v99_streamid_manager_.GetMaxAllowdIncomingUnidirectionalStreams();
}
return stream_id_manager_.max_open_incoming_streams();
}
#undef ENDPOINT // undef for jumbo builds
} // namespace quic