Queueing QUIC frames to be resent instead of packets and packing RST frames with acks and congestio…

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 40 matching lines @@
 // 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;
 
 bool Near(QuicPacketSequenceNumber a, QuicPacketSequenceNumber b) {
   QuicPacketSequenceNumber delta = (a > b) ? a - b : b - a;
   return delta <= kMaxPacketGap;
 }
 
+QuicConnection::UnackedPacket::UnackedPacket(QuicFrames unacked_frames)
+    : frames(unacked_frames),
+      number_nacks(0) {
+}
+
+QuicConnection::UnackedPacket::UnackedPacket(QuicFrames unacked_frames,
+                                             std::string data)
+    : frames(unacked_frames),
+      number_nacks(0),
+      data(data) {
+}
+
+QuicConnection::UnackedPacket::~UnackedPacket() {
+}
+
 QuicConnection::QuicConnection(QuicGuid guid,
                                IPEndPoint address,
                                QuicConnectionHelperInterface* helper)
     : helper_(helper),
       framer_(QuicDecrypter::Create(kNULL), QuicEncrypter::Create(kNULL)),
       clock_(helper->GetClock()),
       random_generator_(helper->GetRandomGenerator()),
       guid_(guid),
       peer_address_(address),
       should_send_ack_(false),
@@ skipping 21 matching lines @@
   /*
   if (FLAGS_fake_packet_loss_percentage > 0) {
     int32 seed = RandomBase::WeakSeed32();
     LOG(INFO) << "Seeding packet loss with " << seed;
     random_.reset(new MTRandom(seed));
   }
   */
 }
 
 QuicConnection::~QuicConnection() {
-  for (UnackedPacketMap::iterator it = unacked_packets_.begin();
-       it != unacked_packets_.end(); ++it) {
-    delete it->second.packet;
+  for (UnackedPacketMap::iterator u = unacked_packets_.begin();

[jar (doing other things), 2013/01/16 23:59:11]

nit: avoid single letter variable names.  Use of "it" as an iterator appears
commonly enough, but "u" is not so good.

[Ian Swett, 2013/01/17 15:33:27]

Agreed, that is a terrible name.  I have no idea why I changed it.

+       u != unacked_packets_.end(); ++u) {
+    DeleteUnackedPacket(u->second);
   }
+  STLDeleteValues(&unacked_packets_);

[jar (doing other things), 2013/01/16 23:59:11]

nit: It seems strange to iterate over the map starting at line 117, and then
call for another iteration and delete.  I had to look closely to convince myself
that you were doing a correct destructor call.  

I sure wish we had a destructor-like method (Ryan pointed out we can't use a
real destructor, since this is a copyable struct :-()

Nicer might be just add after line 119: 
delete u->second;

To be honest, I'm not sure what the best pattern is... but this seemed more
complex to read than I would have liked.

[Ian Swett, 2013/01/17 15:33:27]

I hated writing it.  I actually made a destructor, realized copy was being used,
contemplating not using it anymore, then decided that was rather ugly as well. 
I'm going to add a comment above the loop to clarify why it's doing this.

What about renaming DeleteUnackedPacket() to DeleteEnclosedFrames()?

[jar (doing other things), 2013/01/17 16:14:33]

+1  That would certainly help IMO.

   STLDeleteValues(&group_map_);
-  // Queued packets that are not to be resent are owned
-  // by the packet queue.
   for (QueuedPacketList::iterator q = queued_packets_.begin();

[jar (doing other things), 2013/01/16 23:59:11]

nit: "q" should be a  better name (...yeah... ya' can't use "it" in both cases
:-/ ).

[Ian Swett, 2013/01/17 15:33:27]

Done.

        q != queued_packets_.end(); ++q) {
-    if (!q->resend) delete q->packet;
+    delete q->packet;
   }
 }
 
+void QuicConnection::DeleteEnclosedFrame(QuicFrame* frame) {
+  switch (frame->type) {
+    case STREAM_FRAME:
+      delete frame->stream_frame;
+      break;
+    case ACK_FRAME:
+      delete frame->ack_frame;
+      break;
+    case CONGESTION_FEEDBACK_FRAME:
+      delete frame->congestion_feedback_frame;
+      break;
+    case RST_STREAM_FRAME:
+      delete frame->rst_stream_frame;
+      break;
+    case CONNECTION_CLOSE_FRAME:
+      delete frame->connection_close_frame;
+      break;
+    case PDU_FRAME:  // Fall through.
+    case NUM_FRAME_TYPES:
+      DCHECK(false) << "Cannot delete type: " << frame->type;
+  }
+}
+
+void QuicConnection::DeleteUnackedPacket(UnackedPacket* unacked) {
+  for (QuicFrames::iterator it = unacked->frames.begin();
+       it != unacked->frames.end(); ++it) {
+    DCHECK(ShouldResend(*it));
+    DeleteEnclosedFrame(&(*it));
+  }
+}
+
+bool QuicConnection::ShouldResend(const QuicFrame& frame) {
+  return frame.type != ACK_FRAME && frame.type != CONGESTION_FEEDBACK_FRAME;
+}
+
 void QuicConnection::OnError(QuicFramer* framer) {
   SendConnectionClose(framer->error());
 }
 
 void QuicConnection::OnPacket(const IPEndPoint& self_address,
                               const IPEndPoint& peer_address) {
   time_of_last_packet_ = clock_->Now();
   DVLOG(1) << "last packet: " << time_of_last_packet_.ToMicroseconds();
 
   // TODO(alyssar, rch) handle migration!
@@ skipping 138 matching lines @@
   int resent_packets = 0;
 
   // Go through the packets we have not received an ack for and see if this
   // incoming_ack shows they've been seen by the peer.
   UnackedPacketMap::iterator it = unacked_packets_.begin();
   while (it != unacked_packets_.end()) {
     if (!incoming_ack.received_info.IsAwaitingPacket(it->first)) {
       // Packet was acked, so remove it from our unacked packet list.
       DVLOG(1) << "Got an ack for " << it->first;
       // TODO(rch): This is inefficient and should be sped up.
+      // TODO(ianswett): Ensure this inner loop is applicable now that we're
+      // always sending packets with new sequence numbers.  I believe it may
+      // only be relevant for the first crypto connect packet, which doesn't
+      // get a new packet sequence number.
       // The acked packet might be queued (if a resend had been attempted).
       for (QueuedPacketList::iterator q = queued_packets_.begin();
            q != queued_packets_.end(); ++q) {
         if (q->sequence_number == it->first) {
           queued_packets_.erase(q);
           break;
         }
       }
       acked_packets.insert(it->first);
-      delete it->second.packet;
+      DeleteUnackedPacket(it->second);
+      delete it->second;
       UnackedPacketMap::iterator it_tmp = it;
       ++it;
       unacked_packets_.erase(it_tmp);
     } else {
       // This is a packet which we planned on resending and has not been
       // seen at the time of this ack being sent out.  See if it's our new
       // lowest unacked packet.
       DVLOG(1) << "still missing " << it->first;
       if (it->first < lowest_unacked) {
         lowest_unacked = it->first;
       }
 
       // Determine if this packet is being explicitly nacked and, if so, if it
       // is worth resending.
       QuicPacketSequenceNumber resend_number = 0;
       if (it->first < peer_largest_observed_packet_) {
         // The peer got packets after this sequence number.  This is an explicit
         // nack.
-        ++(it->second.number_nacks);
-        if (it->second.number_nacks >= kNumberOfNacksBeforeResend &&
+        ++(it->second->number_nacks);
+        if (it->second->number_nacks >= kNumberOfNacksBeforeResend &&
             resent_packets < kMaxResendPerAck) {
           resend_number = it->first;
         }
       }
 
       ++it;
       if (resend_number > 0) {
         ++resent_packets;
         DVLOG(1) << "Trying to resend packet " << resend_number
                  << " as it has been nacked 3 or more times.";
@@ skipping 104 matching lines @@
     QuicStreamId id,
     StringPiece data,
     QuicStreamOffset offset,
     bool fin,
     QuicPacketSequenceNumber* last_packet) {
   size_t total_bytes_consumed = 0;
   bool fin_consumed = false;
 
   while (queued_packets_.empty()) {
     vector<PacketPair> packets;
+    QuicFrames frames;
     size_t bytes_consumed =
-        packet_creator_.DataToStream(id, data, offset, fin, &packets);
+        packet_creator_.DataToStream(id, data, offset, fin, &packets, &frames);
     total_bytes_consumed += bytes_consumed;
     offset += bytes_consumed;
     fin_consumed = fin && bytes_consumed == data.size();
     data.remove_prefix(bytes_consumed);
     DCHECK_LT(0u, packets.size());
 
     for (size_t i = 0; i < packets.size(); ++i) {
+      // Resend is false for FEC packets.
+      bool should_resend = !packets[i].second->IsFecPacket();
       SendPacket(packets[i].first,
                  packets[i].second,
-                 // Resend is false for FEC packets.
-                 !packets[i].second->IsFecPacket(),
+                 should_resend,
                  false,
                  false);
-      unacked_packets_.insert(make_pair(packets[i].first,
-                                        UnackedPacket(packets[i].second)));
+      if (should_resend) {
+        QuicFrames unacked_frames;
+        unacked_frames.push_back(frames[i]);
+        UnackedPacket* unacked = new UnackedPacket(
+            unacked_frames, frames[i].stream_frame->data.as_string());
+        // Ensure the string piece points to the owned copy of the data.
+        unacked->frames[0].stream_frame->data = StringPiece(unacked->data);
+        unacked_packets_.insert(make_pair(packets[i].first, unacked));
+      } else {
+        DeleteEnclosedFrame(&frames[i]);
+      }
     }
 
     if (last_packet != NULL) {
       *last_packet = packets[packets.size() - 1].first;
     }
     if (data.size() == 0) {
       // We're done writing the data.   Exit the loop.
       // We don't make this a precondition beacuse we could have 0 bytes of data
       // if we're simply writing a fin.
       break;
     }
   }
   return QuicConsumedData(total_bytes_consumed, fin_consumed);
 }
 
 void QuicConnection::SendRstStream(QuicStreamId id,
                                    QuicErrorCode error,
                                    QuicStreamOffset offset) {
-  // TODO(ianswett): Queue the frame and use WriteData instead of SendPacket
-  // once we support re-sending frames instead of packets.
-  PacketPair packetpair = packet_creator_.ResetStream(id, offset, error);
+  queued_control_frames_.push_back(QuicFrame(
+      new QuicRstStreamFrame(id, offset, error)));
 
-  SendPacket(packetpair.first, packetpair.second, true, false, false);
-  unacked_packets_.insert(make_pair(packetpair.first,
-                                    UnackedPacket(packetpair.second)));
+  // Try to write immediately if possible.
+  if (CanWrite(false)) {
+    WriteData();
+  }
 }
 
 void QuicConnection::ProcessUdpPacket(const IPEndPoint& self_address,
                                       const IPEndPoint& peer_address,
                                       const QuicEncryptedPacket& packet) {
   last_packet_revived_ = false;
   last_size_ = packet.length();
   framer_.ProcessPacket(self_address, peer_address, packet);
 
   MaybeProcessRevivedPacket();
@@ skipping 20 matching lines @@
       helper_->SetSendAlarm(QuicTime::Delta());
     }
   }
 
   return !write_blocked_;
 }
 
 bool QuicConnection::WriteData() {
   DCHECK_EQ(false, write_blocked_);
   // Serialize the ack and congestion frames before draining the pending queue.
-  QuicFrames frames;
   if (should_send_ack_) {
-    frames.push_back(QuicFrame(&outgoing_ack_));
+    queued_control_frames_.push_back(QuicFrame(&outgoing_ack_));
   }
   if (should_send_congestion_feedback_) {
-    frames.push_back(QuicFrame(&outgoing_congestion_feedback_));
+    queued_control_frames_.push_back(QuicFrame(&outgoing_congestion_feedback_));
   }
-  while (!frames.empty()) {
+  while (!queued_control_frames_.empty()) {
     size_t num_serialized;
-    PacketPair pair = packet_creator_.SerializeFrames(frames, &num_serialized);
+    PacketPair pair = packet_creator_.SerializeFrames(
+        queued_control_frames_, &num_serialized);
+    // If any serialized frames need to be resent, add them to unacked_packets.
+    QuicFrames unacked_frames;
+    for (QuicFrames::const_iterator iter = queued_control_frames_.begin();
+         iter != queued_control_frames_.begin() + num_serialized; ++iter) {
+      if (ShouldResend(*iter)) {
+        unacked_frames.push_back(*iter);
+      }
+    }
+    if (!unacked_frames.empty()) {
+      unacked_packets_.insert(make_pair(pair.first,
+                                        new UnackedPacket(unacked_frames)));
+    }
     queued_packets_.push_back(QueuedPacket(
-        pair.first, pair.second, false, false));
-    frames.erase(frames.begin(), frames.begin() + num_serialized);
+        pair.first, pair.second, !unacked_frames.empty(), false));
+    queued_control_frames_.erase(
+        queued_control_frames_.begin(),
+        queued_control_frames_.begin() + num_serialized);
   }
   should_send_ack_ = false;
   should_send_congestion_feedback_ = false;
 
   size_t num_queued_packets = queued_packets_.size() + 1;
   while (!write_blocked_ && !helper_->IsSendAlarmSet() &&
          !queued_packets_.empty()) {
     // Ensure that from one iteration of this loop to the next we
     // succeeded in sending a packet so we don't infinitely loop.
     // TODO(rch): clean up and close the connection if we really hit this.
@@ skipping 27 matching lines @@
     MaybeResendPacket(sequence_number);
     return true;
   }
 }
 
 void QuicConnection::MaybeResendPacket(
     QuicPacketSequenceNumber sequence_number) {
   UnackedPacketMap::iterator it = unacked_packets_.find(sequence_number);
 
   if (it != unacked_packets_.end()) {
-    QuicPacket* packet = it->second.packet;
+    UnackedPacket* unacked = it->second;
+    // TODO(ianswett): Never change the sequence number of the connect packet.
     unacked_packets_.erase(it);
-    // Re-frame the packet with a new sequence number for resend.
-    QuicPacketSequenceNumber new_sequence_number  =
-        packet_creator_.SetNewSequenceNumber(packet);
+    // Re-packetize the frames with a new sequence number for resend.
+    // Resent data packets do not use FEC, even when it's enabled.
+    PacketPair packetpair = packet_creator_.SerializeAllFrames(unacked->frames);
     DVLOG(1) << "Resending unacked packet " << sequence_number << " as "
-             << new_sequence_number;
-    // Clear the FEC group.
-    framer_.WriteFecGroup(0u, packet);
-    unacked_packets_.insert(make_pair(new_sequence_number,
-                                      UnackedPacket(packet)));
+             << packetpair.first;
+    unacked_packets_.insert(make_pair(packetpair.first, unacked));
 
     // Make sure if this was our least unacked packet, that we update our
     // outgoing ack.  If this wasn't the least unacked, this is a no-op.
     UpdateLeastUnacked(sequence_number);
-    SendPacket(new_sequence_number, packet, true, false, true);
+    SendPacket(packetpair.first, packetpair.second, true, false, true);
   } else {
     DVLOG(2) << "alarm fired for " << sequence_number
              << " but it has been acked";
   }
 }
 
 bool QuicConnection::CanWrite(bool is_retransmit) {
   // TODO(ianswett): If the packet is a retransmit, the current send alarm may
   // be too long.
   if (write_blocked_ || helper_->IsSendAlarmSet()) {
@@ skipping 56 matching lines @@
       */
       return false;
     }
     // TODO(wtc): is it correct to fall through to return true?
   }
 
   time_of_last_packet_ = clock_->Now();
   DVLOG(1) << "last packet: " << time_of_last_packet_.ToMicroseconds();
 
   scheduler_->SentPacket(sequence_number, packet->length(), is_retransmit);
-  if (!should_resend) delete packet;
+  delete packet;
   return true;
 }
 
 bool QuicConnection::ShouldSimulateLostPacket() {
   // TODO(rch): enable this
   return false;
   /*
   return FLAGS_fake_packet_loss_percentage > 0 &&
       random_->Rand32() % 100 < FLAGS_fake_packet_loss_percentage;
   */
@@ skipping 100 matching lines @@
            << " delta:" << delta.ToMicroseconds();
   if (delta >= timeout_) {
     SendConnectionClose(QUIC_CONNECTION_TIMED_OUT);
     return true;
   }
   helper_->SetTimeoutAlarm(timeout_.Subtract(delta));
   return false;
 }
 
 }  // namespace net