// 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/quic/quic_session.h" #include "base/stl_util.h" #include "net/quic/crypto/proof_verifier.h" #include "net/quic/quic_connection.h" #include "net/quic/quic_flags.h" #include "net/quic/quic_flow_controller.h" #include "net/quic/quic_headers_stream.h" #include "net/ssl/ssl_info.h" using base::StringPiece; using base::hash_map; using base::hash_set; using std::make_pair; using std::vector; namespace net { #define ENDPOINT (is_server() ? "Server: " : " Client: ") // We want to make sure we delete any closed streams in a safe manner. // To avoid deleting a stream in mid-operation, we have a simple shim between // us and the stream, so we can delete any streams when we return from // processing. // // We could just override the base methods, but this makes it easier to make // sure we don't miss any. class VisitorShim : public QuicConnectionVisitorInterface { public: explicit VisitorShim(QuicSession* session) : session_(session) {} virtual void OnStreamFrames(const vector<QuicStreamFrame>& frames) OVERRIDE { session_->OnStreamFrames(frames); session_->PostProcessAfterData(); } virtual void OnRstStream(const QuicRstStreamFrame& frame) OVERRIDE { session_->OnRstStream(frame); session_->PostProcessAfterData(); } virtual void OnGoAway(const QuicGoAwayFrame& frame) OVERRIDE { session_->OnGoAway(frame); session_->PostProcessAfterData(); } virtual void OnWindowUpdateFrames(const vector<QuicWindowUpdateFrame>& frames) OVERRIDE { session_->OnWindowUpdateFrames(frames); session_->PostProcessAfterData(); } virtual void OnBlockedFrames(const vector<QuicBlockedFrame>& frames) OVERRIDE { session_->OnBlockedFrames(frames); session_->PostProcessAfterData(); } virtual void OnCanWrite() OVERRIDE { session_->OnCanWrite(); session_->PostProcessAfterData(); } virtual void OnSuccessfulVersionNegotiation( const QuicVersion& version) OVERRIDE { session_->OnSuccessfulVersionNegotiation(version); } virtual void OnConnectionClosed( QuicErrorCode error, bool from_peer) OVERRIDE { session_->OnConnectionClosed(error, from_peer); // The session will go away, so don't bother with cleanup. } virtual void OnWriteBlocked() OVERRIDE { session_->OnWriteBlocked(); } virtual bool WillingAndAbleToWrite() const OVERRIDE { return session_->WillingAndAbleToWrite(); } virtual bool HasPendingHandshake() const OVERRIDE { return session_->HasPendingHandshake(); } virtual bool HasOpenDataStreams() const OVERRIDE { return session_->HasOpenDataStreams(); } private: QuicSession* session_; }; QuicSession::QuicSession(QuicConnection* connection, const QuicConfig& config) : connection_(connection), visitor_shim_(new VisitorShim(this)), config_(config), max_open_streams_(config_.max_streams_per_connection()), next_stream_id_(is_server() ? 2 : 3), largest_peer_created_stream_id_(0), error_(QUIC_NO_ERROR), goaway_received_(false), goaway_sent_(false), has_pending_handshake_(false) { if (connection_->version() <= QUIC_VERSION_19) { flow_controller_.reset(new QuicFlowController( connection_.get(), 0, is_server(), kDefaultFlowControlSendWindow, config_.GetInitialFlowControlWindowToSend(), config_.GetInitialFlowControlWindowToSend())); } else { flow_controller_.reset(new QuicFlowController( connection_.get(), 0, is_server(), kDefaultFlowControlSendWindow, config_.GetInitialSessionFlowControlWindowToSend(), config_.GetInitialSessionFlowControlWindowToSend())); } connection_->set_visitor(visitor_shim_.get()); connection_->SetFromConfig(config_); if (connection_->connected()) { connection_->SetOverallConnectionTimeout( config_.max_time_before_crypto_handshake()); } headers_stream_.reset(new QuicHeadersStream(this)); if (!is_server()) { // For version above QUIC v12, the headers stream is stream 3, so the // next available local stream ID should be 5. DCHECK_EQ(kHeadersStreamId, next_stream_id_); next_stream_id_ += 2; } } QuicSession::~QuicSession() { STLDeleteElements(&closed_streams_); STLDeleteValues(&stream_map_); DLOG_IF(WARNING, locally_closed_streams_highest_offset_.size() > max_open_streams_) << "Surprisingly high number of locally closed streams still waiting for " "final byte offset: " << locally_closed_streams_highest_offset_.size(); } void QuicSession::OnStreamFrames(const vector<QuicStreamFrame>& frames) { for (size_t i = 0; i < frames.size(); ++i) { // TODO(rch) deal with the error case of stream id 0. const QuicStreamFrame& frame = frames[i]; QuicStreamId stream_id = frame.stream_id; ReliableQuicStream* stream = GetStream(stream_id); if (!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.TotalBufferSize(); UpdateFlowControlOnFinalReceivedByteOffset(stream_id, final_byte_offset); } continue; } stream->OnStreamFrame(frames[i]); } } void QuicSession::OnStreamHeaders(QuicStreamId stream_id, StringPiece headers_data) { QuicDataStream* stream = GetDataStream(stream_id); if (!stream) { // It's quite possible to receive headers after a stream has been reset. return; } stream->OnStreamHeaders(headers_data); } void QuicSession::OnStreamHeadersPriority(QuicStreamId stream_id, QuicPriority priority) { QuicDataStream* stream = GetDataStream(stream_id); if (!stream) { // It's quite possible to receive headers after a stream has been reset. return; } stream->OnStreamHeadersPriority(priority); } void QuicSession::OnStreamHeadersComplete(QuicStreamId stream_id, bool fin, size_t frame_len) { QuicDataStream* stream = GetDataStream(stream_id); if (!stream) { // It's quite possible to receive headers after a stream has been reset. return; } stream->OnStreamHeadersComplete(fin, frame_len); } void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) { if (frame.stream_id == kCryptoStreamId) { connection()->SendConnectionCloseWithDetails( QUIC_INVALID_STREAM_ID, "Attempt to reset the crypto stream"); return; } if (frame.stream_id == kHeadersStreamId) { connection()->SendConnectionCloseWithDetails( QUIC_INVALID_STREAM_ID, "Attempt to reset the headers stream"); return; } QuicDataStream* stream = GetDataStream(frame.stream_id); if (!stream) { // The RST frame contains the final byte offset for the stream: we can now // update the connection level flow controller if needed. UpdateFlowControlOnFinalReceivedByteOffset(frame.stream_id, frame.byte_offset); return; // Errors are handled by GetStream. } stream->OnStreamReset(frame); } void QuicSession::OnGoAway(const QuicGoAwayFrame& frame) { DCHECK(frame.last_good_stream_id < next_stream_id_); goaway_received_ = true; } void QuicSession::OnConnectionClosed(QuicErrorCode error, bool from_peer) { DCHECK(!connection_->connected()); if (error_ == QUIC_NO_ERROR) { error_ = error; } while (!stream_map_.empty()) { DataStreamMap::iterator it = stream_map_.begin(); QuicStreamId id = it->first; it->second->OnConnectionClosed(error, from_peer); // The stream should call CloseStream as part of OnConnectionClosed. if (stream_map_.find(id) != stream_map_.end()) { LOG(DFATAL) << ENDPOINT << "Stream failed to close under OnConnectionClosed"; CloseStream(id); } } } void QuicSession::OnWindowUpdateFrames( const vector<QuicWindowUpdateFrame>& frames) { bool connection_window_updated = false; for (size_t i = 0; i < frames.size(); ++i) { // Stream may be closed by the time we receive a WINDOW_UPDATE, so we can't // assume that it still exists. QuicStreamId stream_id = frames[i].stream_id; if (stream_id == 0) { // This is a window update that applies to the connection, rather than an // individual stream. DVLOG(1) << ENDPOINT << "Received connection level flow control window update with " "byte offset: " << frames[i].byte_offset; if (FLAGS_enable_quic_connection_flow_control_2 && flow_controller_->UpdateSendWindowOffset(frames[i].byte_offset)) { connection_window_updated = true; } continue; } QuicDataStream* stream = GetDataStream(stream_id); if (stream) { stream->OnWindowUpdateFrame(frames[i]); } } // Connection level flow control window has increased, so blocked streams can // write again. if (connection_window_updated) { OnCanWrite(); } } void QuicSession::OnBlockedFrames(const vector<QuicBlockedFrame>& frames) { for (size_t i = 0; i < frames.size(); ++i) { // 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. DVLOG(1) << ENDPOINT << "Received BLOCKED frame with stream id: " << frames[i].stream_id; } } void QuicSession::OnCanWrite() { // 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. size_t num_writes = write_blocked_streams_.NumBlockedStreams(); if (flow_controller_->IsBlocked()) { // 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. num_writes = 0; if (write_blocked_streams_.crypto_stream_blocked()) { num_writes += 1; } if (write_blocked_streams_.headers_stream_blocked()) { num_writes += 1; } } if (num_writes == 0) { return; } QuicConnection::ScopedPacketBundler ack_bundler( connection_.get(), QuicConnection::NO_ACK); for (size_t i = 0; i < num_writes; ++i) { if (!(write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() || write_blocked_streams_.HasWriteBlockedDataStreams())) { // Writing one stream removed another!? Something's broken. LOG(DFATAL) << "WriteBlockedStream is missing"; connection_->CloseConnection(QUIC_INTERNAL_ERROR, false); return; } if (!connection_->CanWriteStreamData()) { return; } QuicStreamId stream_id = write_blocked_streams_.PopFront(); if (stream_id == kCryptoStreamId) { has_pending_handshake_ = false; // We just popped it. } ReliableQuicStream* stream = GetStream(stream_id); if (stream != NULL && !stream->flow_controller()->IsBlocked()) { // If the stream can't write all bytes, it'll re-add itself to the blocked // list. stream->OnCanWrite(); } } } bool QuicSession::WillingAndAbleToWrite() const { // If the crypto or headers streams are blocked, we want to schedule a write - // they don't get blocked by connection level flow control. Otherwise only // schedule a write if we are not flow control blocked at the connection // level. return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() || (!flow_controller_->IsBlocked() && write_blocked_streams_.HasWriteBlockedDataStreams()); } bool QuicSession::HasPendingHandshake() const { return has_pending_handshake_; } bool QuicSession::HasOpenDataStreams() const { return GetNumOpenStreams() > 0; } QuicConsumedData QuicSession::WritevData( QuicStreamId id, const IOVector& data, QuicStreamOffset offset, bool fin, FecProtection fec_protection, QuicAckNotifier::DelegateInterface* ack_notifier_delegate) { return connection_->SendStreamData(id, data, offset, fin, fec_protection, ack_notifier_delegate); } size_t QuicSession::WriteHeaders( QuicStreamId id, const SpdyHeaderBlock& headers, bool fin, QuicAckNotifier::DelegateInterface* ack_notifier_delegate) { return headers_stream_->WriteHeaders(id, headers, fin, ack_notifier_delegate); } void QuicSession::SendRstStream(QuicStreamId id, QuicRstStreamErrorCode error, QuicStreamOffset bytes_written) { if (connection()->connected()) { // Only send a RST_STREAM frame if still connected. connection_->SendRstStream(id, error, bytes_written); } CloseStreamInner(id, true); } void QuicSession::SendGoAway(QuicErrorCode error_code, const string& reason) { if (goaway_sent_) { return; } goaway_sent_ = true; connection_->SendGoAway(error_code, largest_peer_created_stream_id_, reason); } void QuicSession::CloseStream(QuicStreamId stream_id) { CloseStreamInner(stream_id, false); } void QuicSession::CloseStreamInner(QuicStreamId stream_id, bool locally_reset) { DVLOG(1) << ENDPOINT << "Closing stream " << stream_id; DataStreamMap::iterator it = stream_map_.find(stream_id); if (it == stream_map_.end()) { DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id; return; } QuicDataStream* stream = it->second; // Tell the stream that a RST has been sent. if (locally_reset) { stream->set_rst_sent(true); } closed_streams_.push_back(it->second); // 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. if (!stream->HasFinalReceivedByteOffset() && stream->flow_controller()->IsEnabled() && FLAGS_enable_quic_connection_flow_control_2) { locally_closed_streams_highest_offset_[stream_id] = stream->flow_controller()->highest_received_byte_offset(); } stream_map_.erase(it); stream->OnClose(); } void QuicSession::UpdateFlowControlOnFinalReceivedByteOffset( QuicStreamId stream_id, QuicStreamOffset final_byte_offset) { if (!FLAGS_enable_quic_connection_flow_control_2) { return; } map<QuicStreamId, QuicStreamOffset>::iterator it = locally_closed_streams_highest_offset_.find(stream_id); if (it == locally_closed_streams_highest_offset_.end()) { return; } DVLOG(1) << ENDPOINT << "Received final byte offset " << final_byte_offset << " for stream " << stream_id; uint64 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_->SendConnectionClose( QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA); return; } } flow_controller_->AddBytesConsumed(offset_diff); locally_closed_streams_highest_offset_.erase(it); } bool QuicSession::IsEncryptionEstablished() { return GetCryptoStream()->encryption_established(); } bool QuicSession::IsCryptoHandshakeConfirmed() { return GetCryptoStream()->handshake_confirmed(); } void QuicSession::OnConfigNegotiated() { connection_->SetFromConfig(config_); QuicVersion version = connection()->version(); if (version < QUIC_VERSION_17) { return; } if (version <= QUIC_VERSION_19) { // QUIC_VERSION_17,18,19 don't support independent stream/session flow // control windows. if (config_.HasReceivedInitialFlowControlWindowBytes()) { // Streams which were created before the SHLO was received (0-RTT // requests) are now informed of the peer's initial flow control window. uint32 new_window = config_.ReceivedInitialFlowControlWindowBytes(); OnNewStreamFlowControlWindow(new_window); OnNewSessionFlowControlWindow(new_window); } return; } // QUIC_VERSION_20 and higher can have independent stream and session flow // control windows. 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::OnNewStreamFlowControlWindow(uint32 new_window) { if (new_window < kDefaultFlowControlSendWindow) { LOG(ERROR) << "Peer sent us an invalid stream flow control send window: " << new_window << ", below default: " << kDefaultFlowControlSendWindow; if (connection_->connected()) { connection_->SendConnectionClose(QUIC_FLOW_CONTROL_INVALID_WINDOW); } return; } for (DataStreamMap::iterator it = stream_map_.begin(); it != stream_map_.end(); ++it) { it->second->flow_controller()->UpdateSendWindowOffset(new_window); } } void QuicSession::OnNewSessionFlowControlWindow(uint32 new_window) { if (new_window < kDefaultFlowControlSendWindow) { LOG(ERROR) << "Peer sent us an invalid session flow control send window: " << new_window << ", below default: " << kDefaultFlowControlSendWindow; if (connection_->connected()) { connection_->SendConnectionClose(QUIC_FLOW_CONTROL_INVALID_WINDOW); } 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: break; case ENCRYPTION_REESTABLISHED: // Retransmit originally packets that were sent, since they can't be // decrypted by the peer. connection_->RetransmitUnackedPackets(INITIAL_ENCRYPTION_ONLY); break; case HANDSHAKE_CONFIRMED: LOG_IF(DFATAL, !config_.negotiated()) << ENDPOINT << "Handshake confirmed without parameter negotiation."; // Discard originally encrypted packets, since they can't be decrypted by // the peer. connection_->NeuterUnencryptedPackets(); connection_->SetOverallConnectionTimeout(QuicTime::Delta::Infinite()); max_open_streams_ = config_.max_streams_per_connection(); break; default: LOG(ERROR) << ENDPOINT << "Got unknown handshake event: " << event; } } void QuicSession::OnCryptoHandshakeMessageSent( const CryptoHandshakeMessage& message) { } void QuicSession::OnCryptoHandshakeMessageReceived( const CryptoHandshakeMessage& message) { } QuicConfig* QuicSession::config() { return &config_; } void QuicSession::ActivateStream(QuicDataStream* stream) { DVLOG(1) << ENDPOINT << "num_streams: " << stream_map_.size() << ". activating " << stream->id(); DCHECK_EQ(stream_map_.count(stream->id()), 0u); stream_map_[stream->id()] = stream; } QuicStreamId QuicSession::GetNextStreamId() { QuicStreamId id = next_stream_id_; next_stream_id_ += 2; return id; } ReliableQuicStream* QuicSession::GetStream(const QuicStreamId stream_id) { if (stream_id == kCryptoStreamId) { return GetCryptoStream(); } if (stream_id == kHeadersStreamId) { return headers_stream_.get(); } return GetDataStream(stream_id); } QuicDataStream* QuicSession::GetDataStream(const QuicStreamId stream_id) { if (stream_id == kCryptoStreamId) { DLOG(FATAL) << "Attempt to call GetDataStream with the crypto stream id"; return NULL; } if (stream_id == kHeadersStreamId) { DLOG(FATAL) << "Attempt to call GetDataStream with the headers stream id"; return NULL; } DataStreamMap::iterator it = stream_map_.find(stream_id); if (it != stream_map_.end()) { return it->second; } if (IsClosedStream(stream_id)) { return NULL; } if (stream_id % 2 == next_stream_id_ % 2) { // We've received a frame for a locally-created stream that is not // currently active. This is an error. if (connection()->connected()) { connection()->SendConnectionClose(QUIC_PACKET_FOR_NONEXISTENT_STREAM); } return NULL; } return GetIncomingDataStream(stream_id); } QuicDataStream* QuicSession::GetIncomingDataStream(QuicStreamId stream_id) { if (IsClosedStream(stream_id)) { return NULL; } implicitly_created_streams_.erase(stream_id); if (stream_id > largest_peer_created_stream_id_) { if (stream_id - largest_peer_created_stream_id_ > kMaxStreamIdDelta) { // We may already have sent a connection close due to multiple reset // streams in the same packet. if (connection()->connected()) { LOG(ERROR) << "Trying to get stream: " << stream_id << ", largest peer created stream: " << largest_peer_created_stream_id_ << ", max delta: " << kMaxStreamIdDelta; connection()->SendConnectionClose(QUIC_INVALID_STREAM_ID); } return NULL; } if (largest_peer_created_stream_id_ == 0) { if (is_server()) { largest_peer_created_stream_id_= 3; } else { largest_peer_created_stream_id_= 1; } } for (QuicStreamId id = largest_peer_created_stream_id_ + 2; id < stream_id; id += 2) { implicitly_created_streams_.insert(id); } largest_peer_created_stream_id_ = stream_id; } QuicDataStream* stream = CreateIncomingDataStream(stream_id); if (stream == NULL) { return NULL; } ActivateStream(stream); return stream; } bool QuicSession::IsClosedStream(QuicStreamId id) { DCHECK_NE(0u, id); if (id == kCryptoStreamId) { return false; } if (id == kHeadersStreamId) { return false; } if (ContainsKey(stream_map_, id)) { // Stream is active return false; } if (id % 2 == next_stream_id_ % 2) { // Locally created streams are strictly in-order. If the id is in the // range of created streams and it's not active, it must have been closed. return id < next_stream_id_; } // For peer created streams, we also need to consider implicitly created // streams. return id <= largest_peer_created_stream_id_ && implicitly_created_streams_.count(id) == 0; } size_t QuicSession::GetNumOpenStreams() const { return stream_map_.size() + implicitly_created_streams_.size(); } void QuicSession::MarkWriteBlocked(QuicStreamId id, QuicPriority priority) { #ifndef NDEBUG ReliableQuicStream* stream = GetStream(id); if (stream != NULL) { LOG_IF(DFATAL, priority != stream->EffectivePriority()) << ENDPOINT << "Stream " << id << "Priorities do not match. Got: " << priority << " Expected: " << stream->EffectivePriority(); } else { LOG(DFATAL) << "Marking unknown stream " << id << " blocked."; } #endif if (id == kCryptoStreamId) { DCHECK(!has_pending_handshake_); has_pending_handshake_ = true; // TODO(jar): Be sure to use the highest priority for the crypto stream, // perhaps by adding a "special" priority for it that is higher than // kHighestPriority. priority = kHighestPriority; } write_blocked_streams_.PushBack(id, priority); } bool QuicSession::HasDataToWrite() const { return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() || write_blocked_streams_.HasWriteBlockedDataStreams() || connection_->HasQueuedData(); } bool QuicSession::GetSSLInfo(SSLInfo* ssl_info) const { NOTIMPLEMENTED(); return false; } void QuicSession::PostProcessAfterData() { STLDeleteElements(&closed_streams_); closed_streams_.clear(); } void QuicSession::OnSuccessfulVersionNegotiation(const QuicVersion& version) { if (version < QUIC_VERSION_19) { flow_controller_->Disable(); } // Inform all streams about the negotiated version. They may have been created // with a different version. for (DataStreamMap::iterator it = stream_map_.begin(); it != stream_map_.end(); ++it) { if (version < QUIC_VERSION_17) { it->second->flow_controller()->Disable(); } } } } // namespace net