Land Recent QUIC changes.

net/quic/quic_connection.cc

 // 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_connection.h"
 
 #include <algorithm>
 
 #include "base/logging.h"
 #include "base/stl_util.h"
@@ skipping 12 matching lines @@
 using std::set;
 using std::string;
 
 namespace net {
 namespace {
 
 // The largest gap in packets we'll accept without closing the connection.
 // This will likely have to be tuned.
 const QuicPacketSequenceNumber kMaxPacketGap = 5000;
 
-// The maximum number of nacks which can be transmitted in a single ack packet
-// without exceeding kMaxPacketSize.
-// TODO(satyamshekhar): Get rid of magic numbers and move this to protocol.h
-// 16 - Min ack frame size.
-// 16 - Crypto hash for integrity. Not a static value. Use
-// QuicEncrypter::GetMaxPlaintextSize.
-size_t GetMaxUnackedPackets(bool include_version)  {
-  return (kMaxPacketSize - GetPacketHeaderSize(include_version) - 16 - 16) /
-      kSequenceNumberSize;
-}
-
 // We want to make sure if we get a large nack packet, we don't queue up too
 // many packets at once.  10 is arbitrary.
 const int kMaxRetransmissionsPerAck = 10;
 
 // TCP retransmits after 2 nacks.  We allow for a third in case of out-of-order
 // delivery.
 // TODO(ianswett): Change to match TCP's rule of retransmitting once an ack
 // at least 3 sequence numbers larger arrives.
 const size_t kNumberOfNacksBeforeRetransmission = 3;
 
 // The maxiumum number of packets we'd like to queue.  We may end up queueing
 // more in the case of many control frames.
 // 6 is arbitrary.
 const int kMaxPacketsToSerializeAtOnce = 6;
 
 // Limit the number of packets we send per retransmission-alarm so we
 // eventually cede.  10 is arbitrary.
-const int kMaxPacketsPerRetransmissionAlarm = 10;
+const size_t kMaxPacketsPerRetransmissionAlarm = 10;
+
+// Limit the number of FEC groups to two.  If we get enough out of order packets
+// that this becomes limiting, we can revisit.
+const size_t kMaxFecGroups = 2;
 
 bool Near(QuicPacketSequenceNumber a, QuicPacketSequenceNumber b) {
   QuicPacketSequenceNumber delta = (a > b) ? a - b : b - a;
   return delta <= kMaxPacketGap;
 }
 
 }  // namespace
 
 #define ENDPOINT (is_server_ ? "Server: " : " Client: ")
 
@@ skipping 12 matching lines @@
       peer_address_(address),
       largest_seen_packet_with_ack_(0),
       peer_largest_observed_packet_(0),
       least_packet_awaited_by_peer_(1),
       peer_least_packet_awaiting_ack_(0),
       handling_retransmission_timeout_(false),
       write_blocked_(false),
       debug_visitor_(NULL),
       packet_creator_(guid_, &framer_, random_generator_, is_server),
       packet_generator_(this, &packet_creator_),
-      timeout_(QuicTime::Delta::FromMicroseconds(kDefaultTimeoutUs)),
+      timeout_(QuicTime::Delta::FromSeconds(kDefaultInitialTimeoutSecs)),
       time_of_last_received_packet_(clock_->ApproximateNow()),
       time_of_last_sent_packet_(clock_->ApproximateNow()),
       time_largest_observed_(QuicTime::Zero()),
       congestion_manager_(clock_, kTCP),
       version_negotiation_state_(START_NEGOTIATION),
       quic_version_(kQuicVersion1),
+      max_packets_per_retransmission_alarm_(kMaxPacketsPerRetransmissionAlarm),
       is_server_(is_server),
       connected_(true),
       received_truncated_ack_(false),
-      send_ack_in_response_to_packet_(false) {
+      send_ack_in_response_to_packet_(false),
+      address_migrating_(false) {
   helper_->SetConnection(this);
-  // TODO(satyamshekhar): Have a smaller timeout till version is negotiated and
-  // connection is established (CHLO fully processed).
   helper_->SetTimeoutAlarm(timeout_);
   framer_.set_visitor(this);
   framer_.set_entropy_calculator(&entropy_manager_);
   outgoing_ack_.sent_info.least_unacked = 0;
   outgoing_ack_.sent_info.entropy_hash = 0;
   outgoing_ack_.received_info.largest_observed = 0;
   outgoing_ack_.received_info.entropy_hash = 0;
 
   /*
   if (FLAGS_fake_packet_loss_percentage > 0) {
@@ skipping 35 matching lines @@
   }
   SendConnectionClose(framer->error());
 }
 
 void QuicConnection::OnPacket() {
   // TODO(satyamshekhar): Validate packet before updating the time
   // since it affects the timeout of the connection.
   time_of_last_received_packet_ = clock_->Now();
   DVLOG(1) << "time of last received packet: "
            << time_of_last_received_packet_.ToDebuggingValue();
-
-  // TODO(alyssar, rch) handle migration!
-  self_address_ = last_self_address_;
-  peer_address_ = last_peer_address_;
 }
 
 void QuicConnection::OnPublicResetPacket(
     const QuicPublicResetPacket& packet) {
   if (debug_visitor_) {
     debug_visitor_->OnPublicResetPacket(packet);
   }
   CloseConnection(QUIC_PUBLIC_RESET, true);
 }
 
 bool QuicConnection::OnProtocolVersionMismatch(QuicTag received_version) {
+  if (address_migrating_) {
+    SendConnectionCloseWithDetails(
+        QUIC_ERROR_MIGRATING_ADDRESS,
+        "Address migration is not yet a supported feature");
+  }
+
   // TODO(satyamshekhar): Implement no server state in this mode.
   if (!is_server_) {
     LOG(DFATAL) << "Framer called OnProtocolVersionMismatch for server. "
                 << "Closing connection.";
     CloseConnection(QUIC_INTERNAL_ERROR, false);
     return false;
   }
   DCHECK_NE(quic_version_, received_version);
 
   if (debug_visitor_) {
@@ skipping 31 matching lines @@
   version_negotiation_state_ = NEGOTIATED_VERSION;
   // TODO(satyamshekhar): Store the sequence number of this packet and close the
   // connection if we ever received a packet with incorrect version and whose
   // sequence number is greater.
   return true;
 }
 
 // Handles version negotiation for client connection.
 void QuicConnection::OnVersionNegotiationPacket(
     const QuicVersionNegotiationPacket& packet) {
+  if (address_migrating_) {
+    SendConnectionCloseWithDetails(
+        QUIC_ERROR_MIGRATING_ADDRESS,
+        "Address migration is not yet a supported feature");
+  }
   if (is_server_) {
     LOG(DFATAL) << "Framer parsed VersionNegotiationPacket for server."
                 << "Closing connection.";
     CloseConnection(QUIC_INTERNAL_ERROR, false);
     return;
   }
   if (debug_visitor_) {
     debug_visitor_->OnVersionNegotiationPacket(packet);
   }
 
@@ skipping 22 matching lines @@
 }
 
 void QuicConnection::OnRevivedPacket() {
 }
 
 bool QuicConnection::OnPacketHeader(const QuicPacketHeader& header) {
   if (debug_visitor_) {
     debug_visitor_->OnPacketHeader(header);
   }
 
+  if (address_migrating_) {
+    SendConnectionCloseWithDetails(
+        QUIC_ERROR_MIGRATING_ADDRESS,
+        "Address migration is not yet a supported feature");
+    return false;
+  }
+
   // Will be decrement below if we fall through to return true;
   ++stats_.packets_dropped;
 
   if (header.public_header.guid != guid_) {
     DLOG(INFO) << ENDPOINT << "Ignoring packet from unexpected GUID: "
                << header.public_header.guid << " instead of " << guid_;
     return false;
   }
 
   if (!Near(header.packet_sequence_number,
             last_header_.packet_sequence_number)) {
     DLOG(INFO) << ENDPOINT << "Packet " << header.packet_sequence_number
                << " out of bounds.  Discarding";
-    // TODO(alyssar) close the connection entirely.
+    SendConnectionCloseWithDetails(QUIC_INVALID_PACKET_HEADER,
+                                   "Packet sequence number out of bounds");
     return false;
   }
 
   // If this packet has already been seen, or that the sender
   // has told us will not be retransmitted, then stop processing the packet.
   if (!IsAwaitingPacket(outgoing_ack_.received_info,
                         header.packet_sequence_number)) {
     return false;
   }
 
@@ skipping 24 matching lines @@
 
   --stats_.packets_dropped;
   DVLOG(1) << "Received packet header: " << header;
   last_header_ = header;
   return true;
 }
 
 void QuicConnection::OnFecProtectedPayload(StringPiece payload) {
   DCHECK_NE(0u, last_header_.fec_group);
   QuicFecGroup* group = GetFecGroup();
-  group->Update(last_header_, payload);
+  if (group != NULL) {
+    group->Update(last_header_, payload);
+  }
 }
 
 bool QuicConnection::OnStreamFrame(const QuicStreamFrame& frame) {
   DCHECK(connected_);
   if (debug_visitor_) {
     debug_visitor_->OnStreamFrame(frame);
   }
   last_stream_frames_.push_back(frame);
   return true;
 }
 
 bool QuicConnection::OnAckFrame(const QuicAckFrame& incoming_ack) {
   DCHECK(connected_);
   if (debug_visitor_) {
     debug_visitor_->OnAckFrame(incoming_ack);
   }
   DVLOG(1) << "OnAckFrame: " << incoming_ack;
 
   if (last_header_.packet_sequence_number <= largest_seen_packet_with_ack_) {
     DLOG(INFO) << ENDPOINT << "Received an old ack frame: ignoring";
     return true;
   }
   largest_seen_packet_with_ack_ = last_header_.packet_sequence_number;
 
   if (!ValidateAckFrame(incoming_ack)) {
     SendConnectionClose(QUIC_INVALID_ACK_DATA);
     return false;
   }
 
-  // TODO(satyamshekhar): Not true if missing_packets.size() was actually
-  // kMaxUnackedPackets. This can result in a dead connection if all the
-  // missing packets get lost during retransmission. Now the new packets(or the
-  // older packets) will not be retransmitted due to RTO
-  // since received_truncated_ack_ is true and their sequence_number is >
-  // peer_largest_observed_packet. Fix either by resetting it in
-  // MaybeRetransmitPacketForRTO or keeping an explicit flag for ack truncation.
   received_truncated_ack_ =
       incoming_ack.received_info.missing_packets.size() >=
-      GetMaxUnackedPackets(last_header_.public_header.version_flag);
+      QuicFramer::GetMaxUnackedPackets(last_header_.public_header.version_flag);
 
   UpdatePacketInformationReceivedByPeer(incoming_ack);
   UpdatePacketInformationSentByPeer(incoming_ack);
   congestion_manager_.OnIncomingAckFrame(incoming_ack,
                                          time_of_last_received_packet_);
 
   // Now the we have received an ack, we might be able to send queued packets.
   if (!queued_packets_.empty()) {
     QuicTime::Delta delay = congestion_manager_.TimeUntilSend(
         time_of_last_received_packet_, NOT_RETRANSMISSION,
@@ skipping 36 matching lines @@
     DLOG(ERROR) << ENDPOINT << "Peer's largest_observed packet decreased:"
                 << incoming_ack.received_info.largest_observed << " vs "
                 << peer_largest_observed_packet_;
     // We got an error for data we have not sent.  Error out.
     return false;
   }
 
   // We can't have too many unacked packets, or our ack frames go over
   // kMaxPacketSize.
   DCHECK_LE(incoming_ack.received_info.missing_packets.size(),
-            GetMaxUnackedPackets(last_header_.public_header.version_flag));
+            QuicFramer::GetMaxUnackedPackets(
+                last_header_.public_header.version_flag));
 
   if (incoming_ack.sent_info.least_unacked < peer_least_packet_awaiting_ack_) {
     DLOG(ERROR) << ENDPOINT << "Peer's sent low least_unacked: "
                 << incoming_ack.sent_info.least_unacked
                 << " vs " << peer_least_packet_awaiting_ack_;
     // We never process old ack frames, so this number should only increase.
     return false;
   }
 
   if (incoming_ack.sent_info.least_unacked >
@@ skipping 138 matching lines @@
   DCHECK(outgoing_ack_.received_info.missing_packets.empty() ||
          *outgoing_ack_.received_info.missing_packets.begin() >=
              peer_least_packet_awaiting_ack_);
   // Possibly close any FecGroups which are now irrelevant
   CloseFecGroupsBefore(incoming_ack.sent_info.least_unacked + 1);
 }
 
 void QuicConnection::OnFecData(const QuicFecData& fec) {
   DCHECK_NE(0u, last_header_.fec_group);
   QuicFecGroup* group = GetFecGroup();
-  group->UpdateFec(last_header_.packet_sequence_number,
-                   last_header_.fec_entropy_flag, fec);
+  if (group != NULL) {
+    group->UpdateFec(last_header_.packet_sequence_number,
+                     last_header_.fec_entropy_flag, fec);
+  }
 }
 
 bool QuicConnection::OnRstStreamFrame(const QuicRstStreamFrame& frame) {
   DCHECK(connected_);
   if (debug_visitor_) {
     debug_visitor_->OnRstStreamFrame(frame);
   }
   DLOG(INFO) << "Stream reset with error "
              << QuicUtils::StreamErrorToString(frame.error_code);
   visitor_->OnRstStream(frame);
@@ skipping 15 matching lines @@
 bool QuicConnection::OnGoAwayFrame(const QuicGoAwayFrame& frame) {
   DCHECK(connected_);
   DLOG(INFO) << ENDPOINT << "Go away received with error "
              << QuicUtils::ErrorToString(frame.error_code)
              << " and reason:" << frame.reason_phrase;
   visitor_->OnGoAway(frame);
   return connected_;
 }
 
 void QuicConnection::OnPacketComplete() {
-  // TODO(satyamshekhar): Don't do anything if this packet closed the
-  // connection.
+  // Don't do anything if this packet closed the connection.
+  if (!connected_) {
+    last_stream_frames_.clear();
+    return;
+  }
+
   if (!last_packet_revived_) {
     DLOG(INFO) << ENDPOINT << "Got packet "
                << last_header_.packet_sequence_number
                << " with " << last_stream_frames_.size()
                << " stream frames for " << last_header_.public_header.guid;
     congestion_manager_.RecordIncomingPacket(
         last_size_, last_header_.packet_sequence_number,
         time_of_last_received_packet_, last_packet_revived_);
   } else {
     DLOG(INFO) << ENDPOINT << "Got revived packet with "
@@ skipping 77 matching lines @@
                                       const IPEndPoint& peer_address,
                                       const QuicEncryptedPacket& packet) {
   if (!connected_) {
     return;
   }
   if (debug_visitor_) {
     debug_visitor_->OnPacketReceived(self_address, peer_address, packet);
   }
   last_packet_revived_ = false;
   last_size_ = packet.length();
-  last_self_address_ = self_address;
-  last_peer_address_ = peer_address;
+
+  address_migrating_ = false;
+
+  if (peer_address_.address().empty()) {
+    peer_address_ = peer_address;
+  }
+  if (self_address_.address().empty()) {
+    self_address_ = self_address;
+  }
+
+  if (!(peer_address == peer_address_) && (self_address == self_address_)) {
+    address_migrating_ = true;
+  }
 
   stats_.bytes_received += packet.length();
   ++stats_.packets_received;
 
   if (!framer_.ProcessPacket(packet)) {
     return;
   }
   MaybeProcessRevivedPacket();
 }
 
@@ skipping 87 matching lines @@
             ContainsKey(retransmission_map_, sequence_number));
 
   if (!ContainsKey(unacked_packets_, sequence_number)) {
     DVLOG(2) << "alarm fired for " << sequence_number
              << " but it has been acked or already retransmitted with"
              << " different sequence number.";
     // So no extra delay is added for this packet.
     return true;
   }
 
+  RetransmissionMap::iterator retransmission_it =
+      retransmission_map_.find(sequence_number);
   // If the packet hasn't been acked and we're getting truncated acks, ignore
   // any RTO for packets larger than the peer's largest observed packet; it may
   // have been received by the peer and just wasn't acked due to the ack frame
   // running out of space.
   if (received_truncated_ack_ &&
-      sequence_number > peer_largest_observed_packet_) {
+      sequence_number > peer_largest_observed_packet_ &&
+      // We allow retransmission of already retransmitted packets so that we
+      // retransmit packets that were retransmissions of the packet with
+      // sequence number < the largest observed field of the truncated ack.
+      retransmission_it->second.number_retransmissions == 0) {
     return false;
   } else {
     ++stats_.rto_count;
     RetransmitPacket(sequence_number);
     return true;
   }
 }
 
 void QuicConnection::RetransmitAllUnackedPackets() {
   for (UnackedPacketMap::iterator it = unacked_packets_.begin();
@@ skipping 49 matching lines @@
   // TODO(ianswett): If the packet is a retransmit, the current send alarm may
   // be too long.
   if (write_blocked_ || helper_->IsSendAlarmSet()) {
     return false;
   }
 
   QuicTime now = clock_->Now();
   QuicTime::Delta delay = congestion_manager_.TimeUntilSend(
       now, retransmission, retransmittable);
   if (delay.IsInfinite()) {
-    // TODO(pwestin): should be false but trigger other bugs see b/8350327.
-    return true;
+    return false;
   }
 
   // If the scheduler requires a delay, then we can not send this packet now.
   if (!delay.IsZero()) {
     helper_->SetSendAlarm(now.Add(delay));
     return false;
   }
   return true;
 }
 
@@ skipping 60 matching lines @@
                       ? "data bearing " : " ack only "))
              << " Packet length:" << packet->length();
 
   DCHECK(encrypted->length() <= kMaxPacketSize)
       << "Packet " << sequence_number << " will not be read; too large: "
       << packet->length() << " " << encrypted->length() << " "
       << outgoing_ack_ << " forced: " << (forced == FORCE ? "yes" : "no");
 
   int error;
   QuicTime now = clock_->Now();
-  int rv = helper_->WritePacketToWire(*encrypted, &error);
-  if (rv == -1 && helper_->IsWriteBlocked(error)) {
-    // TODO(satyashekhar): It might be more efficient (fewer system calls), if
-    // all connections share this variable i.e this becomes a part of
-    // PacketWriterInterface.
-    write_blocked_ = true;
-    // If the socket buffers the the data, then the packet should not
-    // be queued and sent again, which would result in an unnecessary duplicate
-    // packet being sent.
-    return helper_->IsWriteBlockedDataBuffered();
+  if (helper_->WritePacketToWire(*encrypted, &error) == -1) {
+    if (helper_->IsWriteBlocked(error)) {
+      // TODO(satyashekhar): It might be more efficient (fewer system calls), if
+      // all connections share this variable i.e this becomes a part of
+      // PacketWriterInterface.
+      write_blocked_ = true;
+      // If the socket buffers the the data, then the packet should not
+      // be queued and sent again, which would result in an unnecessary
+      // duplicate packet being sent.
+      return helper_->IsWriteBlockedDataBuffered();
+    }
+    // We can't send an error as the socket is presumably borked.
+    CloseConnection(QUIC_PACKET_WRITE_ERROR, false);
+    return false;
   }
   time_of_last_sent_packet_ = now;
   DVLOG(1) << "time of last sent packet: " << now.ToDebuggingValue();
-  // TODO(wtc): Is it correct to continue if the write failed.
 
   // Set the retransmit alarm only when we have sent the packet to the client
   // and not when it goes to the pending queue, otherwise we will end up adding
   // an entry to retransmission_timeout_ every time we attempt a write.
   MaybeSetupRetransmission(sequence_number);
 
   congestion_manager_.SentPacket(sequence_number, now, packet->length(),
                                  retransmission);
 
   stats_.bytes_sent += encrypted->length();
@@ skipping 96 matching lines @@
 QuicTime QuicConnection::OnRetransmissionTimeout() {
   // This guards against registering the alarm later than we should.
   //
   // If we have packet A and B in the list and we call
   // MaybeRetransmitPacketForRTO on A, that may trigger a call to
   // SetRetransmissionAlarm if A is retransmitted as C.  In that case we
   // don't want to register the alarm under SetRetransmissionAlarm; we
   // want to set it to the RTO of B when we return from this function.
   handling_retransmission_timeout_ = true;
 
-  for (int i = 0; i < kMaxPacketsPerRetransmissionAlarm &&
+  for (size_t i = 0; i < max_packets_per_retransmission_alarm_ &&
            !retransmission_timeouts_.empty(); ++i) {
     RetransmissionInfo retransmission_info = retransmission_timeouts_.top();
     DCHECK(retransmission_info.scheduled_time.IsInitialized());
     if (retransmission_info.scheduled_time > clock_->ApproximateNow()) {
       break;
     }
     retransmission_timeouts_.pop();
     if (!MaybeRetransmitPacketForRTO(retransmission_info.sequence_number)) {
       DLOG(INFO) << ENDPOINT << "MaybeRetransmitPacketForRTO failed: "
                  << "adding an extra delay for "
@@ skipping 72 matching lines @@
   framer_.ProcessRevivedPacket(&revived_header,
                                StringPiece(revived_payload, len));
 }
 
 QuicFecGroup* QuicConnection::GetFecGroup() {
   QuicFecGroupNumber fec_group_num = last_header_.fec_group;
   if (fec_group_num == 0) {
     return NULL;
   }
   if (group_map_.count(fec_group_num) == 0) {
-    // TODO(rch): limit the number of active FEC groups.
+    if (group_map_.size() >= kMaxFecGroups) {  // Too many groups
+      if (fec_group_num < group_map_.begin()->first) {
+        // The group being requested is a group we've seen before and deleted.
+        // Don't recreate it.
+        return NULL;
+      }
+      // Clear the lowest group number.
+      delete group_map_.begin()->second;
+      group_map_.erase(group_map_.begin());
+    }
     group_map_[fec_group_num] = new QuicFecGroup();
   }
   return group_map_[fec_group_num];
 }
 
 void QuicConnection::SendConnectionClose(QuicErrorCode error) {
   SendConnectionCloseWithDetails(error, string());
 }
 
 void QuicConnection::SendConnectionClosePacket(QuicErrorCode error,
@@ skipping 85 matching lines @@
            << " delta:" << delta.ToMicroseconds();
   if (delta >= timeout_) {
     SendConnectionClose(QUIC_CONNECTION_TIMED_OUT);
     return true;
   }
   helper_->SetTimeoutAlarm(timeout_.Subtract(delta));
   return false;
 }
 
 }  // namespace net