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"
 #include "net/quic/crypto/quic_decrypter.h"
 #include "net/quic/crypto/quic_encrypter.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;
+using std::numeric_limits;
 using std::vector;
 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;
@@ skipping 243 matching lines @@
     case NEGOTIATED_VERSION:
       // Might be old packets that were sent by the client before the version
       // was negotiated. Drop these.
       return false;
 
     default:
       DCHECK(false);
   }
 
   version_negotiation_state_ = NEGOTIATED_VERSION;
+  visitor_->OnSuccessfulVersionNegotiation(received_version);
 
   // Store the new version.
   framer_.set_version(received_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;
 }
 
@@ skipping 78 matching lines @@
         DLOG(WARNING) << ENDPOINT << "Got packet without version flag before "
                       << "version negotiated.";
         // Packets should have the version flag till version negotiation is
         // done.
         CloseConnection(QUIC_INVALID_VERSION, false);
         return false;
       } else {
         DCHECK_EQ(1u, header.public_header.versions.size());
         DCHECK_EQ(header.public_header.versions[0], version());
         version_negotiation_state_ = NEGOTIATED_VERSION;
+        visitor_->OnSuccessfulVersionNegotiation(version());
       }
     } else {
       DCHECK(!header.public_header.version_flag);
       // If the client gets a packet without the version flag from the server
       // it should stop sending version since the version negotiation is done.
       packet_creator_.StopSendingVersion();
       version_negotiation_state_ = NEGOTIATED_VERSION;
+      visitor_->OnSuccessfulVersionNegotiation(version());
     }
   }
 
   DCHECK_EQ(NEGOTIATED_VERSION, version_negotiation_state_);
 
   --stats_.packets_dropped;
   DVLOG(1) << ENDPOINT << "Received packet header: " << header;
   last_header_ = header;
   DCHECK(connected_);
   return true;
@@ skipping 374 matching lines @@
   packet_generator_.StartBatchOperations();
 
   if (last_packet_should_instigate_ack ||
       !g_acks_do_not_instigate_acks) {
     if (send_ack_in_response_to_packet_) {
       SendAck();
     } else if (last_packet_should_instigate_ack) {
       // Set the ack alarm for when any retransmittable frame is received.
       if (!ack_alarm_->IsSet()) {
         ack_alarm_->Set(clock_->ApproximateNow().Add(
-            congestion_manager_.DefaultRetransmissionTime()));
+            congestion_manager_.DelayedAckTime()));
       }
     }
     send_ack_in_response_to_packet_ = !send_ack_in_response_to_packet_;
   }
 
   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 = congestion_manager_.TimeUntilSend(
         time_of_last_received_packet_, NOT_RETRANSMISSION,
@@ skipping 15 matching lines @@
     supported_versions.push_back(kSupportedQuicVersions[i]);
   }
   QuicEncryptedPacket* encrypted =
       packet_creator_.SerializeVersionNegotiationPacket(supported_versions);
   // TODO(satyamshekhar): implement zero server state negotiation.
   int error;
   helper_->WritePacketToWire(*encrypted, &error);
   delete encrypted;
 }
 
-QuicConsumedData QuicConnection::SendStreamData(QuicStreamId id,
-                                                StringPiece data,
-                                                QuicStreamOffset offset,
-                                                bool fin) {
-  // To make reasoning about crypto frames easier, we don't combine them with
-  // any other frames in a single packet.
-  const bool crypto_frame_while_batch_mode =
-      id == kCryptoStreamId && packet_generator_.InBatchMode();
+QuicConsumedData QuicConnection::SendvStreamData(QuicStreamId id,
+                                                 const struct iovec* iov,
+                                                 int count,
+                                                 QuicStreamOffset offset,
+                                                 bool fin) {
+  // TODO(ianswett): Further improve sending by passing the iovec down
+  // instead of batching into multiple stream frames in a single packet.
+  const bool already_in_batch_mode = packet_generator_.InBatchMode();
+  packet_generator_.StartBatchOperations();
 
-  if (crypto_frame_while_batch_mode) {
-    // Flush pending frames to make room for a crypto frame.
+  size_t bytes_written = 0;
+  bool fin_consumed = false;
+  for (int i = 0; i < count; ++i) {
+    bool send_fin = fin && (i == count - 1);
+    if (!send_fin && iov[i].iov_len == 0) {
+      LOG(DFATAL) << "Attempt to send empty stream frame";
+    }
+    QuicConsumedData data_consumed = packet_generator_.ConsumeData(
+        id,
+        StringPiece(static_cast<char*>(iov[i].iov_base), iov[i].iov_len),
+        offset + bytes_written,
+        send_fin);
+    DCHECK_LE(data_consumed.bytes_consumed, numeric_limits<uint32>::max());
+    bytes_written += data_consumed.bytes_consumed;
+    fin_consumed = data_consumed.fin_consumed;
+    // If no bytes were consumed, bail now, because the stream can not write
+    // more data.
+    if (data_consumed.bytes_consumed < iov[i].iov_len) {
+      break;
+    }
+  }
+  // Handle the 0 byte write properly.
+  if (count == 0) {
+    DCHECK(fin);
+    QuicConsumedData data_consumed = packet_generator_.ConsumeData(
+        id, StringPiece(), offset, fin);
+    fin_consumed = data_consumed.fin_consumed;
+  }
+
+  // Leave the generator in the original batch state.
+  if (!already_in_batch_mode) {
     packet_generator_.FinishBatchOperations();
   }
-  QuicConsumedData consumed_data =
-      packet_generator_.ConsumeData(id, data, offset, fin);
-  if (crypto_frame_while_batch_mode) {
-    // Restore batch mode.
-    packet_generator_.StartBatchOperations();
-  }
-  return consumed_data;
+  DCHECK_EQ(already_in_batch_mode, packet_generator_.InBatchMode());
+  return QuicConsumedData(bytes_written, fin_consumed);
 }
 
 QuicConsumedData QuicConnection::SendStreamDataAndNotifyWhenAcked(
     QuicStreamId id,
     StringPiece data,
     QuicStreamOffset offset,
     bool fin,
     QuicAckNotifier::DelegateInterface* delegate) {
+  if (!fin && data.empty()) {
+    LOG(DFATAL) << "Attempt to send empty stream frame";
+  }
   // This notifier will be deleted in ProcessAckFrame once it has seen ACKs for
   // all the consumed data (or below if no data was consumed).
   QuicAckNotifier* notifier = new QuicAckNotifier(delegate);
   QuicConsumedData consumed_data =
       packet_generator_.ConsumeData(id, data, 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_notifiers_.push_back(notifier);
@@ skipping 88 matching lines @@
   const IsHandshake maybe_handshake =
       encryption_level_ == ENCRYPTION_FORWARD_SECURE ? NOT_HANDSHAKE
                                                      : IS_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,
                maybe_handshake)) {
-    const bool in_batch_mode = packet_generator_.InBatchMode();
-    if (!in_batch_mode) {
+    const bool already_in_batch_mode = packet_generator_.InBatchMode();
+    if (!already_in_batch_mode) {
       packet_generator_.StartBatchOperations();
     }
     bool all_bytes_written = visitor_->OnCanWrite();
-    if (!in_batch_mode) {
+    if (!already_in_batch_mode) {
       packet_generator_.FinishBatchOperations();
     }
 
     // After the visitor writes, it may have caused the socket to become write
     // blocked or the congestion manager to prohibit sending, so check again.
     // TODO(jar): we need to pass NOT_HANDSHAKE instead of maybe_handshake to
     // this CanWrite call to avoid getting into an infinite loop calling
     // DoWrite.
     if (!write_blocked_ && !all_bytes_written &&
         CanWrite(NOT_RETRANSMISSION, HAS_RETRANSMITTABLE_DATA,
@@ skipping 326 matching lines @@
            << now.ToDebuggingValue();
   if (WritePacketToWire(sequence_number, level, *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();
+      if (helper_->IsWriteBlockedDataBuffered()) {
+        delete packet;
+        return true;
+      }
+      return false;
     }
     // We can't send an error as the socket is presumably borked.
     CloseConnection(QUIC_PACKET_WRITE_ERROR, false);
     return false;
   }
 
   // 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.
   if (retransmittable == HAS_RETRANSMITTABLE_DATA) {
     SetupRetransmission(sequence_number, level);
   } else if (packet->is_fec_packet()) {
     SetupAbandonFecTimer(sequence_number);
   }
 
   // TODO(ianswett): Change the sequence number length and other packet creator
   // options by a more explicit API than setting a struct value directly.
-  packet_creator_.options()->send_sequence_number_length =
-      CalculateSequenceNumberLength(sequence_number);
+  packet_creator_.UpdateSequenceNumberLength(
+      received_packet_manager_.least_packet_awaited_by_peer(),
+      congestion_manager_.BandwidthEstimate().ToBytesPerPeriod(
+          congestion_manager_.SmoothedRtt()));
 
   congestion_manager_.SentPacket(sequence_number, now, packet->length(),
-                                 retransmission);
+                                 retransmission, retransmittable);
 
   stats_.bytes_sent += encrypted->length();
   ++stats_.packets_sent;
 
   if (retransmission == IS_RETRANSMISSION) {
     stats_.bytes_retransmitted += encrypted->length();
     ++stats_.packets_retransmitted;
   }
 
   delete packet;
   return true;
 }
 
 int QuicConnection::WritePacketToWire(QuicPacketSequenceNumber sequence_number,
                                       EncryptionLevel level,
                                       const QuicEncryptedPacket& packet,
                                       int* error) {
   int bytes_written = helper_->WritePacketToWire(packet, error);
   if (debug_visitor_) {
     // 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;
 }
 
-QuicSequenceNumberLength QuicConnection::CalculateSequenceNumberLength(
-      QuicPacketSequenceNumber sequence_number) {
-  DCHECK_LE(received_packet_manager_.least_packet_awaited_by_peer(),
-            sequence_number);
-  // Since the packet creator will not change sequence number length mid FEC
-  // group, include the size of an FEC group to be safe.
-  const QuicPacketSequenceNumber current_delta =
-      packet_creator_.options()->max_packets_per_fec_group + sequence_number
-      - received_packet_manager_.least_packet_awaited_by_peer();
-  const uint64 congestion_window =
-      congestion_manager_.BandwidthEstimate().ToBytesPerPeriod(
-          congestion_manager_.SmoothedRtt()) /
-          packet_creator_.options()->max_packet_length;
-  const uint64 delta = max(current_delta, congestion_window);
-
-  if (delta < 1 << ((PACKET_1BYTE_SEQUENCE_NUMBER * 8) - 2)) {
-    return PACKET_1BYTE_SEQUENCE_NUMBER;
-  } else if (delta < 1 << ((PACKET_2BYTE_SEQUENCE_NUMBER * 8) - 2)) {
-    return PACKET_2BYTE_SEQUENCE_NUMBER;
-  } else if (delta < 1 << ((PACKET_4BYTE_SEQUENCE_NUMBER * 8) - 2)) {
-    return PACKET_4BYTE_SEQUENCE_NUMBER;
-  }
-  return PACKET_6BYTE_SEQUENCE_NUMBER;
-}
-
 bool QuicConnection::OnSerializedPacket(
     const SerializedPacket& serialized_packet) {
   if (serialized_packet.retransmittable_frames != NULL) {
     DCHECK(unacked_packets_.empty() ||
            unacked_packets_.rbegin()->first <
                serialized_packet.sequence_number);
     // Retransmitted frames will be sent with the same encryption level as the
     // original.
     serialized_packet.retransmittable_frames->set_encryption_level(
         encryption_level_);
@@ skipping 316 matching lines @@
     QuicFecGroup* fec_group = it->second;
     DCHECK(!fec_group->CanRevive());
     FecGroupMap::iterator next = it;
     ++next;
     group_map_.erase(it);
     delete fec_group;
     it = next;
   }
 }
 
+void QuicConnection::Flush() {
+  if (!packet_generator_.InBatchMode()) {
+    return;
+  }
+  packet_generator_.FinishBatchOperations();
+  packet_generator_.StartBatchOperations();
+}
+
 bool QuicConnection::HasQueuedData() const {
   return !queued_packets_.empty() || packet_generator_.HasQueuedFrames();
 }
 
 void QuicConnection::SetIdleNetworkTimeout(QuicTime::Delta timeout) {
   if (timeout < idle_network_timeout_) {
     idle_network_timeout_ = timeout;
     CheckForTimeout();
   } else {
      idle_network_timeout_ = timeout;
@@ skipping 51 matching lines @@
       timeout = connection_timeout;
     }
   }
 
   timeout_alarm_->Cancel();
   timeout_alarm_->Set(clock_->ApproximateNow().Add(timeout));
   return false;
 }
 
 }  // namespace net