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>
 #include <limits>
 #include <memory>
 #include <set>
 #include <utility>
 
 #include "base/logging.h"
 #include "base/rand_util.h"
 #include "base/sha1.h"
 #include "base/stl_util.h"
 #include "net/quic/crypto/quic_decrypter.h"
 #include "net/quic/crypto/quic_encrypter.h"
-#include "net/quic/quic_ack_notifier_manager.h"
 #include "net/quic/quic_bandwidth.h"
 #include "net/quic/quic_utils.h"
 
 using base::hash_map;
 using base::hash_set;
 using base::StringPiece;
 using std::list;
 using std::make_pair;
 using std::min;
 using std::max;
@@ skipping 97 matching lines @@
 
 // An alarm that is scheduled when the sent scheduler requires a
 // a delay before sending packets and fires when the packet may be sent.
 class SendAlarm : public QuicAlarm::Delegate {
  public:
   explicit SendAlarm(QuicConnection* connection)
       : connection_(connection) {
   }
 
   virtual QuicTime OnAlarm() OVERRIDE {
-    connection_->OnCanWrite();
+    connection_->WriteIfNotBlocked();
     // Never reschedule the alarm, since OnCanWrite does that.
     return QuicTime::Zero();
   }
 
  private:
   QuicConnection* connection_;
 };
 
 class TimeoutAlarm : public QuicAlarm::Delegate {
  public:
@@ skipping 63 matching lines @@
       clock_(helper->GetClock()),
       random_generator_(helper->GetRandomGenerator()),
       guid_(guid),
       peer_address_(address),
       largest_seen_packet_with_ack_(0),
       write_blocked_(false),
       ack_alarm_(helper->CreateAlarm(new AckAlarm(this))),
       retransmission_alarm_(helper->CreateAlarm(new RetransmissionAlarm(this))),
       abandon_fec_alarm_(helper->CreateAlarm(new AbandonFecAlarm(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)),
       overall_connection_timeout_(QuicTime::Delta::Infinite()),
       creation_time_(clock_->ApproximateNow()),
       time_of_last_received_packet_(clock_->ApproximateNow()),
       time_of_last_sent_packet_(clock_->ApproximateNow()),
+      sequence_number_of_last_inorder_packet_(0),
       congestion_manager_(clock_, kTCP),
       sent_packet_manager_(is_server, this),
       version_negotiation_state_(START_NEGOTIATION),
       consecutive_rto_count_(0),
       is_server_(is_server),
       connected_(true),
       received_truncated_ack_(false),
       send_ack_in_response_to_packet_(false),
       address_migrating_(false) {
   helper_->SetConnection(this);
@@ skipping 271 matching lines @@
   CloseFecGroupsBefore(incoming_ack.sent_info.least_unacked + 1);
 
   sent_entropy_manager_.ClearEntropyBefore(
       received_packet_manager_.least_packet_awaited_by_peer() - 1);
 
   retransmitted_nacked_packet_count_ = 0;
   SequenceNumberSet acked_packets;
   sent_packet_manager_.OnIncomingAck(
       incoming_ack.received_info, received_truncated_ack_, &acked_packets);
   if (acked_packets.size() > 0) {
-    // The AckNotifierManager should be informed of every ACKed sequence number.
-    ack_notifier_manager_.OnIncomingAck(acked_packets);
-
     // Reset the RTO timeout for each packet when an ack is received.
     if (retransmission_alarm_->IsSet()) {
       retransmission_alarm_->Cancel();
       // Only reschedule the timer if there are outstanding packets.
       if (sent_packet_manager_.HasUnackedPackets()) {
         QuicTime::Delta retransmission_delay =
             congestion_manager_.GetRetransmissionDelay(
                 sent_packet_manager_.GetNumUnackedPackets(), 0);
         retransmission_alarm_->Set(clock_->ApproximateNow().Add(
             retransmission_delay));
@@ skipping 373 matching lines @@
   if (!fin && iov_count == 0) {
     LOG(DFATAL) << "Attempt to send empty stream frame";
   }
 
   // This notifier will be owned by the AckNotifierManager (or deleted below if
   // no data was consumed).
   QuicAckNotifier* notifier = new QuicAckNotifier(delegate);
   QuicConsumedData consumed_data =
       SendvStreamDataInner(id, iov, iov_count, offset, fin, notifier);
 
-  if (consumed_data.bytes_consumed > 0) {
-    // If some data was consumed, then the delegate should be registered for
-    // notification when the data is ACKed.
-    ack_notifier_manager_.AddAckNotifier(notifier);
-    DLOG(INFO) << "Registered AckNotifier.";
-  } else {
+  if (consumed_data.bytes_consumed == 0) {
     // No data was consumed, delete the notifier.
     delete notifier;
   }
 
   return consumed_data;
 }
 
 void QuicConnection::SendRstStream(QuicStreamId id,
                                    QuicRstStreamErrorCode error) {
   LOG(INFO) << "Sending RST_STREAM: " << id << " code: " << error;
@@ skipping 83 matching lines @@
                pending_handshake)) {
     // Set |include_ack| to false in bundler; ack inclusion happens elsewhere.
     scoped_ptr<ScopedPacketBundler> bundler(
         new ScopedPacketBundler(this, false));
     bool all_bytes_written = visitor_->OnCanWrite();
     bundler.reset();
     // After the visitor writes, it may have caused the socket to become write
     // blocked or the congestion manager to prohibit sending, so check again.
     pending_handshake = visitor_->HasPendingHandshake() ? IS_HANDSHAKE
                                                         : NOT_HANDSHAKE;
-    if (!write_blocked_ && !all_bytes_written &&
+    if (!all_bytes_written && !resume_writes_alarm_->IsSet() &&
         CanWrite(NOT_RETRANSMISSION, HAS_RETRANSMITTABLE_DATA,
                  pending_handshake)) {
       // We're not write blocked, but some stream didn't write out all of its
       // bytes.  Register for 'immediate' resumption so we'll keep writing after
       // other quic connections have had a chance to use the socket.
-      send_alarm_->Cancel();
-      send_alarm_->Set(clock_->ApproximateNow());
+      resume_writes_alarm_->Set(clock_->ApproximateNow());
     }
   }
 
   return !write_blocked_;
 }
 
 bool QuicConnection::ProcessValidatedPacket() {
   if (address_migrating_) {
     SendConnectionCloseWithDetails(
         QUIC_ERROR_MIGRATING_ADDRESS,
@@ skipping 45 matching lines @@
     // Retransmitted packets use the same sequence number length as the
     // original.
     // Flush the packet creator before making a new packet.
     // TODO(ianswett): Implement ReserializeAllFrames as a separate path that
     // does not require the creator to be flushed.
     Flush();
     SerializedPacket serialized_packet = packet_creator_.ReserializeAllFrames(
         pending.retransmittable_frames.frames(),
         pending.sequence_number_length);
 
-    // A notifier may be waiting to hear about ACKs for the original sequence
-    // number. Inform them that the sequence number has changed.
-    ack_notifier_manager_.UpdateSequenceNumber(
-        pending.sequence_number, serialized_packet.sequence_number);
-
     DLOG(INFO) << ENDPOINT << "Retransmitting " << pending.sequence_number
                << " as " << serialized_packet.sequence_number;
     if (debug_visitor_) {
       debug_visitor_->OnPacketRetransmitted(
           pending.sequence_number, serialized_packet.sequence_number);
     }
     sent_packet_manager_.OnRetransmittedPacket(
         pending.sequence_number, serialized_packet.sequence_number);
 
     SendOrQueuePacket(pending.retransmittable_frames.encryption_level(),
@@ skipping 61 matching lines @@
   if (handshake == IS_HANDSHAKE) {
     return true;
   }
 
   return CanWrite(transmission_type, retransmittable, handshake);
 }
 
 bool QuicConnection::CanWrite(TransmissionType transmission_type,
                               HasRetransmittableData retransmittable,
                               IsHandshake handshake) {
-  // TODO(ianswett): If the packet is a retransmit, the current send alarm may
-  // be too long.
+  // This check assumes that if the send alarm is set, it applies equally to all
+  // types of transmissions.
   if (write_blocked_ || send_alarm_->IsSet()) {
     return false;
   }
 
   QuicTime now = clock_->Now();
   QuicTime::Delta delay = congestion_manager_.TimeUntilSend(
       now, transmission_type, retransmittable, handshake);
   if (delay.IsInfinite()) {
     return false;
   }
@@ skipping 110 matching lines @@
   // in the SerializedPacket to determine this for a packet.
   IsHandshake handshake = level == ENCRYPTION_NONE ? IS_HANDSHAKE
                                                    : NOT_HANDSHAKE;
 
   // If we are not forced and we can't write, then simply return false;
   if (forced == NO_FORCE &&
       !CanWrite(transmission_type, retransmittable, handshake)) {
     return false;
   }
 
+  // Some encryption algorithms require the packet sequence numbers not be
+  // repeated.
+  DCHECK_LE(sequence_number_of_last_inorder_packet_, sequence_number);
+  // Only increase this when packets have not been queued.  Once they're queued
+  // due to a write block, there is the chance of sending forced and other
+  // higher priority packets out of order.
+  if (queued_packets_.empty()) {
+    sequence_number_of_last_inorder_packet_ = sequence_number;
+  }
+
   scoped_ptr<QuicEncryptedPacket> encrypted(
       framer_.EncryptPacket(level, sequence_number, *packet));
   if (encrypted.get() == NULL) {
     LOG(DFATAL) << ENDPOINT << "Failed to encrypt packet number "
                 << sequence_number;
     CloseConnection(QUIC_ENCRYPTION_FAILURE, false);
     return false;
   }
   DLOG(INFO) << ENDPOINT << "Sending packet number " << sequence_number
              << " : " << (packet->is_fec_packet() ? "FEC " :
@@ skipping 78 matching lines @@
     // WritePacketToWire returned -1, then |error| will be populated with
     // an error code, which we want to pass along to the visitor.
     debug_visitor_->OnPacketSent(sequence_number, level, packet,
                                  bytes_written == -1 ? *error : bytes_written);
   }
   return bytes_written;
 }
 
 bool QuicConnection::OnSerializedPacket(
     const SerializedPacket& serialized_packet) {
-  ack_notifier_manager_.OnSerializedPacket(serialized_packet);
-
   if (serialized_packet.retransmittable_frames) {
     serialized_packet.retransmittable_frames->
         set_encryption_level(encryption_level_);
   }
   sent_packet_manager_.OnSerializedPacket(serialized_packet,
                                           clock_->ApproximateNow());
   // The TransmissionType is NOT_RETRANSMISSION because all retransmissions
   // serialize packets and invoke SendOrQueuePacket directly.
   return SendOrQueuePacket(encryption_level_,
                            serialized_packet.sequence_number,
@@ skipping 432 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