| // 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_(connection->transport_version()), |
| config_(config), |
| stream_id_manager_(this, |
| kDefaultMaxStreamsPerConnection, |
| config_.GetMaxIncomingBidirectionalStreamsToSend()), |
| v99_streamid_manager_( |
| this, |
| kDefaultMaxStreamsPerConnection, |
| kDefaultMaxStreamsPerConnection, |
| config_.GetMaxIncomingBidirectionalStreamsToSend(), |
| config_.GetMaxIncomingUnidirectionalStreamsToSend()), |
| 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, |
| connection->version().AllowsLowFlowControlLimits() |
| ? 0 |
| : 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), |
| eliminate_static_stream_map_( |
| GetQuicReloadableFlag(quic_eliminate_static_stream_map_3) || |
| QuicVersionUsesCryptoFrames(connection->transport_version())) { |
| 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_); |
| |
| if (QuicVersionUsesCryptoFrames(connection_->transport_version())) { |
| return; |
| } |
| |
| DCHECK_EQ(QuicUtils::GetCryptoStreamId(connection_->transport_version()), |
| GetMutableCryptoStream()->id()); |
| if (!eliminate_static_stream_map_) { |
| RegisterStaticStream( |
| QuicUtils::GetCryptoStreamId(connection_->transport_version()), |
| GetMutableCryptoStream()); |
| } else { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 10, 17); |
| QuicStreamId id = |
| QuicUtils::GetCryptoStreamId(connection_->transport_version()); |
| largest_static_stream_id_ = std::max(id, largest_static_stream_id_); |
| if (connection_->transport_version() == QUIC_VERSION_99) { |
| v99_streamid_manager_.RegisterStaticStream(id); |
| } |
| } |
| } |
| |
| 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(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::PendingStreamOnStreamFrame(const QuicStreamFrame& frame) { |
| DCHECK(VersionHasControlStreams(connection()->transport_version())); |
| QuicStreamId stream_id = frame.stream_id; |
| |
| PendingStream* pending = GetOrCreatePendingStream(stream_id); |
| |
| if (!pending) { |
| if (frame.fin) { |
| QuicStreamOffset final_byte_offset = frame.offset + frame.data_length; |
| OnFinalByteOffsetReceived(stream_id, final_byte_offset); |
| } |
| return; |
| } |
| |
| pending->OnStreamFrame(frame); |
| ProcessPendingStream(pending); |
| } |
| |
| 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; |
| } |
| |
| if (VersionHasControlStreams(connection()->transport_version()) && |
| UsesPendingStreams() && |
| QuicUtils::GetStreamType(stream_id, perspective(), |
| IsIncomingStream(stream_id)) == |
| READ_UNIDIRECTIONAL && |
| dynamic_stream_map_.find(stream_id) == dynamic_stream_map_.end()) { |
| PendingStreamOnStreamFrame(frame); |
| return; |
| } |
| |
| StreamHandler handler = GetOrCreateStreamImpl(stream_id); |
| |
| 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 (eliminate_static_stream_map_ && frame.fin && |
| handler.stream->is_static()) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 1, 17); |
| 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) || |
| QuicUtils::IsCryptoStreamId(connection_->transport_version(), |
| 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 (eliminate_static_stream_map_ && stream->is_static()) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 2, 17); |
| 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::PendingStreamOnRstStream(const QuicRstStreamFrame& frame) { |
| DCHECK(VersionHasControlStreams(connection()->transport_version())); |
| QuicStreamId stream_id = frame.stream_id; |
| |
| PendingStream* pending = GetOrCreatePendingStream(stream_id); |
| |
| if (!pending) { |
| HandleRstOnValidNonexistentStream(frame); |
| return; |
| } |
| |
| pending->OnRstStreamFrame(frame); |
| ClosePendingStream(stream_id); |
| } |
| |
| 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); |
| } |
| |
| if (VersionHasControlStreams(connection()->transport_version()) && |
| UsesPendingStreams() && |
| QuicUtils::GetStreamType(stream_id, perspective(), |
| IsIncomingStream(stream_id)) == |
| READ_UNIDIRECTIONAL && |
| dynamic_stream_map_.find(stream_id) == dynamic_stream_map_.end()) { |
| PendingStreamOnRstStream(frame); |
| return; |
| } |
| |
| StreamHandler handler = GetOrCreateStreamImpl(stream_id); |
| |
| if (!handler.stream) { |
| HandleRstOnValidNonexistentStream(frame); |
| return; // Errors are handled by GetOrCreateStream. |
| } |
| if (eliminate_static_stream_map_ && handler.stream->is_static()) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 3, 17); |
| 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 (!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_3, 4, 17); |
| // 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::SendProbingData() { |
| if (connection()->sent_packet_manager().MaybeRetransmitOldestPacket( |
| PROBING_RETRANSMISSION)) { |
| return true; |
| } |
| return false; |
| } |
| |
| 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 { |
| if (QuicVersionUsesCryptoFrames(connection_->transport_version())) { |
| // Writing CRYPTO frames is not subject to flow control, so there can't be |
| // pending data to write, only pending retransmissions. |
| return GetCryptoStream()->HasPendingCryptoRetransmission(); |
| } |
| 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 (QuicUtils::IsCryptoStreamId(connection_->transport_version(), id) && |
| 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() && |
| !QuicUtils::IsCryptoStreamId(connection_->transport_version(), id)) { |
| // 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 (eliminate_static_stream_map_ && it->second->is_static()) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 5, 17); |
| 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 (eliminate_static_stream_map_ && stream->is_static()) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 6, 17); |
| 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_); |
| |
| if (connection_->transport_version() == QUIC_VERSION_99) { |
| uint32_t max_streams = 0; |
| if (config_.HasReceivedMaxIncomingBidirectionalStreams()) { |
| max_streams = config_.ReceivedMaxIncomingBidirectionalStreams(); |
| } |
| QUIC_DVLOG(1) << "Setting Bidirectional outgoing_max_streams_ to " |
| << max_streams; |
| v99_streamid_manager_.AdjustMaxOpenOutgoingBidirectionalStreams( |
| max_streams); |
| |
| max_streams = 0; |
| if (config_.HasReceivedMaxIncomingUnidirectionalStreams()) { |
| max_streams = config_.ReceivedMaxIncomingUnidirectionalStreams(); |
| } |
| QUIC_DVLOG(1) << "Setting Unidirectional outgoing_max_streams_ to " |
| << max_streams; |
| v99_streamid_manager_.AdjustMaxOpenOutgoingUnidirectionalStreams( |
| max_streams); |
| } else { |
| uint32_t max_streams = 0; |
| if (config_.HasReceivedMaxIncomingBidirectionalStreams()) { |
| max_streams = config_.ReceivedMaxIncomingBidirectionalStreams(); |
| } |
| QUIC_DVLOG(1) << "Setting max_open_outgoing_streams_ to " << max_streams; |
| 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()); |
| } |
| |
| if (connection_->transport_version() == QUIC_VERSION_99) { |
| v99_streamid_manager_.SetMaxOpenIncomingBidirectionalStreams( |
| config_.GetMaxIncomingBidirectionalStreamsToSend()); |
| v99_streamid_manager_.SetMaxOpenIncomingUnidirectionalStreams( |
| config_.GetMaxIncomingUnidirectionalStreamsToSend()); |
| } else { |
| // 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_.GetMaxIncomingBidirectionalStreamsToSend(); |
| 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); |
| } |
| if (eliminate_static_stream_map_ && |
| !QuicVersionUsesCryptoFrames(connection_->transport_version())) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 11, 17); |
| GetMutableCryptoStream()->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 && |
| !connection_->version().AllowsLowFlowControlLimits()) { |
| QUIC_LOG_FIRST_N(ERROR, 1) |
| << "Peer sent us an invalid stream flow control send window: " |
| << new_window << ", below minimum: " << 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); |
| } |
| if (eliminate_static_stream_map_ && |
| !QuicVersionUsesCryptoFrames(connection_->transport_version())) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 12, 17); |
| GetMutableCryptoStream()->UpdateSendWindowOffset(new_window); |
| } |
| } |
| |
| void QuicSession::OnNewSessionFlowControlWindow(QuicStreamOffset new_window) { |
| if (new_window < kMinimumFlowControlSendWindow && |
| !connection_->version().AllowsLowFlowControlLimits()) { |
| 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); |
| DCHECK(!handler.is_pending); |
| return handler.stream; |
| } |
| |
| QuicSession::StreamHandler QuicSession::GetOrCreateStreamImpl( |
| QuicStreamId stream_id) { |
| if (eliminate_static_stream_map_ && |
| QuicUtils::IsCryptoStreamId(connection_->transport_version(), |
| stream_id)) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 13, 17); |
| return StreamHandler(GetMutableCryptoStream()); |
| } |
| StaticStreamMap::iterator it = static_stream_map_.find(stream_id); |
| if (it != static_stream_map_.end()) { |
| return StreamHandler(it->second); |
| } |
| return GetOrCreateDynamicStreamImpl(stream_id); |
| } |
| |
| 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); |
| } |
| |
| PendingStream* QuicSession::GetOrCreatePendingStream(QuicStreamId stream_id) { |
| auto it = pending_stream_map_.find(stream_id); |
| if (it != pending_stream_map_.end()) { |
| return it->second.get(); |
| } |
| |
| if (IsClosedStream(stream_id) || |
| !MaybeIncreaseLargestPeerStreamId(stream_id)) { |
| return nullptr; |
| } |
| |
| auto pending = QuicMakeUnique<PendingStream>(stream_id, this); |
| PendingStream* unowned_pending = pending.get(); |
| pending_stream_map_[stream_id] = std::move(pending); |
| return unowned_pending; |
| } |
| |
| QuicStream* QuicSession::GetOrCreateDynamicStream( |
| const QuicStreamId stream_id) { |
| StreamHandler handler = GetOrCreateDynamicStreamImpl(stream_id); |
| DCHECK(!handler.is_pending); |
| return handler.stream; |
| } |
| |
| QuicSession::StreamHandler QuicSession::GetOrCreateDynamicStreamImpl( |
| QuicStreamId stream_id) { |
| 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(); |
| } |
| |
| // 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(); |
| } |
| } |
| |
| 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) || |
| QuicUtils::IsCryptoStreamId(connection_->transport_version(), id)) { |
| // Stream is active |
| return true; |
| } |
| return false; |
| } |
| |
| bool QuicSession::IsStaticStream(QuicStreamId id) const { |
| if (eliminate_static_stream_map()) { |
| auto it = dynamic_stream_map_.find(id); |
| if (it == dynamic_stream_map_.end()) { |
| return false; |
| } |
| return it->second->is_static(); |
| } |
| |
| return QuicContainsKey(static_streams(), id); |
| } |
| |
| 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; |
| } |
| } |
| if (eliminate_static_stream_map_ && |
| !QuicVersionUsesCryptoFrames(connection_->transport_version()) && |
| GetMutableCryptoStream()->flow_controller()->IsBlocked()) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 14, 17); |
| 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(); |
| } |
| |
| if (eliminate_static_stream_map_ && |
| QuicUtils::IsCryptoStreamId(connection_->transport_version(), id)) { |
| QUIC_RELOADABLE_FLAG_COUNT_N(quic_eliminate_static_stream_map_3, 15, 17); |
| return const_cast<QuicCryptoStream*>(GetCryptoStream()); |
| } |
| |
| 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() && |
| !QuicVersionUsesCryptoFrames(connection_->transport_version())) { |
| 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 |