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 148 matching lines @@
               is_server),
       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),
-      write_blocked_(false),
       received_packet_manager_(kTCP),
       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_(
@@ skipping 585 matching lines @@
 }
 
 void QuicConnection::SendVersionNegotiationPacket() {
   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_BLOCKED) {
-    write_blocked_ = true;
-  }
   if (result.status == WRITE_STATUS_OK ||
       (result.status == WRITE_STATUS_BLOCKED &&
        writer_->IsWriteBlockedDataBuffered())) {
     pending_version_negotiation_packet_ = false;
     return;
   }
   if (result.status == WRITE_STATUS_ERROR) {
     // We can't send an error as the socket is presumably borked.
     CloseConnection(QUIC_PACKET_WRITE_ERROR, false);
   }
@@ skipping 86 matching lines @@
     }
     DVLOG(1) << ENDPOINT << "Unable to process packet.  Last packet processed: "
              << last_header_.packet_sequence_number;
     return;
   }
   MaybeProcessUndecryptablePackets();
   MaybeProcessRevivedPacket();
 }
 
 bool QuicConnection::OnCanWrite() {
-  write_blocked_ = false;
-  return DoWrite();
-}
+  DCHECK(!writer_->IsWriteBlocked());
 
-bool QuicConnection::WriteIfNotBlocked() {
-  if (write_blocked_) {
-    return false;
-  }
-  return DoWrite();
-}
-
-bool QuicConnection::DoWrite() {
-  DCHECK(!write_blocked_);
   WriteQueuedPackets();
-
   WritePendingRetransmissions();
 
   IsHandshake pending_handshake = visitor_->HasPendingHandshake() ?
       IS_HANDSHAKE : NOT_HANDSHAKE;
   // Sending queued packets may have caused the socket to become write blocked,
   // or the congestion manager to prohibit sending.  If we've sent everything
   // we had queued and we're still not blocked, let the visitor know it can
   // write more.
   if (CanWrite(NOT_RETRANSMISSION, HAS_RETRANSMITTABLE_DATA,
                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 (!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.
       resume_writes_alarm_->Set(clock_->ApproximateNow());
     }
   }
 
-  return !write_blocked_;
+  return !writer_->IsWriteBlocked();
+}
+
+void QuicConnection::WriteIfNotBlocked() {
+  if (!writer_->IsWriteBlocked()) {
+    OnCanWrite();
+  }
 }
 
 bool QuicConnection::ProcessValidatedPacket() {
   if (address_migrating_) {
     SendConnectionCloseWithDetails(
         QUIC_ERROR_MIGRATING_ADDRESS,
         "Address migration is not yet a supported feature");
     return false;
   }
   time_of_last_received_packet_ = clock_->Now();
   DVLOG(1) << ENDPOINT << "time of last received packet: "
            << time_of_last_received_packet_.ToDebuggingValue();
 
   if (is_server_ && encryption_level_ == ENCRYPTION_NONE &&
       last_size_ > options()->max_packet_length) {
     options()->max_packet_length = last_size_;
   }
   return true;
 }
 
-bool QuicConnection::WriteQueuedPackets() {
-  DCHECK(!write_blocked_);
+void QuicConnection::WriteQueuedPackets() {
+  DCHECK(!writer_->IsWriteBlocked());
 
   if (pending_version_negotiation_packet_) {
     SendVersionNegotiationPacket();
   }
 
   QueuedPacketList::iterator packet_iterator = queued_packets_.begin();
-  while (!write_blocked_ && packet_iterator != queued_packets_.end()) {
+  while (!writer_->IsWriteBlocked() &&
+         packet_iterator != queued_packets_.end()) {
     if (WritePacket(packet_iterator->encryption_level,
                     packet_iterator->sequence_number,
                     packet_iterator->packet,
                     packet_iterator->transmission_type,
                     packet_iterator->retransmittable,
                     packet_iterator->handshake,
                     packet_iterator->forced)) {
+      delete packet_iterator->packet;
       packet_iterator = queued_packets_.erase(packet_iterator);
     } else {
       // Continue, because some queued packets may still be writable.
-      // This can happen if a retransmit send fail.
+      // This can happen if a retransmit send fails.
       ++packet_iterator;
     }
   }
-
-  return !write_blocked_;
 }
 
 void QuicConnection::WritePendingRetransmissions() {
   // Keep writing as long as there's a pending retransmission which can be
   // written.
   while (sent_packet_manager_.HasPendingRetransmissions()) {
     const QuicSentPacketManager::PendingRetransmission pending =
         sent_packet_manager_.NextPendingRetransmission();
     if (HasForcedFrames(&pending.retransmittable_frames) == NO_FORCE &&
         !CanWrite(pending.transmission_type, HAS_RETRANSMITTABLE_DATA,
@@ skipping 44 matching lines @@
   if (handshake == IS_HANDSHAKE) {
     return true;
   }
 
   return CanWrite(transmission_type, retransmittable, handshake);
 }
 
 bool QuicConnection::CanWrite(TransmissionType transmission_type,
                               HasRetransmittableData retransmittable,
                               IsHandshake handshake) {
-  if (write_blocked_) {
+  if (writer_->IsWriteBlocked()) {
     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 17 matching lines @@
 }
 
 bool QuicConnection::WritePacket(EncryptionLevel level,
                                  QuicPacketSequenceNumber sequence_number,
                                  QuicPacket* packet,
                                  TransmissionType transmission_type,
                                  HasRetransmittableData retransmittable,
                                  IsHandshake handshake,
                                  Force forced) {
   if (ShouldDiscardPacket(level, sequence_number, retransmittable)) {
-    delete packet;
     return true;
   }
 
-  // If we're write blocked, we know we can't write.
-  if (write_blocked_) {
-    return false;
-  }
-
-  // If we are not forced and we can't write, then simply return false;
+  // If the writer is blocked, we must not write. However, if the packet is
+  // forced (i.e., it's the ConnectionClose packet), we still need to encrypt it
+  // and hand it off to TimeWaitListManager.
+  // We check nonforced packets here and forced after encryption.
   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) {
+  QuicEncryptedPacket* encrypted =
+      framer_.EncryptPacket(level, sequence_number, *packet);
+  if (encrypted == NULL) {
     LOG(DFATAL) << ENDPOINT << "Failed to encrypt packet number "
                 << sequence_number;
+    // CloseConnection does not send close packet, so no infinite loop here.
     CloseConnection(QUIC_ENCRYPTION_FAILURE, false);
     return false;
   }
 
-  // If it's the ConnectionClose packet, the only FORCED frame type,
-  // clone a copy for resending later by the TimeWaitListManager.
-  if (forced == FORCE) {
+  // Forced packets are eventually owned by TimeWaitListManager; nonforced are
+  // 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->Clone());
+    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()) {
+      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
              << " : " << (packet->is_fec_packet() ? "FEC " :
                  (retransmittable == HAS_RETRANSMITTABLE_DATA
@@ skipping 21 matching lines @@
       writer_->WritePacket(encrypted->data(), encrypted->length(),
                            self_address().address(), peer_address(), this);
   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) {
-    // 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.  The helper must call OnPacketSent
     // when the packet is actually sent.
     if (writer_->IsWriteBlockedDataBuffered()) {
-      delete packet;
       return true;
     }
     pending_write_.reset();
     return false;
   }
 
   if (OnPacketSent(result)) {
-    delete packet;
     return true;
   }
   return false;
 }
 
 bool QuicConnection::ShouldDiscardPacket(
     EncryptionLevel level,
     QuicPacketSequenceNumber sequence_number,
     HasRetransmittableData retransmittable) {
   if (!connected_) {
     DVLOG(1) << ENDPOINT
                << "Not sending packet as connection is disconnected.";
     return true;
   }
 
   if (encryption_level_ == ENCRYPTION_FORWARD_SECURE &&
       level == ENCRYPTION_NONE) {
     // Drop packets that are NULL encrypted since the peer won't accept them
     // anymore.
     DVLOG(1) << ENDPOINT << "Dropping packet: " << sequence_number
                << " since the packet is NULL encrypted.";
     sent_packet_manager_.DiscardUnackedPacket(sequence_number);
     return true;
   }
 
+  // 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_.IsUnacked(sequence_number)) {
-      // This is a crazy edge case, but if we retransmit a packet,
-      // (but have to queue it for some reason) then receive an ack
-      // for the previous transmission (but not the retransmission)
-      // then receive a truncated ACK which causes us to raise the
-      // high water mark, all before we're able to send the packet
-      // then we can simply drop it.
-      DVLOG(1) << ENDPOINT << "Dropping packet: " << sequence_number
-                 << " since it has already been acked.";
-      return true;
-    }
-
     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;
@@ skipping 93 matching lines @@
       NOT_HANDSHAKE : packet.retransmittable_frames->HasCryptoHandshake();
   Force forced = HasForcedFrames(packet.retransmittable_frames);
   HasRetransmittableData retransmittable =
       (transmission_type != NOT_RETRANSMISSION ||
        packet.retransmittable_frames != NULL) ?
            HAS_RETRANSMITTABLE_DATA : NO_RETRANSMITTABLE_DATA;
   sent_entropy_manager_.RecordPacketEntropyHash(packet.sequence_number,
                                                 packet.entropy_hash);
   if (WritePacket(level, packet.sequence_number, packet.packet,
                   transmission_type, retransmittable, handshake, forced)) {
+    delete packet.packet;
     return true;
   }
   queued_packets_.push_back(QueuedPacket(packet.sequence_number, packet.packet,
                                          level, transmission_type,
                                          retransmittable, handshake, forced));
   return false;
 }
 
 void QuicConnection::UpdateSentPacketInfo(SentPacketInfo* sent_info) {
   sent_info->least_unacked = sent_packet_manager_.GetLeastUnackedSentPacket();
@@ skipping 149 matching lines @@
   }
   return group_map_[fec_group_num];
 }
 
 void QuicConnection::SendConnectionClose(QuicErrorCode error) {
   SendConnectionCloseWithDetails(error, string());
 }
 
 void QuicConnection::SendConnectionCloseWithDetails(QuicErrorCode error,
                                                     const string& details) {
-  if (!write_blocked_) {
-    SendConnectionClosePacket(error, details);
-  }
+  // If we're write blocked, WritePacket() will not send, but will capture the
+  // serialized packet.
+  SendConnectionClosePacket(error, details);
   CloseConnection(error, false);
 }
 
 void QuicConnection::SendConnectionClosePacket(QuicErrorCode error,
                                                const string& details) {
   DVLOG(1) << ENDPOINT << "Force closing " << guid() << " with error "
              << QuicUtils::ErrorToString(error) << " (" << error << ") "
              << details;
   ScopedPacketBundler ack_bundler(this, true);
   QuicConnectionCloseFrame* frame = new QuicConnectionCloseFrame();
@@ skipping 52 matching lines @@
 
 void QuicConnection::Flush() {
   packet_generator_.FlushAllQueuedFrames();
 }
 
 bool QuicConnection::HasQueuedData() const {
   return pending_version_negotiation_packet_ ||
       !queued_packets_.empty() || packet_generator_.HasQueuedFrames();
 }
 
+bool QuicConnection::CanWriteStreamData() {
+  if (HasQueuedData()) {
+    return false;
+  }
+
+  IsHandshake pending_handshake = visitor_->HasPendingHandshake() ?
+      IS_HANDSHAKE : NOT_HANDSHAKE;
+  // Sending queued packets may have caused the socket to become write blocked,
+  // or the congestion manager to prohibit sending.  If we've sent everything
+  // we had queued and we're still not blocked, let the visitor know it can
+  // write more.
+  return ShouldGeneratePacket(NOT_RETRANSMISSION, HAS_RETRANSMITTABLE_DATA,
+                              pending_handshake);
+}
+
 void QuicConnection::SetIdleNetworkTimeout(QuicTime::Delta timeout) {
   if (timeout < idle_network_timeout_) {
     idle_network_timeout_ = timeout;
     CheckForTimeout();
   } else {
     idle_network_timeout_ = timeout;
   }
 }
 
 void QuicConnection::SetOverallConnectionTimeout(QuicTime::Delta timeout) {
@@ skipping 74 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