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 <string.h>
 #include <sys/types.h>
 #include <algorithm>
 #include <iterator>
@@ skipping 149 matching lines @@
       helper_(helper),
       writer_(writer),
       encryption_level_(ENCRYPTION_NONE),
       clock_(helper->GetClock()),
       random_generator_(helper->GetRandomGenerator()),
       guid_(guid),
       peer_address_(address),
       largest_seen_packet_with_ack_(0),
       pending_version_negotiation_packet_(false),
       received_packet_manager_(kTCP),
+      ack_queued_(false),
       ack_alarm_(helper->CreateAlarm(new AckAlarm(this))),
       retransmission_alarm_(helper->CreateAlarm(new RetransmissionAlarm(this))),
       send_alarm_(helper->CreateAlarm(new SendAlarm(this))),
       resume_writes_alarm_(helper->CreateAlarm(new SendAlarm(this))),
       timeout_alarm_(helper->CreateAlarm(new TimeoutAlarm(this))),
       debug_visitor_(NULL),
       packet_creator_(guid_, &framer_, random_generator_, is_server),
       packet_generator_(this, NULL, &packet_creator_),
       idle_network_timeout_(
           QuicTime::Delta::FromSeconds(kDefaultInitialTimeoutSecs)),
@@ skipping 449 matching lines @@
   DVLOG(1) << ENDPOINT << (last_packet_revived_ ? "Revived" : "Got")
              << " packet " << last_header_.packet_sequence_number
              << " with " << last_ack_frames_.size() << " acks, "
              << last_congestion_frames_.size() << " congestions, "
              << last_goaway_frames_.size() << " goaways, "
              << last_rst_frames_.size() << " rsts, "
              << last_close_frames_.size() << " closes, "
              << last_stream_frames_.size()
              << " stream frames for " << last_header_.public_header.guid;
 
-  // Must called before ack processing, because processing acks removes entries
-  // from unacket_packets_, increasing the least_unacked.
-  const bool last_packet_should_instigate_ack = ShouldLastPacketInstigateAck();
-
-  // If the incoming packet was missing, send an ack immediately.
-  bool send_ack_immediately = received_packet_manager_.IsMissing(
-      last_header_.packet_sequence_number);
+  MaybeQueueAck();
 
   // Discard the packet if the visitor fails to process the stream frames.
   if (!last_stream_frames_.empty() &&
       !visitor_->OnStreamFrames(last_stream_frames_)) {
     return;
   }
 
   received_packet_manager_.RecordPacketReceived(last_size_,
                                                 last_header_,
                                                 time_of_last_received_packet_,
@@ skipping 17 matching lines @@
     sent_packet_manager_.OnIncomingQuicCongestionFeedbackFrame(
         last_congestion_frames_[i], time_of_last_received_packet_);
   }
   if (!last_close_frames_.empty()) {
     CloseConnection(last_close_frames_[0].error_code, true);
     DCHECK(!connected_);
   }
 
   // If there are new missing packets to report, send an ack immediately.
   if (received_packet_manager_.HasNewMissingPackets()) {
-    send_ack_immediately = true;
+    ack_queued_ = true;
+    ack_alarm_->Cancel();
   }
 
-  MaybeSendInResponseToPacket(send_ack_immediately,
-                              last_packet_should_instigate_ack);
+  ClearLastFrames();
+}
 
-  ClearLastFrames();
+void QuicConnection::MaybeQueueAck() {
+  // If the incoming packet was missing, send an ack immediately.
+  ack_queued_ = received_packet_manager_.IsMissing(
+      last_header_.packet_sequence_number);
+
+  // ShouldLastPacketInstigateAck must called before ack processing, because
+  // processing acks removes entries from unacket_packets_, increasing the
+  // least_unacked.
+  if (!ack_queued_ && ShouldLastPacketInstigateAck()) {
+    if (ack_alarm_->IsSet()) {
+      ack_queued_ = true;
+    } else {
+      ack_alarm_->Set(clock_->ApproximateNow().Add(
+          sent_packet_manager_.DelayedAckTime()));
+      DVLOG(1) << "Ack timer set; next packet or timer will trigger ACK.";
+    }
+  }
+
+  if (ack_queued_) {
+    ack_alarm_->Cancel();
+  }
 }
 
 void QuicConnection::ClearLastFrames() {
   last_stream_frames_.clear();
   last_goaway_frames_.clear();
   last_rst_frames_.clear();
   last_ack_frames_.clear();
   last_congestion_frames_.clear();
 }
 
@@ skipping 10 matching lines @@
   return new QuicCongestionFeedbackFrame(outgoing_congestion_feedback_);
 }
 
 bool QuicConnection::ShouldLastPacketInstigateAck() {
   if (!last_stream_frames_.empty() ||
       !last_goaway_frames_.empty() ||
       !last_rst_frames_.empty()) {
     return true;
   }
 
-  // If the peer is still waiting for a packet that we are no
-  // longer planning to send, we should send an ack to raise
-  // the high water mark.
+  // If the peer is still waiting for a packet that we are no longer planning to
+  // send, send an ack to raise the high water mark.
   if (!last_ack_frames_.empty() &&
       !last_ack_frames_.back().received_info.missing_packets.empty()) {
     return sent_packet_manager_.GetLeastUnackedSentPacket() >
         *last_ack_frames_.back().received_info.missing_packets.begin();
   }
   return false;
 }
 
-void QuicConnection::MaybeSendInResponseToPacket(
-    bool send_ack_immediately,
-    bool last_packet_should_instigate_ack) {
-  // |include_ack| is false since we decide about ack bundling below.
+void QuicConnection::MaybeSendInResponseToPacket() {
+  if (!connected_) {
+    return;
+  }
   ScopedPacketBundler bundler(this, false);
-
-  if (last_packet_should_instigate_ack) {
-    // In general, we ack every second packet.  When we don't ack the first
-    // packet, we set the delayed ack alarm.  Thus, if the ack alarm is set
-    // then we know this is the second packet, and we should send an ack.
-    if (send_ack_immediately || ack_alarm_->IsSet()) {
-      SendAck();
-      DCHECK(!ack_alarm_->IsSet());
-    } else {
-      ack_alarm_->Set(clock_->ApproximateNow().Add(
-          sent_packet_manager_.DelayedAckTime()));
-      DVLOG(1) << "Ack timer set; next packet or timer will trigger ACK.";
-    }
+  if (ack_queued_) {
+    SendAck();
   }
 
-  if (!last_ack_frames_.empty()) {
-    // Now the we have received an ack, we might be able to send packets which
-    // are queued locally, or drain streams which are blocked.
-    QuicTime::Delta delay = sent_packet_manager_.TimeUntilSend(
-        time_of_last_received_packet_, NOT_RETRANSMISSION,
-        HAS_RETRANSMITTABLE_DATA, NOT_HANDSHAKE);
-    if (delay.IsZero()) {
-      send_alarm_->Cancel();
-      WriteIfNotBlocked();
-    } else if (!delay.IsInfinite()) {
-      send_alarm_->Cancel();
-      send_alarm_->Set(time_of_last_received_packet_.Add(delay));
-    }
+  // Now that we have received an ack, we might be able to send packets which
+  // are queued locally, or drain streams which are blocked.
+  QuicTime::Delta delay = sent_packet_manager_.TimeUntilSend(
+      time_of_last_received_packet_, NOT_RETRANSMISSION,
+      HAS_RETRANSMITTABLE_DATA, NOT_HANDSHAKE);
+  if (delay.IsZero()) {
+    send_alarm_->Cancel();
+    WriteIfNotBlocked();
+  } else if (!delay.IsInfinite()) {
+    send_alarm_->Cancel();
+    send_alarm_->Set(time_of_last_received_packet_.Add(delay));
   }
 }
 
 void QuicConnection::SendVersionNegotiationPacket() {
+  // TODO(alyssar): implement zero server state negotiation.
+  pending_version_negotiation_packet_ = true;
+  if (writer_->IsWriteBlocked()) {
+    visitor_->OnWriteBlocked();
+    return;
+  }
   scoped_ptr<QuicEncryptedPacket> version_packet(
       packet_creator_.SerializeVersionNegotiationPacket(
           framer_.supported_versions()));
-  // TODO(satyamshekhar): implement zero server state negotiation.
-  WriteResult result =
-      writer_->WritePacket(version_packet->data(), version_packet->length(),
-                           self_address().address(), peer_address(), this);
-  if (result.status == WRITE_STATUS_OK ||
-      (result.status == WRITE_STATUS_BLOCKED &&
-       writer_->IsWriteBlockedDataBuffered())) {
-    pending_version_negotiation_packet_ = false;
-    return;
-  }
+  WriteResult result = writer_->WritePacket(
+      version_packet->data(), version_packet->length(),
+      self_address().address(), peer_address());
+
   if (result.status == WRITE_STATUS_ERROR) {
     // We can't send an error as the socket is presumably borked.
     CloseConnection(QUIC_PACKET_WRITE_ERROR, false);
+    return;
   }
-  pending_version_negotiation_packet_ = true;
+  if (result.status == WRITE_STATUS_BLOCKED) {
+    visitor_->OnWriteBlocked();
+    if (writer_->IsWriteBlockedDataBuffered()) {
+      pending_version_negotiation_packet_ = false;
+    }
+    return;
+  }
+
+  pending_version_negotiation_packet_ = false;
 }
 
 QuicConsumedData QuicConnection::SendStreamData(
     QuicStreamId id,
     const IOVector& data,
     QuicStreamOffset offset,
     bool fin,
     QuicAckNotifier::DelegateInterface* delegate) {
   if (!fin && data.Empty()) {
     LOG(DFATAL) << "Attempt to send empty stream frame";
@@ skipping 70 matching lines @@
     // because the CHLO or SHLO packet was lost.
     if (encryption_level_ != ENCRYPTION_FORWARD_SECURE &&
         framer_.error() == QUIC_DECRYPTION_FAILURE &&
         undecryptable_packets_.size() < kMaxUndecryptablePackets) {
       QueueUndecryptablePacket(packet);
     }
     DVLOG(1) << ENDPOINT << "Unable to process packet.  Last packet processed: "
              << last_header_.packet_sequence_number;
     return;
   }
+
   MaybeProcessUndecryptablePackets();
   MaybeProcessRevivedPacket();
+  MaybeSendInResponseToPacket();
 }
 
 bool QuicConnection::OnCanWrite() {
   DCHECK(!writer_->IsWriteBlocked());
 
   WriteQueuedPackets();
   WritePendingRetransmissions();
 
   IsHandshake pending_handshake = visitor_->HasPendingHandshake() ?
       IS_HANDSHAKE : NOT_HANDSHAKE;
@@ skipping 132 matching lines @@
     return true;
   }
 
   return CanWrite(transmission_type, retransmittable, handshake);
 }
 
 bool QuicConnection::CanWrite(TransmissionType transmission_type,
                               HasRetransmittableData retransmittable,
                               IsHandshake handshake) {
   if (writer_->IsWriteBlocked()) {
+    visitor_->OnWriteBlocked();
     return false;
   }
 
   // TODO(rch): consider removing this check so that if an ACK comes in
   // before the alarm goes it, we might be able send out a packet.
   // This check assumes that if the send alarm is set, it applies equally to all
   // types of transmissions.
   if (send_alarm_->IsSet()) {
     DVLOG(1) << "Send alarm set.  Not sending.";
     return false;
@@ skipping 60 matching lines @@
   // deleted at the end of this call.
   scoped_ptr<QuicEncryptedPacket> encrypted_deleter;
   if (forced == NO_FORCE) {
     encrypted_deleter.reset(encrypted);
   } else {  // forced == FORCE
     DCHECK(connection_close_packet_.get() == NULL);
     connection_close_packet_.reset(encrypted);
     // This assures we won't try to write *forced* packets when blocked.
     // Return true to stop processing.
     if (writer_->IsWriteBlocked()) {
+      visitor_->OnWriteBlocked();
       return true;
     }
   }
 
   if (encrypted->length() > options()->max_packet_length) {
     LOG(DFATAL) << "Writing an encrypted packet larger than max_packet_length:"
                 << options()->max_packet_length << " encrypted length: "
                 << encrypted->length();
   }
   DVLOG(1) << ENDPOINT << "Sending packet number " << sequence_number
@@ skipping 14 matching lines @@
       << " forced: " << (forced == FORCE ? "yes" : "no");
 
   DCHECK(pending_write_.get() == NULL);
   pending_write_.reset(new PendingWrite(sequence_number, transmission_type,
                                         retransmittable, level,
                                         packet.is_fec_packet(),
                                         packet.length()));
 
   WriteResult result =
       writer_->WritePacket(encrypted->data(), encrypted->length(),
-                           self_address().address(), peer_address(), this);
+                           self_address().address(), peer_address());
   if (result.error_code == ERR_IO_PENDING) {
     DCHECK_EQ(WRITE_STATUS_BLOCKED, result.status);
   }
   if (debug_visitor_) {
     // Pass the write result to the visitor.
     debug_visitor_->OnPacketSent(sequence_number, level, *encrypted, result);
   }
   if (result.status == WRITE_STATUS_BLOCKED) {
+    visitor_->OnWriteBlocked();
     // 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.  The helper must call OnPacketSent
     // when the packet is actually sent.
     if (writer_->IsWriteBlockedDataBuffered()) {
       return true;
     }
     pending_write_.reset();
     return false;
   }
@@ skipping 25 matching lines @@
   }
 
   // If the packet has been discarded before sending, don't send it.
   // This occurs if a packet gets serialized, queued, then discarded.
   if (!sent_packet_manager_.IsUnacked(sequence_number)) {
     DVLOG(1) << ENDPOINT << "Dropping packet before sending: "
              << sequence_number << " since it has already been discarded.";
     return true;
   }
 
-  if (retransmittable == HAS_RETRANSMITTABLE_DATA) {
-    if (sent_packet_manager_.IsPreviousTransmission(sequence_number)) {
-      // If somehow we have already retransmitted this packet *before*
-      // we actually send it for the first time (I think this is probably
-      // impossible in the real world), then don't bother sending it.
-      // We don't want to call DiscardUnackedPacket because in this case
-      // the peer has not yet ACK'd the data.  We need the subsequent
-      // retransmission to be sent.
-      DVLOG(1) << ENDPOINT << "Dropping packet: " << sequence_number
-                 << " since it has already been retransmitted.";
-      return true;
-    }
-
-    if (!sent_packet_manager_.HasRetransmittableFrames(sequence_number)) {
-      DVLOG(1) << ENDPOINT << "Dropping packet: " << sequence_number
-                 << " since a previous transmission has been acked.";
-      sent_packet_manager_.DiscardUnackedPacket(sequence_number);
-      return true;
-    }
+  if (retransmittable == HAS_RETRANSMITTABLE_DATA &&
+      !sent_packet_manager_.HasRetransmittableFrames(sequence_number)) {
+    DVLOG(1) << ENDPOINT << "Dropping packet: " << sequence_number
+             << " since a previous transmission has been acked.";
+    sent_packet_manager_.DiscardUnackedPacket(sequence_number);
+    return true;
   }
 
   return false;
 }
 
 bool QuicConnection::OnPacketSent(WriteResult result) {
   DCHECK_NE(WRITE_STATUS_BLOCKED, result.status);
   if (pending_write_.get() == NULL) {
     LOG(DFATAL) << "OnPacketSent called without a pending write.";
     return false;
@@ skipping 434 matching lines @@
   // If we changed the generator's batch state, restore original batch state.
   if (!already_in_batch_mode_) {
     DVLOG(1) << "Leaving Batch Mode.";
     connection_->packet_generator_.FinishBatchOperations();
   }
   DCHECK_EQ(already_in_batch_mode_,
             connection_->packet_generator_.InBatchMode());
 }
 
 }  // namespace net