1// Copyright (c) 2012 The Chromium Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5#include "net/quic/quic_connection.h"
6
7#include <string.h>
8#include <sys/types.h>
9#include <algorithm>
10#include <iterator>
11#include <limits>
12#include <memory>
13#include <set>
14#include <utility>
15
16#include "base/debug/stack_trace.h"
17#include "base/logging.h"
18#include "base/stl_util.h"
19#include "net/base/net_errors.h"
20#include "net/quic/crypto/quic_decrypter.h"
21#include "net/quic/crypto/quic_encrypter.h"
22#include "net/quic/iovector.h"
23#include "net/quic/quic_bandwidth.h"
24#include "net/quic/quic_config.h"
25#include "net/quic/quic_fec_group.h"
26#include "net/quic/quic_flags.h"
27#include "net/quic/quic_utils.h"
28
29using base::StringPiece;
30using base::hash_map;
31using base::hash_set;
32using std::list;
33using std::make_pair;
34using std::max;
35using std::min;
36using std::numeric_limits;
37using std::set;
38using std::string;
39using std::vector;
40
41namespace net {
42
43class QuicDecrypter;
44class QuicEncrypter;
45
46namespace {
47
48// The largest gap in packets we'll accept without closing the connection.
49// This will likely have to be tuned.
50const QuicPacketSequenceNumber kMaxPacketGap = 5000;
51
52// Limit the number of FEC groups to two.  If we get enough out of order packets
53// that this becomes limiting, we can revisit.
54const size_t kMaxFecGroups = 2;
55
56// Limit the number of undecryptable packets we buffer in
57// expectation of the CHLO/SHLO arriving.
58const size_t kMaxUndecryptablePackets = 10;
59
60// Maximum number of acks received before sending an ack in response.
61const size_t kMaxPacketsReceivedBeforeAckSend = 20;
62
63bool Near(QuicPacketSequenceNumber a, QuicPacketSequenceNumber b) {
64  QuicPacketSequenceNumber delta = (a > b) ? a - b : b - a;
65  return delta <= kMaxPacketGap;
66}
67
68// An alarm that is scheduled to send an ack if a timeout occurs.
69class AckAlarm : public QuicAlarm::Delegate {
70 public:
71  explicit AckAlarm(QuicConnection* connection)
72      : connection_(connection) {
73  }
74
75  virtual QuicTime OnAlarm() OVERRIDE {
76    connection_->SendAck();
77    return QuicTime::Zero();
78  }
79
80 private:
81  QuicConnection* connection_;
82
83  DISALLOW_COPY_AND_ASSIGN(AckAlarm);
84};
85
86// This alarm will be scheduled any time a data-bearing packet is sent out.
87// When the alarm goes off, the connection checks to see if the oldest packets
88// have been acked, and retransmit them if they have not.
89class RetransmissionAlarm : public QuicAlarm::Delegate {
90 public:
91  explicit RetransmissionAlarm(QuicConnection* connection)
92      : connection_(connection) {
93  }
94
95  virtual QuicTime OnAlarm() OVERRIDE {
96    connection_->OnRetransmissionTimeout();
97    return QuicTime::Zero();
98  }
99
100 private:
101  QuicConnection* connection_;
102
103  DISALLOW_COPY_AND_ASSIGN(RetransmissionAlarm);
104};
105
106// An alarm that is scheduled when the sent scheduler requires a
107// a delay before sending packets and fires when the packet may be sent.
108class SendAlarm : public QuicAlarm::Delegate {
109 public:
110  explicit SendAlarm(QuicConnection* connection)
111      : connection_(connection) {
112  }
113
114  virtual QuicTime OnAlarm() OVERRIDE {
115    connection_->WriteIfNotBlocked();
116    // Never reschedule the alarm, since CanWrite does that.
117    return QuicTime::Zero();
118  }
119
120 private:
121  QuicConnection* connection_;
122
123  DISALLOW_COPY_AND_ASSIGN(SendAlarm);
124};
125
126class TimeoutAlarm : public QuicAlarm::Delegate {
127 public:
128  explicit TimeoutAlarm(QuicConnection* connection)
129      : connection_(connection) {
130  }
131
132  virtual QuicTime OnAlarm() OVERRIDE {
133    connection_->CheckForTimeout();
134    // Never reschedule the alarm, since CheckForTimeout does that.
135    return QuicTime::Zero();
136  }
137
138 private:
139  QuicConnection* connection_;
140
141  DISALLOW_COPY_AND_ASSIGN(TimeoutAlarm);
142};
143
144class PingAlarm : public QuicAlarm::Delegate {
145 public:
146  explicit PingAlarm(QuicConnection* connection)
147      : connection_(connection) {
148  }
149
150  virtual QuicTime OnAlarm() OVERRIDE {
151    connection_->SendPing();
152    return QuicTime::Zero();
153  }
154
155 private:
156  QuicConnection* connection_;
157
158  DISALLOW_COPY_AND_ASSIGN(PingAlarm);
159};
160
161}  // namespace
162
163QuicConnection::QueuedPacket::QueuedPacket(SerializedPacket packet,
164                                           EncryptionLevel level)
165  : serialized_packet(packet),
166    encryption_level(level),
167    transmission_type(NOT_RETRANSMISSION),
168    original_sequence_number(0) {
169}
170
171QuicConnection::QueuedPacket::QueuedPacket(
172    SerializedPacket packet,
173    EncryptionLevel level,
174    TransmissionType transmission_type,
175    QuicPacketSequenceNumber original_sequence_number)
176    : serialized_packet(packet),
177      encryption_level(level),
178      transmission_type(transmission_type),
179      original_sequence_number(original_sequence_number) {
180}
181
182#define ENDPOINT (is_server_ ? "Server: " : " Client: ")
183
184QuicConnection::QuicConnection(QuicConnectionId connection_id,
185                               IPEndPoint address,
186                               QuicConnectionHelperInterface* helper,
187                               const PacketWriterFactory& writer_factory,
188                               bool owns_writer,
189                               bool is_server,
190                               const QuicVersionVector& supported_versions)
191    : framer_(supported_versions, helper->GetClock()->ApproximateNow(),
192              is_server),
193      helper_(helper),
194      writer_(writer_factory.Create(this)),
195      owns_writer_(owns_writer),
196      encryption_level_(ENCRYPTION_NONE),
197      clock_(helper->GetClock()),
198      random_generator_(helper->GetRandomGenerator()),
199      connection_id_(connection_id),
200      peer_address_(address),
201      migrating_peer_port_(0),
202      last_packet_revived_(false),
203      last_size_(0),
204      last_decrypted_packet_level_(ENCRYPTION_NONE),
205      largest_seen_packet_with_ack_(0),
206      largest_seen_packet_with_stop_waiting_(0),
207      pending_version_negotiation_packet_(false),
208      received_packet_manager_(&stats_),
209      ack_queued_(false),
210      num_packets_received_since_last_ack_sent_(0),
211      stop_waiting_count_(0),
212      ack_alarm_(helper->CreateAlarm(new AckAlarm(this))),
213      retransmission_alarm_(helper->CreateAlarm(new RetransmissionAlarm(this))),
214      send_alarm_(helper->CreateAlarm(new SendAlarm(this))),
215      resume_writes_alarm_(helper->CreateAlarm(new SendAlarm(this))),
216      timeout_alarm_(helper->CreateAlarm(new TimeoutAlarm(this))),
217      ping_alarm_(helper->CreateAlarm(new PingAlarm(this))),
218      packet_generator_(connection_id_, &framer_, random_generator_, this),
219      idle_network_timeout_(
220          QuicTime::Delta::FromSeconds(kDefaultInitialTimeoutSecs)),
221      overall_connection_timeout_(QuicTime::Delta::Infinite()),
222      time_of_last_received_packet_(
223          FLAGS_quic_timeouts_require_activity
224              ? QuicTime::Zero() : clock_->ApproximateNow()),
225      time_of_last_sent_new_packet_(
226          FLAGS_quic_timeouts_require_activity
227              ? QuicTime::Zero() : clock_->ApproximateNow()),
228      sequence_number_of_last_sent_packet_(0),
229      sent_packet_manager_(
230          is_server, clock_, &stats_,
231          FLAGS_quic_use_bbr_congestion_control ? kBBR : kCubic,
232          FLAGS_quic_use_time_loss_detection ? kTime : kNack),
233      version_negotiation_state_(START_NEGOTIATION),
234      is_server_(is_server),
235      connected_(true),
236      peer_ip_changed_(false),
237      peer_port_changed_(false),
238      self_ip_changed_(false),
239      self_port_changed_(false) {
240#if 0
241  // TODO(rtenneti): Should we enable this code in chromium?
242  if (!is_server_) {
243    // Pacing will be enabled if the client negotiates it.
244    sent_packet_manager_.MaybeEnablePacing();
245  }
246#endif
247  DVLOG(1) << ENDPOINT << "Created connection with connection_id: "
248           << connection_id;
249  timeout_alarm_->Set(clock_->ApproximateNow().Add(idle_network_timeout_));
250  framer_.set_visitor(this);
251  framer_.set_received_entropy_calculator(&received_packet_manager_);
252  stats_.connection_creation_time = clock_->ApproximateNow();
253  sent_packet_manager_.set_network_change_visitor(this);
254}
255
256QuicConnection::~QuicConnection() {
257  if (owns_writer_) {
258    delete writer_;
259  }
260  STLDeleteElements(&undecryptable_packets_);
261  STLDeleteValues(&group_map_);
262  for (QueuedPacketList::iterator it = queued_packets_.begin();
263       it != queued_packets_.end(); ++it) {
264    delete it->serialized_packet.retransmittable_frames;
265    delete it->serialized_packet.packet;
266  }
267}
268
269void QuicConnection::SetFromConfig(const QuicConfig& config) {
270  SetIdleNetworkTimeout(config.idle_connection_state_lifetime());
271  sent_packet_manager_.SetFromConfig(config);
272}
273
274bool QuicConnection::SelectMutualVersion(
275    const QuicVersionVector& available_versions) {
276  // Try to find the highest mutual version by iterating over supported
277  // versions, starting with the highest, and breaking out of the loop once we
278  // find a matching version in the provided available_versions vector.
279  const QuicVersionVector& supported_versions = framer_.supported_versions();
280  for (size_t i = 0; i < supported_versions.size(); ++i) {
281    const QuicVersion& version = supported_versions[i];
282    if (std::find(available_versions.begin(), available_versions.end(),
283                  version) != available_versions.end()) {
284      framer_.set_version(version);
285      return true;
286    }
287  }
288
289  return false;
290}
291
292void QuicConnection::OnError(QuicFramer* framer) {
293  // Packets that we cannot decrypt are dropped.
294  // TODO(rch): add stats to measure this.
295  if (!connected_ || framer->error() == QUIC_DECRYPTION_FAILURE) {
296    return;
297  }
298  SendConnectionCloseWithDetails(framer->error(), framer->detailed_error());
299}
300
301void QuicConnection::OnPacket() {
302  DCHECK(last_stream_frames_.empty() &&
303         last_ack_frames_.empty() &&
304         last_congestion_frames_.empty() &&
305         last_stop_waiting_frames_.empty() &&
306         last_rst_frames_.empty() &&
307         last_goaway_frames_.empty() &&
308         last_window_update_frames_.empty() &&
309         last_blocked_frames_.empty() &&
310         last_ping_frames_.empty() &&
311         last_close_frames_.empty());
312}
313
314void QuicConnection::OnPublicResetPacket(
315    const QuicPublicResetPacket& packet) {
316  if (debug_visitor_.get() != NULL) {
317    debug_visitor_->OnPublicResetPacket(packet);
318  }
319  CloseConnection(QUIC_PUBLIC_RESET, true);
320
321  DVLOG(1) << ENDPOINT << "Connection " << connection_id()
322           << " closed via QUIC_PUBLIC_RESET from peer.";
323}
324
325bool QuicConnection::OnProtocolVersionMismatch(QuicVersion received_version) {
326  DVLOG(1) << ENDPOINT << "Received packet with mismatched version "
327           << received_version;
328  // TODO(satyamshekhar): Implement no server state in this mode.
329  if (!is_server_) {
330    LOG(DFATAL) << ENDPOINT << "Framer called OnProtocolVersionMismatch. "
331                << "Closing connection.";
332    CloseConnection(QUIC_INTERNAL_ERROR, false);
333    return false;
334  }
335  DCHECK_NE(version(), received_version);
336
337  if (debug_visitor_.get() != NULL) {
338    debug_visitor_->OnProtocolVersionMismatch(received_version);
339  }
340
341  switch (version_negotiation_state_) {
342    case START_NEGOTIATION:
343      if (!framer_.IsSupportedVersion(received_version)) {
344        SendVersionNegotiationPacket();
345        version_negotiation_state_ = NEGOTIATION_IN_PROGRESS;
346        return false;
347      }
348      break;
349
350    case NEGOTIATION_IN_PROGRESS:
351      if (!framer_.IsSupportedVersion(received_version)) {
352        SendVersionNegotiationPacket();
353        return false;
354      }
355      break;
356
357    case NEGOTIATED_VERSION:
358      // Might be old packets that were sent by the client before the version
359      // was negotiated. Drop these.
360      return false;
361
362    default:
363      DCHECK(false);
364  }
365
366  version_negotiation_state_ = NEGOTIATED_VERSION;
367  visitor_->OnSuccessfulVersionNegotiation(received_version);
368  if (debug_visitor_.get() != NULL) {
369    debug_visitor_->OnSuccessfulVersionNegotiation(received_version);
370  }
371  DVLOG(1) << ENDPOINT << "version negotiated " << received_version;
372
373  // Store the new version.
374  framer_.set_version(received_version);
375
376  // TODO(satyamshekhar): Store the sequence number of this packet and close the
377  // connection if we ever received a packet with incorrect version and whose
378  // sequence number is greater.
379  return true;
380}
381
382// Handles version negotiation for client connection.
383void QuicConnection::OnVersionNegotiationPacket(
384    const QuicVersionNegotiationPacket& packet) {
385  if (is_server_) {
386    LOG(DFATAL) << ENDPOINT << "Framer parsed VersionNegotiationPacket."
387                << " Closing connection.";
388    CloseConnection(QUIC_INTERNAL_ERROR, false);
389    return;
390  }
391  if (debug_visitor_.get() != NULL) {
392    debug_visitor_->OnVersionNegotiationPacket(packet);
393  }
394
395  if (version_negotiation_state_ != START_NEGOTIATION) {
396    // Possibly a duplicate version negotiation packet.
397    return;
398  }
399
400  if (std::find(packet.versions.begin(),
401                packet.versions.end(), version()) !=
402      packet.versions.end()) {
403    DLOG(WARNING) << ENDPOINT << "The server already supports our version. "
404                  << "It should have accepted our connection.";
405    // Just drop the connection.
406    CloseConnection(QUIC_INVALID_VERSION_NEGOTIATION_PACKET, false);
407    return;
408  }
409
410  if (!SelectMutualVersion(packet.versions)) {
411    SendConnectionCloseWithDetails(QUIC_INVALID_VERSION,
412                                   "no common version found");
413    return;
414  }
415
416  DVLOG(1) << ENDPOINT
417           << "Negotiated version: " << QuicVersionToString(version());
418  server_supported_versions_ = packet.versions;
419  version_negotiation_state_ = NEGOTIATION_IN_PROGRESS;
420  RetransmitUnackedPackets(ALL_UNACKED_RETRANSMISSION);
421}
422
423void QuicConnection::OnRevivedPacket() {
424}
425
426bool QuicConnection::OnUnauthenticatedPublicHeader(
427    const QuicPacketPublicHeader& header) {
428  return true;
429}
430
431bool QuicConnection::OnUnauthenticatedHeader(const QuicPacketHeader& header) {
432  return true;
433}
434
435void QuicConnection::OnDecryptedPacket(EncryptionLevel level) {
436  last_decrypted_packet_level_ = level;
437}
438
439bool QuicConnection::OnPacketHeader(const QuicPacketHeader& header) {
440  if (debug_visitor_.get() != NULL) {
441    debug_visitor_->OnPacketHeader(header);
442  }
443
444  if (!ProcessValidatedPacket()) {
445    return false;
446  }
447
448  // Will be decrement below if we fall through to return true;
449  ++stats_.packets_dropped;
450
451  if (header.public_header.connection_id != connection_id_) {
452    DVLOG(1) << ENDPOINT << "Ignoring packet from unexpected ConnectionId: "
453             << header.public_header.connection_id << " instead of "
454             << connection_id_;
455    if (debug_visitor_.get() != NULL) {
456      debug_visitor_->OnIncorrectConnectionId(
457          header.public_header.connection_id);
458    }
459    return false;
460  }
461
462  if (!Near(header.packet_sequence_number,
463            last_header_.packet_sequence_number)) {
464    DVLOG(1) << ENDPOINT << "Packet " << header.packet_sequence_number
465             << " out of bounds.  Discarding";
466    SendConnectionCloseWithDetails(QUIC_INVALID_PACKET_HEADER,
467                                   "Packet sequence number out of bounds");
468    return false;
469  }
470
471  // If this packet has already been seen, or that the sender
472  // has told us will not be retransmitted, then stop processing the packet.
473  if (!received_packet_manager_.IsAwaitingPacket(
474          header.packet_sequence_number)) {
475    DVLOG(1) << ENDPOINT << "Packet " << header.packet_sequence_number
476             << " no longer being waited for.  Discarding.";
477    if (debug_visitor_.get() != NULL) {
478      debug_visitor_->OnDuplicatePacket(header.packet_sequence_number);
479    }
480    return false;
481  }
482
483  if (version_negotiation_state_ != NEGOTIATED_VERSION) {
484    if (is_server_) {
485      if (!header.public_header.version_flag) {
486        DLOG(WARNING) << ENDPOINT << "Packet " << header.packet_sequence_number
487                      << " without version flag before version negotiated.";
488        // Packets should have the version flag till version negotiation is
489        // done.
490        CloseConnection(QUIC_INVALID_VERSION, false);
491        return false;
492      } else {
493        DCHECK_EQ(1u, header.public_header.versions.size());
494        DCHECK_EQ(header.public_header.versions[0], version());
495        version_negotiation_state_ = NEGOTIATED_VERSION;
496        visitor_->OnSuccessfulVersionNegotiation(version());
497        if (debug_visitor_.get() != NULL) {
498          debug_visitor_->OnSuccessfulVersionNegotiation(version());
499        }
500      }
501    } else {
502      DCHECK(!header.public_header.version_flag);
503      // If the client gets a packet without the version flag from the server
504      // it should stop sending version since the version negotiation is done.
505      packet_generator_.StopSendingVersion();
506      version_negotiation_state_ = NEGOTIATED_VERSION;
507      visitor_->OnSuccessfulVersionNegotiation(version());
508      if (debug_visitor_.get() != NULL) {
509        debug_visitor_->OnSuccessfulVersionNegotiation(version());
510      }
511    }
512  }
513
514  DCHECK_EQ(NEGOTIATED_VERSION, version_negotiation_state_);
515
516  --stats_.packets_dropped;
517  DVLOG(1) << ENDPOINT << "Received packet header: " << header;
518  last_header_ = header;
519  DCHECK(connected_);
520  return true;
521}
522
523void QuicConnection::OnFecProtectedPayload(StringPiece payload) {
524  DCHECK_EQ(IN_FEC_GROUP, last_header_.is_in_fec_group);
525  DCHECK_NE(0u, last_header_.fec_group);
526  QuicFecGroup* group = GetFecGroup();
527  if (group != NULL) {
528    group->Update(last_decrypted_packet_level_, last_header_, payload);
529  }
530}
531
532bool QuicConnection::OnStreamFrame(const QuicStreamFrame& frame) {
533  DCHECK(connected_);
534  if (debug_visitor_.get() != NULL) {
535    debug_visitor_->OnStreamFrame(frame);
536  }
537  if (frame.stream_id != kCryptoStreamId &&
538      last_decrypted_packet_level_ == ENCRYPTION_NONE) {
539    DLOG(WARNING) << ENDPOINT
540                  << "Received an unencrypted data frame: closing connection";
541    SendConnectionClose(QUIC_UNENCRYPTED_STREAM_DATA);
542    return false;
543  }
544  last_stream_frames_.push_back(frame);
545  return true;
546}
547
548bool QuicConnection::OnAckFrame(const QuicAckFrame& incoming_ack) {
549  DCHECK(connected_);
550  if (debug_visitor_.get() != NULL) {
551    debug_visitor_->OnAckFrame(incoming_ack);
552  }
553  DVLOG(1) << ENDPOINT << "OnAckFrame: " << incoming_ack;
554
555  if (last_header_.packet_sequence_number <= largest_seen_packet_with_ack_) {
556    DVLOG(1) << ENDPOINT << "Received an old ack frame: ignoring";
557    return true;
558  }
559
560  if (!ValidateAckFrame(incoming_ack)) {
561    SendConnectionClose(QUIC_INVALID_ACK_DATA);
562    return false;
563  }
564
565  last_ack_frames_.push_back(incoming_ack);
566  return connected_;
567}
568
569void QuicConnection::ProcessAckFrame(const QuicAckFrame& incoming_ack) {
570  largest_seen_packet_with_ack_ = last_header_.packet_sequence_number;
571  sent_packet_manager_.OnIncomingAck(incoming_ack,
572                                     time_of_last_received_packet_);
573  sent_entropy_manager_.ClearEntropyBefore(
574      sent_packet_manager_.least_packet_awaited_by_peer() - 1);
575  if (sent_packet_manager_.HasPendingRetransmissions()) {
576    WriteIfNotBlocked();
577  }
578
579  // Always reset the retransmission alarm when an ack comes in, since we now
580  // have a better estimate of the current rtt than when it was set.
581  QuicTime retransmission_time = sent_packet_manager_.GetRetransmissionTime();
582  retransmission_alarm_->Update(retransmission_time,
583                                QuicTime::Delta::FromMilliseconds(1));
584}
585
586void QuicConnection::ProcessStopWaitingFrame(
587    const QuicStopWaitingFrame& stop_waiting) {
588  largest_seen_packet_with_stop_waiting_ = last_header_.packet_sequence_number;
589  received_packet_manager_.UpdatePacketInformationSentByPeer(stop_waiting);
590  // Possibly close any FecGroups which are now irrelevant.
591  CloseFecGroupsBefore(stop_waiting.least_unacked + 1);
592}
593
594bool QuicConnection::OnCongestionFeedbackFrame(
595    const QuicCongestionFeedbackFrame& feedback) {
596  DCHECK(connected_);
597  if (debug_visitor_.get() != NULL) {
598    debug_visitor_->OnCongestionFeedbackFrame(feedback);
599  }
600  last_congestion_frames_.push_back(feedback);
601  return connected_;
602}
603
604bool QuicConnection::OnStopWaitingFrame(const QuicStopWaitingFrame& frame) {
605  DCHECK(connected_);
606
607  if (last_header_.packet_sequence_number <=
608      largest_seen_packet_with_stop_waiting_) {
609    DVLOG(1) << ENDPOINT << "Received an old stop waiting frame: ignoring";
610    return true;
611  }
612
613  if (!ValidateStopWaitingFrame(frame)) {
614    SendConnectionClose(QUIC_INVALID_STOP_WAITING_DATA);
615    return false;
616  }
617
618  if (debug_visitor_.get() != NULL) {
619    debug_visitor_->OnStopWaitingFrame(frame);
620  }
621
622  last_stop_waiting_frames_.push_back(frame);
623  return connected_;
624}
625
626bool QuicConnection::OnPingFrame(const QuicPingFrame& frame) {
627  DCHECK(connected_);
628  if (debug_visitor_.get() != NULL) {
629    debug_visitor_->OnPingFrame(frame);
630  }
631  last_ping_frames_.push_back(frame);
632  return true;
633}
634
635bool QuicConnection::ValidateAckFrame(const QuicAckFrame& incoming_ack) {
636  if (incoming_ack.largest_observed > packet_generator_.sequence_number()) {
637    DLOG(ERROR) << ENDPOINT << "Peer's observed unsent packet:"
638                << incoming_ack.largest_observed << " vs "
639                << packet_generator_.sequence_number();
640    // We got an error for data we have not sent.  Error out.
641    return false;
642  }
643
644  if (incoming_ack.largest_observed < sent_packet_manager_.largest_observed()) {
645    DLOG(ERROR) << ENDPOINT << "Peer's largest_observed packet decreased:"
646                << incoming_ack.largest_observed << " vs "
647                << sent_packet_manager_.largest_observed();
648    // A new ack has a diminished largest_observed value.  Error out.
649    // If this was an old packet, we wouldn't even have checked.
650    return false;
651  }
652
653  if (!incoming_ack.missing_packets.empty() &&
654      *incoming_ack.missing_packets.rbegin() > incoming_ack.largest_observed) {
655    DLOG(ERROR) << ENDPOINT << "Peer sent missing packet: "
656                << *incoming_ack.missing_packets.rbegin()
657                << " which is greater than largest observed: "
658                << incoming_ack.largest_observed;
659    return false;
660  }
661
662  if (!incoming_ack.missing_packets.empty() &&
663      *incoming_ack.missing_packets.begin() <
664      sent_packet_manager_.least_packet_awaited_by_peer()) {
665    DLOG(ERROR) << ENDPOINT << "Peer sent missing packet: "
666                << *incoming_ack.missing_packets.begin()
667                << " which is smaller than least_packet_awaited_by_peer_: "
668                << sent_packet_manager_.least_packet_awaited_by_peer();
669    return false;
670  }
671
672  if (!sent_entropy_manager_.IsValidEntropy(
673          incoming_ack.largest_observed,
674          incoming_ack.missing_packets,
675          incoming_ack.entropy_hash)) {
676    DLOG(ERROR) << ENDPOINT << "Peer sent invalid entropy.";
677    return false;
678  }
679
680  for (SequenceNumberSet::const_iterator iter =
681           incoming_ack.revived_packets.begin();
682       iter != incoming_ack.revived_packets.end(); ++iter) {
683    if (!ContainsKey(incoming_ack.missing_packets, *iter)) {
684      DLOG(ERROR) << ENDPOINT
685                  << "Peer specified revived packet which was not missing.";
686      return false;
687    }
688  }
689  return true;
690}
691
692bool QuicConnection::ValidateStopWaitingFrame(
693    const QuicStopWaitingFrame& stop_waiting) {
694  if (stop_waiting.least_unacked <
695      received_packet_manager_.peer_least_packet_awaiting_ack()) {
696    DLOG(ERROR) << ENDPOINT << "Peer's sent low least_unacked: "
697                << stop_waiting.least_unacked << " vs "
698                << received_packet_manager_.peer_least_packet_awaiting_ack();
699    // We never process old ack frames, so this number should only increase.
700    return false;
701  }
702
703  if (stop_waiting.least_unacked >
704      last_header_.packet_sequence_number) {
705    DLOG(ERROR) << ENDPOINT << "Peer sent least_unacked:"
706                << stop_waiting.least_unacked
707                << " greater than the enclosing packet sequence number:"
708                << last_header_.packet_sequence_number;
709    return false;
710  }
711
712  return true;
713}
714
715void QuicConnection::OnFecData(const QuicFecData& fec) {
716  DCHECK_EQ(IN_FEC_GROUP, last_header_.is_in_fec_group);
717  DCHECK_NE(0u, last_header_.fec_group);
718  QuicFecGroup* group = GetFecGroup();
719  if (group != NULL) {
720    group->UpdateFec(last_decrypted_packet_level_,
721                     last_header_.packet_sequence_number, fec);
722  }
723}
724
725bool QuicConnection::OnRstStreamFrame(const QuicRstStreamFrame& frame) {
726  DCHECK(connected_);
727  if (debug_visitor_.get() != NULL) {
728    debug_visitor_->OnRstStreamFrame(frame);
729  }
730  DVLOG(1) << ENDPOINT << "Stream reset with error "
731           << QuicUtils::StreamErrorToString(frame.error_code);
732  last_rst_frames_.push_back(frame);
733  return connected_;
734}
735
736bool QuicConnection::OnConnectionCloseFrame(
737    const QuicConnectionCloseFrame& frame) {
738  DCHECK(connected_);
739  if (debug_visitor_.get() != NULL) {
740    debug_visitor_->OnConnectionCloseFrame(frame);
741  }
742  DVLOG(1) << ENDPOINT << "Connection " << connection_id()
743           << " closed with error "
744           << QuicUtils::ErrorToString(frame.error_code)
745           << " " << frame.error_details;
746  last_close_frames_.push_back(frame);
747  return connected_;
748}
749
750bool QuicConnection::OnGoAwayFrame(const QuicGoAwayFrame& frame) {
751  DCHECK(connected_);
752  if (debug_visitor_.get() != NULL) {
753    debug_visitor_->OnGoAwayFrame(frame);
754  }
755  DVLOG(1) << ENDPOINT << "Go away received with error "
756           << QuicUtils::ErrorToString(frame.error_code)
757           << " and reason:" << frame.reason_phrase;
758  last_goaway_frames_.push_back(frame);
759  return connected_;
760}
761
762bool QuicConnection::OnWindowUpdateFrame(const QuicWindowUpdateFrame& frame) {
763  DCHECK(connected_);
764  if (debug_visitor_.get() != NULL) {
765    debug_visitor_->OnWindowUpdateFrame(frame);
766  }
767  DVLOG(1) << ENDPOINT << "WindowUpdate received for stream: "
768           << frame.stream_id << " with byte offset: " << frame.byte_offset;
769  last_window_update_frames_.push_back(frame);
770  return connected_;
771}
772
773bool QuicConnection::OnBlockedFrame(const QuicBlockedFrame& frame) {
774  DCHECK(connected_);
775  if (debug_visitor_.get() != NULL) {
776    debug_visitor_->OnBlockedFrame(frame);
777  }
778  DVLOG(1) << ENDPOINT << "Blocked frame received for stream: "
779           << frame.stream_id;
780  last_blocked_frames_.push_back(frame);
781  return connected_;
782}
783
784void QuicConnection::OnPacketComplete() {
785  // Don't do anything if this packet closed the connection.
786  if (!connected_) {
787    ClearLastFrames();
788    return;
789  }
790
791  DVLOG(1) << ENDPOINT << (last_packet_revived_ ? "Revived" : "Got")
792           << " packet " << last_header_.packet_sequence_number
793           << " with " << last_stream_frames_.size()<< " stream frames "
794           << last_ack_frames_.size() << " acks, "
795           << last_congestion_frames_.size() << " congestions, "
796           << last_stop_waiting_frames_.size() << " stop_waiting, "
797           << last_rst_frames_.size() << " rsts, "
798           << last_goaway_frames_.size() << " goaways, "
799           << last_window_update_frames_.size() << " window updates, "
800           << last_blocked_frames_.size() << " blocked, "
801           << last_ping_frames_.size() << " pings, "
802           << last_close_frames_.size() << " closes, "
803           << "for " << last_header_.public_header.connection_id;
804
805  ++num_packets_received_since_last_ack_sent_;
806
807  // Call MaybeQueueAck() before recording the received packet, since we want
808  // to trigger an ack if the newly received packet was previously missing.
809  MaybeQueueAck();
810
811  // Record received or revived packet to populate ack info correctly before
812  // processing stream frames, since the processing may result in a response
813  // packet with a bundled ack.
814  if (last_packet_revived_) {
815    received_packet_manager_.RecordPacketRevived(
816        last_header_.packet_sequence_number);
817  } else {
818    received_packet_manager_.RecordPacketReceived(
819        last_size_, last_header_, time_of_last_received_packet_);
820  }
821
822  if (!last_stream_frames_.empty()) {
823    visitor_->OnStreamFrames(last_stream_frames_);
824  }
825
826  for (size_t i = 0; i < last_stream_frames_.size(); ++i) {
827    stats_.stream_bytes_received +=
828        last_stream_frames_[i].data.TotalBufferSize();
829  }
830
831  // Process window updates, blocked, stream resets, acks, then congestion
832  // feedback.
833  if (!last_window_update_frames_.empty()) {
834    visitor_->OnWindowUpdateFrames(last_window_update_frames_);
835  }
836  if (!last_blocked_frames_.empty()) {
837    visitor_->OnBlockedFrames(last_blocked_frames_);
838  }
839  for (size_t i = 0; i < last_goaway_frames_.size(); ++i) {
840    visitor_->OnGoAway(last_goaway_frames_[i]);
841  }
842  for (size_t i = 0; i < last_rst_frames_.size(); ++i) {
843    visitor_->OnRstStream(last_rst_frames_[i]);
844  }
845  for (size_t i = 0; i < last_ack_frames_.size(); ++i) {
846    ProcessAckFrame(last_ack_frames_[i]);
847  }
848  for (size_t i = 0; i < last_congestion_frames_.size(); ++i) {
849    sent_packet_manager_.OnIncomingQuicCongestionFeedbackFrame(
850        last_congestion_frames_[i], time_of_last_received_packet_);
851  }
852  for (size_t i = 0; i < last_stop_waiting_frames_.size(); ++i) {
853    ProcessStopWaitingFrame(last_stop_waiting_frames_[i]);
854  }
855  if (!last_close_frames_.empty()) {
856    CloseConnection(last_close_frames_[0].error_code, true);
857    DCHECK(!connected_);
858  }
859
860  // If there are new missing packets to report, send an ack immediately.
861  if (received_packet_manager_.HasNewMissingPackets()) {
862    ack_queued_ = true;
863    ack_alarm_->Cancel();
864  }
865
866  UpdateStopWaitingCount();
867
868  ClearLastFrames();
869}
870
871void QuicConnection::MaybeQueueAck() {
872  // If the incoming packet was missing, send an ack immediately.
873  ack_queued_ = received_packet_manager_.IsMissing(
874      last_header_.packet_sequence_number);
875
876  if (!ack_queued_ && ShouldLastPacketInstigateAck()) {
877    if (ack_alarm_->IsSet()) {
878      ack_queued_ = true;
879    } else {
880      // Send an ack much more quickly for crypto handshake packets.
881      QuicTime::Delta delayed_ack_time = sent_packet_manager_.DelayedAckTime();
882      if (last_stream_frames_.size() == 1 &&
883          last_stream_frames_[0].stream_id == kCryptoStreamId) {
884        delayed_ack_time = QuicTime::Delta::Zero();
885      }
886      ack_alarm_->Set(clock_->ApproximateNow().Add(delayed_ack_time));
887      DVLOG(1) << "Ack timer set; next packet or timer will trigger ACK.";
888    }
889  }
890
891  if (ack_queued_) {
892    ack_alarm_->Cancel();
893  }
894}
895
896void QuicConnection::ClearLastFrames() {
897  last_stream_frames_.clear();
898  last_ack_frames_.clear();
899  last_congestion_frames_.clear();
900  last_stop_waiting_frames_.clear();
901  last_rst_frames_.clear();
902  last_goaway_frames_.clear();
903  last_window_update_frames_.clear();
904  last_blocked_frames_.clear();
905  last_ping_frames_.clear();
906  last_close_frames_.clear();
907}
908
909QuicAckFrame* QuicConnection::CreateAckFrame() {
910  QuicAckFrame* outgoing_ack = new QuicAckFrame();
911  received_packet_manager_.UpdateReceivedPacketInfo(
912      outgoing_ack, clock_->ApproximateNow());
913  DVLOG(1) << ENDPOINT << "Creating ack frame: " << *outgoing_ack;
914  return outgoing_ack;
915}
916
917QuicCongestionFeedbackFrame* QuicConnection::CreateFeedbackFrame() {
918  return new QuicCongestionFeedbackFrame(outgoing_congestion_feedback_);
919}
920
921QuicStopWaitingFrame* QuicConnection::CreateStopWaitingFrame() {
922  QuicStopWaitingFrame stop_waiting;
923  UpdateStopWaiting(&stop_waiting);
924  return new QuicStopWaitingFrame(stop_waiting);
925}
926
927bool QuicConnection::ShouldLastPacketInstigateAck() const {
928  if (!last_stream_frames_.empty() ||
929      !last_goaway_frames_.empty() ||
930      !last_rst_frames_.empty() ||
931      !last_window_update_frames_.empty() ||
932      !last_blocked_frames_.empty() ||
933      !last_ping_frames_.empty()) {
934    return true;
935  }
936
937  if (!last_ack_frames_.empty() && last_ack_frames_.back().is_truncated) {
938    return true;
939  }
940  // Always send an ack every 20 packets in order to allow the peer to discard
941  // information from the SentPacketManager and provide an RTT measurement.
942  if (num_packets_received_since_last_ack_sent_ >=
943          kMaxPacketsReceivedBeforeAckSend) {
944    return true;
945  }
946  return false;
947}
948
949void QuicConnection::UpdateStopWaitingCount() {
950  if (last_ack_frames_.empty()) {
951    return;
952  }
953
954  // If the peer is still waiting for a packet that we are no longer planning to
955  // send, send an ack to raise the high water mark.
956  if (!last_ack_frames_.back().missing_packets.empty() &&
957      GetLeastUnacked() > *last_ack_frames_.back().missing_packets.begin()) {
958    ++stop_waiting_count_;
959  } else {
960    stop_waiting_count_ = 0;
961  }
962}
963
964QuicPacketSequenceNumber QuicConnection::GetLeastUnacked() const {
965  return sent_packet_manager_.GetLeastUnacked();
966}
967
968void QuicConnection::MaybeSendInResponseToPacket() {
969  if (!connected_) {
970    return;
971  }
972  ScopedPacketBundler bundler(this, ack_queued_ ? SEND_ACK : NO_ACK);
973
974  // Now that we have received an ack, we might be able to send packets which
975  // are queued locally, or drain streams which are blocked.
976  if (CanWrite(HAS_RETRANSMITTABLE_DATA)) {
977    OnCanWrite();
978  }
979}
980
981void QuicConnection::SendVersionNegotiationPacket() {
982  // TODO(alyssar): implement zero server state negotiation.
983  pending_version_negotiation_packet_ = true;
984  if (writer_->IsWriteBlocked()) {
985    visitor_->OnWriteBlocked();
986    return;
987  }
988  DVLOG(1) << ENDPOINT << "Sending version negotiation packet: {"
989           << QuicVersionVectorToString(framer_.supported_versions()) << "}";
990  scoped_ptr<QuicEncryptedPacket> version_packet(
991      packet_generator_.SerializeVersionNegotiationPacket(
992          framer_.supported_versions()));
993  WriteResult result = writer_->WritePacket(
994      version_packet->data(), version_packet->length(),
995      self_address().address(), peer_address());
996
997  if (result.status == WRITE_STATUS_ERROR) {
998    // We can't send an error as the socket is presumably borked.
999    CloseConnection(QUIC_PACKET_WRITE_ERROR, false);
1000    return;
1001  }
1002  if (result.status == WRITE_STATUS_BLOCKED) {
1003    visitor_->OnWriteBlocked();
1004    if (writer_->IsWriteBlockedDataBuffered()) {
1005      pending_version_negotiation_packet_ = false;
1006    }
1007    return;
1008  }
1009
1010  pending_version_negotiation_packet_ = false;
1011}
1012
1013QuicConsumedData QuicConnection::SendStreamData(
1014    QuicStreamId id,
1015    const IOVector& data,
1016    QuicStreamOffset offset,
1017    bool fin,
1018    FecProtection fec_protection,
1019    QuicAckNotifier::DelegateInterface* delegate) {
1020  if (!fin && data.Empty()) {
1021    LOG(DFATAL) << "Attempt to send empty stream frame";
1022  }
1023
1024  // This notifier will be owned by the AckNotifierManager (or deleted below if
1025  // no data or FIN was consumed).
1026  QuicAckNotifier* notifier = NULL;
1027  if (delegate) {
1028    notifier = new QuicAckNotifier(delegate);
1029  }
1030
1031  // Opportunistically bundle an ack with every outgoing packet.
1032  // Particularly, we want to bundle with handshake packets since we don't know
1033  // which decrypter will be used on an ack packet following a handshake
1034  // packet (a handshake packet from client to server could result in a REJ or a
1035  // SHLO from the server, leading to two different decrypters at the server.)
1036  //
1037  // TODO(jri): Note that ConsumeData may cause a response packet to be sent.
1038  // We may end up sending stale ack information if there are undecryptable
1039  // packets hanging around and/or there are revivable packets which may get
1040  // handled after this packet is sent. Change ScopedPacketBundler to do the
1041  // right thing: check ack_queued_, and then check undecryptable packets and
1042  // also if there is possibility of revival. Only bundle an ack if there's no
1043  // processing left that may cause received_info_ to change.
1044  ScopedPacketBundler ack_bundler(this, BUNDLE_PENDING_ACK);
1045  QuicConsumedData consumed_data =
1046      packet_generator_.ConsumeData(id, data, offset, fin, fec_protection,
1047                                    notifier);
1048
1049  if (notifier &&
1050      (consumed_data.bytes_consumed == 0 && !consumed_data.fin_consumed)) {
1051    // No data was consumed, nor was a fin consumed, so delete the notifier.
1052    delete notifier;
1053  }
1054
1055  return consumed_data;
1056}
1057
1058void QuicConnection::SendRstStream(QuicStreamId id,
1059                                   QuicRstStreamErrorCode error,
1060                                   QuicStreamOffset bytes_written) {
1061  // Opportunistically bundle an ack with this outgoing packet.
1062  ScopedPacketBundler ack_bundler(this, BUNDLE_PENDING_ACK);
1063  packet_generator_.AddControlFrame(QuicFrame(new QuicRstStreamFrame(
1064      id, AdjustErrorForVersion(error, version()), bytes_written)));
1065}
1066
1067void QuicConnection::SendWindowUpdate(QuicStreamId id,
1068                                      QuicStreamOffset byte_offset) {
1069  // Opportunistically bundle an ack with this outgoing packet.
1070  ScopedPacketBundler ack_bundler(this, BUNDLE_PENDING_ACK);
1071  packet_generator_.AddControlFrame(
1072      QuicFrame(new QuicWindowUpdateFrame(id, byte_offset)));
1073}
1074
1075void QuicConnection::SendBlocked(QuicStreamId id) {
1076  // Opportunistically bundle an ack with this outgoing packet.
1077  ScopedPacketBundler ack_bundler(this, BUNDLE_PENDING_ACK);
1078  packet_generator_.AddControlFrame(QuicFrame(new QuicBlockedFrame(id)));
1079}
1080
1081const QuicConnectionStats& QuicConnection::GetStats() {
1082  // Update rtt and estimated bandwidth.
1083  stats_.min_rtt_us =
1084      sent_packet_manager_.GetRttStats()->min_rtt().ToMicroseconds();
1085  stats_.srtt_us =
1086      sent_packet_manager_.GetRttStats()->SmoothedRtt().ToMicroseconds();
1087  stats_.estimated_bandwidth =
1088      sent_packet_manager_.BandwidthEstimate().ToBytesPerSecond();
1089  stats_.congestion_window = sent_packet_manager_.GetCongestionWindow();
1090  stats_.slow_start_threshold = sent_packet_manager_.GetSlowStartThreshold();
1091  stats_.max_packet_size = packet_generator_.max_packet_length();
1092  return stats_;
1093}
1094
1095void QuicConnection::ProcessUdpPacket(const IPEndPoint& self_address,
1096                                      const IPEndPoint& peer_address,
1097                                      const QuicEncryptedPacket& packet) {
1098  if (!connected_) {
1099    return;
1100  }
1101  if (debug_visitor_.get() != NULL) {
1102    debug_visitor_->OnPacketReceived(self_address, peer_address, packet);
1103  }
1104  last_packet_revived_ = false;
1105  last_size_ = packet.length();
1106
1107  CheckForAddressMigration(self_address, peer_address);
1108
1109  stats_.bytes_received += packet.length();
1110  ++stats_.packets_received;
1111
1112  if (!framer_.ProcessPacket(packet)) {
1113    // If we are unable to decrypt this packet, it might be
1114    // because the CHLO or SHLO packet was lost.
1115    if (framer_.error() == QUIC_DECRYPTION_FAILURE) {
1116      if (encryption_level_ != ENCRYPTION_FORWARD_SECURE &&
1117          undecryptable_packets_.size() < kMaxUndecryptablePackets) {
1118        QueueUndecryptablePacket(packet);
1119      } else if (debug_visitor_.get() != NULL) {
1120        debug_visitor_->OnUndecryptablePacket();
1121      }
1122    }
1123    DVLOG(1) << ENDPOINT << "Unable to process packet.  Last packet processed: "
1124             << last_header_.packet_sequence_number;
1125    return;
1126  }
1127
1128  ++stats_.packets_processed;
1129  MaybeProcessUndecryptablePackets();
1130  MaybeProcessRevivedPacket();
1131  MaybeSendInResponseToPacket();
1132  SetPingAlarm();
1133}
1134
1135void QuicConnection::CheckForAddressMigration(
1136    const IPEndPoint& self_address, const IPEndPoint& peer_address) {
1137  peer_ip_changed_ = false;
1138  peer_port_changed_ = false;
1139  self_ip_changed_ = false;
1140  self_port_changed_ = false;
1141
1142  if (peer_address_.address().empty()) {
1143    peer_address_ = peer_address;
1144  }
1145  if (self_address_.address().empty()) {
1146    self_address_ = self_address;
1147  }
1148
1149  if (!peer_address.address().empty() && !peer_address_.address().empty()) {
1150    peer_ip_changed_ = (peer_address.address() != peer_address_.address());
1151    peer_port_changed_ = (peer_address.port() != peer_address_.port());
1152
1153    // Store in case we want to migrate connection in ProcessValidatedPacket.
1154    migrating_peer_port_ = peer_address.port();
1155  }
1156
1157  if (!self_address.address().empty() && !self_address_.address().empty()) {
1158    self_ip_changed_ = (self_address.address() != self_address_.address());
1159    self_port_changed_ = (self_address.port() != self_address_.port());
1160  }
1161}
1162
1163void QuicConnection::OnCanWrite() {
1164  DCHECK(!writer_->IsWriteBlocked());
1165
1166  WriteQueuedPackets();
1167  WritePendingRetransmissions();
1168
1169  // Sending queued packets may have caused the socket to become write blocked,
1170  // or the congestion manager to prohibit sending.  If we've sent everything
1171  // we had queued and we're still not blocked, let the visitor know it can
1172  // write more.
1173  if (!CanWrite(HAS_RETRANSMITTABLE_DATA)) {
1174    return;
1175  }
1176
1177  {  // Limit the scope of the bundler.
1178    // Set |include_ack| to false in bundler; ack inclusion happens elsewhere.
1179    ScopedPacketBundler bundler(this, NO_ACK);
1180    visitor_->OnCanWrite();
1181  }
1182
1183  // After the visitor writes, it may have caused the socket to become write
1184  // blocked or the congestion manager to prohibit sending, so check again.
1185  if (visitor_->WillingAndAbleToWrite() &&
1186      !resume_writes_alarm_->IsSet() &&
1187      CanWrite(HAS_RETRANSMITTABLE_DATA)) {
1188    // We're not write blocked, but some stream didn't write out all of its
1189    // bytes. Register for 'immediate' resumption so we'll keep writing after
1190    // other connections and events have had a chance to use the thread.
1191    resume_writes_alarm_->Set(clock_->ApproximateNow());
1192  }
1193}
1194
1195void QuicConnection::WriteIfNotBlocked() {
1196  if (!writer_->IsWriteBlocked()) {
1197    OnCanWrite();
1198  }
1199}
1200
1201bool QuicConnection::ProcessValidatedPacket() {
1202  if (peer_ip_changed_ || self_ip_changed_ || self_port_changed_) {
1203    SendConnectionCloseWithDetails(
1204        QUIC_ERROR_MIGRATING_ADDRESS,
1205        "Neither IP address migration, nor self port migration are supported.");
1206    return false;
1207  }
1208
1209  // Peer port migration is supported, do it now if port has changed.
1210  if (peer_port_changed_) {
1211    DVLOG(1) << ENDPOINT << "Peer's port changed from "
1212             << peer_address_.port() << " to " << migrating_peer_port_
1213             << ", migrating connection.";
1214    peer_address_ = IPEndPoint(peer_address_.address(), migrating_peer_port_);
1215  }
1216
1217  time_of_last_received_packet_ = clock_->Now();
1218  DVLOG(1) << ENDPOINT << "time of last received packet: "
1219           << time_of_last_received_packet_.ToDebuggingValue();
1220
1221  if (is_server_ && encryption_level_ == ENCRYPTION_NONE &&
1222      last_size_ > packet_generator_.max_packet_length()) {
1223    packet_generator_.set_max_packet_length(last_size_);
1224  }
1225  return true;
1226}
1227
1228void QuicConnection::WriteQueuedPackets() {
1229  DCHECK(!writer_->IsWriteBlocked());
1230
1231  if (pending_version_negotiation_packet_) {
1232    SendVersionNegotiationPacket();
1233  }
1234
1235  QueuedPacketList::iterator packet_iterator = queued_packets_.begin();
1236  while (packet_iterator != queued_packets_.end() &&
1237         WritePacket(&(*packet_iterator))) {
1238    packet_iterator = queued_packets_.erase(packet_iterator);
1239  }
1240}
1241
1242void QuicConnection::WritePendingRetransmissions() {
1243  // Keep writing as long as there's a pending retransmission which can be
1244  // written.
1245  while (sent_packet_manager_.HasPendingRetransmissions()) {
1246    const QuicSentPacketManager::PendingRetransmission pending =
1247        sent_packet_manager_.NextPendingRetransmission();
1248    if (!CanWrite(HAS_RETRANSMITTABLE_DATA)) {
1249      break;
1250    }
1251
1252    // Re-packetize the frames with a new sequence number for retransmission.
1253    // Retransmitted data packets do not use FEC, even when it's enabled.
1254    // Retransmitted packets use the same sequence number length as the
1255    // original.
1256    // Flush the packet generator before making a new packet.
1257    // TODO(ianswett): Implement ReserializeAllFrames as a separate path that
1258    // does not require the creator to be flushed.
1259    packet_generator_.FlushAllQueuedFrames();
1260    SerializedPacket serialized_packet = packet_generator_.ReserializeAllFrames(
1261        pending.retransmittable_frames.frames(),
1262        pending.sequence_number_length);
1263
1264    DVLOG(1) << ENDPOINT << "Retransmitting " << pending.sequence_number
1265             << " as " << serialized_packet.sequence_number;
1266    SendOrQueuePacket(
1267        QueuedPacket(serialized_packet,
1268                     pending.retransmittable_frames.encryption_level(),
1269                     pending.transmission_type,
1270                     pending.sequence_number));
1271  }
1272}
1273
1274void QuicConnection::RetransmitUnackedPackets(
1275    TransmissionType retransmission_type) {
1276  sent_packet_manager_.RetransmitUnackedPackets(retransmission_type);
1277
1278  WriteIfNotBlocked();
1279}
1280
1281void QuicConnection::NeuterUnencryptedPackets() {
1282  sent_packet_manager_.NeuterUnencryptedPackets();
1283  // This may have changed the retransmission timer, so re-arm it.
1284  QuicTime retransmission_time = sent_packet_manager_.GetRetransmissionTime();
1285  retransmission_alarm_->Update(retransmission_time,
1286                                QuicTime::Delta::FromMilliseconds(1));
1287}
1288
1289bool QuicConnection::ShouldGeneratePacket(
1290    TransmissionType transmission_type,
1291    HasRetransmittableData retransmittable,
1292    IsHandshake handshake) {
1293  // We should serialize handshake packets immediately to ensure that they
1294  // end up sent at the right encryption level.
1295  if (handshake == IS_HANDSHAKE) {
1296    return true;
1297  }
1298
1299  return CanWrite(retransmittable);
1300}
1301
1302bool QuicConnection::CanWrite(HasRetransmittableData retransmittable) {
1303  if (!connected_) {
1304    return false;
1305  }
1306
1307  if (writer_->IsWriteBlocked()) {
1308    visitor_->OnWriteBlocked();
1309    return false;
1310  }
1311
1312  QuicTime now = clock_->Now();
1313  QuicTime::Delta delay = sent_packet_manager_.TimeUntilSend(
1314      now, retransmittable);
1315  if (delay.IsInfinite()) {
1316    send_alarm_->Cancel();
1317    return false;
1318  }
1319
1320  // If the scheduler requires a delay, then we can not send this packet now.
1321  if (!delay.IsZero()) {
1322    send_alarm_->Update(now.Add(delay), QuicTime::Delta::FromMilliseconds(1));
1323    DVLOG(1) << "Delaying sending.";
1324    return false;
1325  }
1326  send_alarm_->Cancel();
1327  return true;
1328}
1329
1330bool QuicConnection::WritePacket(QueuedPacket* packet) {
1331  if (!WritePacketInner(packet)) {
1332    return false;
1333  }
1334  delete packet->serialized_packet.retransmittable_frames;
1335  delete packet->serialized_packet.packet;
1336  packet->serialized_packet.retransmittable_frames = NULL;
1337  packet->serialized_packet.packet = NULL;
1338  return true;
1339}
1340
1341bool QuicConnection::WritePacketInner(QueuedPacket* packet) {
1342  if (ShouldDiscardPacket(*packet)) {
1343    ++stats_.packets_discarded;
1344    return true;
1345  }
1346  // Connection close packets are encrypted and saved, so don't exit early.
1347  if (writer_->IsWriteBlocked() && !IsConnectionClose(*packet)) {
1348    return false;
1349  }
1350
1351  QuicPacketSequenceNumber sequence_number =
1352      packet->serialized_packet.sequence_number;
1353  // Some encryption algorithms require the packet sequence numbers not be
1354  // repeated.
1355  DCHECK_LE(sequence_number_of_last_sent_packet_, sequence_number);
1356  sequence_number_of_last_sent_packet_ = sequence_number;
1357
1358  QuicEncryptedPacket* encrypted = framer_.EncryptPacket(
1359      packet->encryption_level,
1360      sequence_number,
1361      *packet->serialized_packet.packet);
1362  if (encrypted == NULL) {
1363    LOG(DFATAL) << ENDPOINT << "Failed to encrypt packet number "
1364                << sequence_number;
1365    // CloseConnection does not send close packet, so no infinite loop here.
1366    CloseConnection(QUIC_ENCRYPTION_FAILURE, false);
1367    return false;
1368  }
1369
1370  // Connection close packets are eventually owned by TimeWaitListManager.
1371  // Others are deleted at the end of this call.
1372  scoped_ptr<QuicEncryptedPacket> encrypted_deleter;
1373  if (IsConnectionClose(*packet)) {
1374    DCHECK(connection_close_packet_.get() == NULL);
1375    connection_close_packet_.reset(encrypted);
1376    // This assures we won't try to write *forced* packets when blocked.
1377    // Return true to stop processing.
1378    if (writer_->IsWriteBlocked()) {
1379      visitor_->OnWriteBlocked();
1380      return true;
1381    }
1382  } else {
1383    encrypted_deleter.reset(encrypted);
1384  }
1385
1386  if (!FLAGS_quic_allow_oversized_packets_for_test) {
1387    DCHECK_LE(encrypted->length(), kMaxPacketSize);
1388  }
1389  DCHECK_LE(encrypted->length(), packet_generator_.max_packet_length());
1390  DVLOG(1) << ENDPOINT << "Sending packet " << sequence_number << " : "
1391           << (packet->serialized_packet.packet->is_fec_packet() ? "FEC " :
1392               (IsRetransmittable(*packet) == HAS_RETRANSMITTABLE_DATA
1393                ? "data bearing " : " ack only "))
1394           << ", encryption level: "
1395           << QuicUtils::EncryptionLevelToString(packet->encryption_level)
1396           << ", length:"
1397           << packet->serialized_packet.packet->length()
1398           << ", encrypted length:"
1399           << encrypted->length();
1400  DVLOG(2) << ENDPOINT << "packet(" << sequence_number << "): " << std::endl
1401           << QuicUtils::StringToHexASCIIDump(
1402               packet->serialized_packet.packet->AsStringPiece());
1403
1404  WriteResult result = writer_->WritePacket(encrypted->data(),
1405                                            encrypted->length(),
1406                                            self_address().address(),
1407                                            peer_address());
1408  if (result.error_code == ERR_IO_PENDING) {
1409    DCHECK_EQ(WRITE_STATUS_BLOCKED, result.status);
1410  }
1411  if (debug_visitor_.get() != NULL) {
1412    // Pass the write result to the visitor.
1413    debug_visitor_->OnPacketSent(sequence_number,
1414                                 packet->encryption_level,
1415                                 packet->transmission_type,
1416                                 *encrypted,
1417                                 result);
1418  }
1419
1420  if (result.status == WRITE_STATUS_BLOCKED) {
1421    visitor_->OnWriteBlocked();
1422    // If the socket buffers the the data, then the packet should not
1423    // be queued and sent again, which would result in an unnecessary
1424    // duplicate packet being sent.  The helper must call OnCanWrite
1425    // when the write completes, and OnWriteError if an error occurs.
1426    if (!writer_->IsWriteBlockedDataBuffered()) {
1427      return false;
1428    }
1429  }
1430  QuicTime now = clock_->Now();
1431  if (packet->transmission_type == NOT_RETRANSMISSION) {
1432    time_of_last_sent_new_packet_ = now;
1433  }
1434  SetPingAlarm();
1435  DVLOG(1) << ENDPOINT << "time of last sent packet: "
1436           << now.ToDebuggingValue();
1437
1438  // TODO(ianswett): Change the sequence number length and other packet creator
1439  // options by a more explicit API than setting a struct value directly,
1440  // perhaps via the NetworkChangeVisitor.
1441  packet_generator_.UpdateSequenceNumberLength(
1442      sent_packet_manager_.least_packet_awaited_by_peer(),
1443      sent_packet_manager_.GetCongestionWindow());
1444
1445  if (packet->original_sequence_number == 0) {
1446    sent_packet_manager_.OnSerializedPacket(packet->serialized_packet);
1447  } else {
1448    if (debug_visitor_.get() != NULL) {
1449      debug_visitor_->OnPacketRetransmitted(
1450          packet->original_sequence_number, sequence_number);
1451    }
1452    sent_packet_manager_.OnRetransmittedPacket(packet->original_sequence_number,
1453                                               sequence_number);
1454  }
1455  bool reset_retransmission_alarm = sent_packet_manager_.OnPacketSent(
1456      sequence_number,
1457      now,
1458      encrypted->length(),
1459      packet->transmission_type,
1460      IsRetransmittable(*packet));
1461  // The SentPacketManager now owns the retransmittable frames.
1462  packet->serialized_packet.retransmittable_frames = NULL;
1463
1464  if (reset_retransmission_alarm || !retransmission_alarm_->IsSet()) {
1465    retransmission_alarm_->Update(sent_packet_manager_.GetRetransmissionTime(),
1466                                  QuicTime::Delta::FromMilliseconds(1));
1467  }
1468
1469  stats_.bytes_sent += result.bytes_written;
1470  ++stats_.packets_sent;
1471  if (packet->transmission_type != NOT_RETRANSMISSION) {
1472    stats_.bytes_retransmitted += result.bytes_written;
1473    ++stats_.packets_retransmitted;
1474  }
1475
1476  if (result.status == WRITE_STATUS_ERROR) {
1477    OnWriteError(result.error_code);
1478    return false;
1479  }
1480
1481  return true;
1482}
1483
1484bool QuicConnection::ShouldDiscardPacket(const QueuedPacket& packet) {
1485  if (!connected_) {
1486    DVLOG(1) << ENDPOINT
1487             << "Not sending packet as connection is disconnected.";
1488    return true;
1489  }
1490
1491  QuicPacketSequenceNumber sequence_number =
1492      packet.serialized_packet.sequence_number;
1493  if (encryption_level_ == ENCRYPTION_FORWARD_SECURE &&
1494      packet.encryption_level == ENCRYPTION_NONE) {
1495    // Drop packets that are NULL encrypted since the peer won't accept them
1496    // anymore.
1497    DVLOG(1) << ENDPOINT << "Dropping NULL encrypted packet: "
1498             << sequence_number << " since the connection is forward secure.";
1499    return true;
1500  }
1501
1502  // If a retransmission has been acked before sending, don't send it.
1503  // This occurs if a packet gets serialized, queued, then discarded.
1504  if (packet.transmission_type != NOT_RETRANSMISSION &&
1505      (!sent_packet_manager_.IsUnacked(packet.original_sequence_number) ||
1506       !sent_packet_manager_.HasRetransmittableFrames(
1507           packet.original_sequence_number))) {
1508    DVLOG(1) << ENDPOINT << "Dropping unacked packet: " << sequence_number
1509             << " A previous transmission was acked while write blocked.";
1510    return true;
1511  }
1512
1513  return false;
1514}
1515
1516void QuicConnection::OnWriteError(int error_code) {
1517  DVLOG(1) << ENDPOINT << "Write failed with error: " << error_code
1518           << " (" << ErrorToString(error_code) << ")";
1519  // We can't send an error as the socket is presumably borked.
1520  CloseConnection(QUIC_PACKET_WRITE_ERROR, false);
1521}
1522
1523void QuicConnection::OnSerializedPacket(
1524    const SerializedPacket& serialized_packet) {
1525  if (serialized_packet.retransmittable_frames) {
1526    serialized_packet.retransmittable_frames->
1527        set_encryption_level(encryption_level_);
1528  }
1529  SendOrQueuePacket(QueuedPacket(serialized_packet, encryption_level_));
1530}
1531
1532void QuicConnection::OnCongestionWindowChange(QuicByteCount congestion_window) {
1533  packet_generator_.OnCongestionWindowChange(congestion_window);
1534  visitor_->OnCongestionWindowChange(clock_->ApproximateNow());
1535}
1536
1537void QuicConnection::OnHandshakeComplete() {
1538  sent_packet_manager_.SetHandshakeConfirmed();
1539}
1540
1541void QuicConnection::SendOrQueuePacket(QueuedPacket packet) {
1542  // The caller of this function is responsible for checking CanWrite().
1543  if (packet.serialized_packet.packet == NULL) {
1544    LOG(DFATAL) << "NULL packet passed in to SendOrQueuePacket";
1545    return;
1546  }
1547
1548  sent_entropy_manager_.RecordPacketEntropyHash(
1549      packet.serialized_packet.sequence_number,
1550      packet.serialized_packet.entropy_hash);
1551  LOG_IF(DFATAL, !queued_packets_.empty() && !writer_->IsWriteBlocked())
1552      << "Packets should only be left queued if we're write blocked.";
1553  if (!WritePacket(&packet)) {
1554    queued_packets_.push_back(packet);
1555  }
1556}
1557
1558void QuicConnection::UpdateStopWaiting(QuicStopWaitingFrame* stop_waiting) {
1559  stop_waiting->least_unacked = GetLeastUnacked();
1560  stop_waiting->entropy_hash = sent_entropy_manager_.GetCumulativeEntropy(
1561      stop_waiting->least_unacked - 1);
1562}
1563
1564void QuicConnection::SendPing() {
1565  if (retransmission_alarm_->IsSet()) {
1566    return;
1567  }
1568  if (version() == QUIC_VERSION_16) {
1569    // TODO(rch): remove this when we remove version 15 and 16.
1570    // This is a horrible hideous hack which we should not support.
1571    IOVector data;
1572    char c_data[] = "C";
1573    data.Append(c_data, 1);
1574    QuicConsumedData consumed_data =
1575        packet_generator_.ConsumeData(kCryptoStreamId, data, 0, false,
1576                                      MAY_FEC_PROTECT, NULL);
1577    if (consumed_data.bytes_consumed == 0) {
1578      DLOG(ERROR) << "Unable to send ping!?";
1579    }
1580  } else {
1581    packet_generator_.AddControlFrame(QuicFrame(new QuicPingFrame));
1582  }
1583}
1584
1585void QuicConnection::SendAck() {
1586  ack_alarm_->Cancel();
1587  stop_waiting_count_ = 0;
1588  num_packets_received_since_last_ack_sent_ = 0;
1589  bool send_feedback = false;
1590
1591  // Deprecating the Congestion Feedback Frame after QUIC_VERSION_22.
1592  if (version() <= QUIC_VERSION_22) {
1593    if (received_packet_manager_.GenerateCongestionFeedback(
1594            &outgoing_congestion_feedback_)) {
1595      DVLOG(1) << ENDPOINT << "Sending feedback: "
1596               << outgoing_congestion_feedback_;
1597      send_feedback = true;
1598    }
1599  }
1600
1601  packet_generator_.SetShouldSendAck(send_feedback, true);
1602}
1603
1604void QuicConnection::OnRetransmissionTimeout() {
1605  if (!sent_packet_manager_.HasUnackedPackets()) {
1606    return;
1607  }
1608
1609  sent_packet_manager_.OnRetransmissionTimeout();
1610  WriteIfNotBlocked();
1611
1612  // A write failure can result in the connection being closed, don't attempt to
1613  // write further packets, or to set alarms.
1614  if (!connected_) {
1615    return;
1616  }
1617
1618  // In the TLP case, the SentPacketManager gives the connection the opportunity
1619  // to send new data before retransmitting.
1620  if (sent_packet_manager_.MaybeRetransmitTailLossProbe()) {
1621    // Send the pending retransmission now that it's been queued.
1622    WriteIfNotBlocked();
1623  }
1624
1625  // Ensure the retransmission alarm is always set if there are unacked packets
1626  // and nothing waiting to be sent.
1627  if (!HasQueuedData() && !retransmission_alarm_->IsSet()) {
1628    QuicTime rto_timeout = sent_packet_manager_.GetRetransmissionTime();
1629    if (rto_timeout.IsInitialized()) {
1630      retransmission_alarm_->Set(rto_timeout);
1631    }
1632  }
1633}
1634
1635void QuicConnection::SetEncrypter(EncryptionLevel level,
1636                                  QuicEncrypter* encrypter) {
1637  framer_.SetEncrypter(level, encrypter);
1638}
1639
1640const QuicEncrypter* QuicConnection::encrypter(EncryptionLevel level) const {
1641  return framer_.encrypter(level);
1642}
1643
1644void QuicConnection::SetDefaultEncryptionLevel(EncryptionLevel level) {
1645  encryption_level_ = level;
1646  packet_generator_.set_encryption_level(level);
1647}
1648
1649void QuicConnection::SetDecrypter(QuicDecrypter* decrypter,
1650                                  EncryptionLevel level) {
1651  framer_.SetDecrypter(decrypter, level);
1652}
1653
1654void QuicConnection::SetAlternativeDecrypter(QuicDecrypter* decrypter,
1655                                             EncryptionLevel level,
1656                                             bool latch_once_used) {
1657  framer_.SetAlternativeDecrypter(decrypter, level, latch_once_used);
1658}
1659
1660const QuicDecrypter* QuicConnection::decrypter() const {
1661  return framer_.decrypter();
1662}
1663
1664const QuicDecrypter* QuicConnection::alternative_decrypter() const {
1665  return framer_.alternative_decrypter();
1666}
1667
1668void QuicConnection::QueueUndecryptablePacket(
1669    const QuicEncryptedPacket& packet) {
1670  DVLOG(1) << ENDPOINT << "Queueing undecryptable packet.";
1671  undecryptable_packets_.push_back(packet.Clone());
1672}
1673
1674void QuicConnection::MaybeProcessUndecryptablePackets() {
1675  if (undecryptable_packets_.empty() || encryption_level_ == ENCRYPTION_NONE) {
1676    return;
1677  }
1678
1679  while (connected_ && !undecryptable_packets_.empty()) {
1680    DVLOG(1) << ENDPOINT << "Attempting to process undecryptable packet";
1681    QuicEncryptedPacket* packet = undecryptable_packets_.front();
1682    if (!framer_.ProcessPacket(*packet) &&
1683        framer_.error() == QUIC_DECRYPTION_FAILURE) {
1684      DVLOG(1) << ENDPOINT << "Unable to process undecryptable packet...";
1685      break;
1686    }
1687    DVLOG(1) << ENDPOINT << "Processed undecryptable packet!";
1688    ++stats_.packets_processed;
1689    delete packet;
1690    undecryptable_packets_.pop_front();
1691  }
1692
1693  // Once forward secure encryption is in use, there will be no
1694  // new keys installed and hence any undecryptable packets will
1695  // never be able to be decrypted.
1696  if (encryption_level_ == ENCRYPTION_FORWARD_SECURE) {
1697    if (debug_visitor_.get() != NULL) {
1698      // TODO(rtenneti): perhaps more efficient to pass the number of
1699      // undecryptable packets as the argument to OnUndecryptablePacket so that
1700      // we just need to call OnUndecryptablePacket once?
1701      for (size_t i = 0; i < undecryptable_packets_.size(); ++i) {
1702        debug_visitor_->OnUndecryptablePacket();
1703      }
1704    }
1705    STLDeleteElements(&undecryptable_packets_);
1706  }
1707}
1708
1709void QuicConnection::MaybeProcessRevivedPacket() {
1710  QuicFecGroup* group = GetFecGroup();
1711  if (!connected_ || group == NULL || !group->CanRevive()) {
1712    return;
1713  }
1714  QuicPacketHeader revived_header;
1715  char revived_payload[kMaxPacketSize];
1716  size_t len = group->Revive(&revived_header, revived_payload, kMaxPacketSize);
1717  revived_header.public_header.connection_id = connection_id_;
1718  revived_header.public_header.connection_id_length =
1719      last_header_.public_header.connection_id_length;
1720  revived_header.public_header.version_flag = false;
1721  revived_header.public_header.reset_flag = false;
1722  revived_header.public_header.sequence_number_length =
1723      last_header_.public_header.sequence_number_length;
1724  revived_header.fec_flag = false;
1725  revived_header.is_in_fec_group = NOT_IN_FEC_GROUP;
1726  revived_header.fec_group = 0;
1727  group_map_.erase(last_header_.fec_group);
1728  last_decrypted_packet_level_ = group->effective_encryption_level();
1729  DCHECK_LT(last_decrypted_packet_level_, NUM_ENCRYPTION_LEVELS);
1730  delete group;
1731
1732  last_packet_revived_ = true;
1733  if (debug_visitor_.get() != NULL) {
1734    debug_visitor_->OnRevivedPacket(revived_header,
1735                                    StringPiece(revived_payload, len));
1736  }
1737
1738  ++stats_.packets_revived;
1739  framer_.ProcessRevivedPacket(&revived_header,
1740                               StringPiece(revived_payload, len));
1741}
1742
1743QuicFecGroup* QuicConnection::GetFecGroup() {
1744  QuicFecGroupNumber fec_group_num = last_header_.fec_group;
1745  if (fec_group_num == 0) {
1746    return NULL;
1747  }
1748  if (group_map_.count(fec_group_num) == 0) {
1749    if (group_map_.size() >= kMaxFecGroups) {  // Too many groups
1750      if (fec_group_num < group_map_.begin()->first) {
1751        // The group being requested is a group we've seen before and deleted.
1752        // Don't recreate it.
1753        return NULL;
1754      }
1755      // Clear the lowest group number.
1756      delete group_map_.begin()->second;
1757      group_map_.erase(group_map_.begin());
1758    }
1759    group_map_[fec_group_num] = new QuicFecGroup();
1760  }
1761  return group_map_[fec_group_num];
1762}
1763
1764void QuicConnection::SendConnectionClose(QuicErrorCode error) {
1765  SendConnectionCloseWithDetails(error, string());
1766}
1767
1768void QuicConnection::SendConnectionCloseWithDetails(QuicErrorCode error,
1769                                                    const string& details) {
1770  // If we're write blocked, WritePacket() will not send, but will capture the
1771  // serialized packet.
1772  SendConnectionClosePacket(error, details);
1773  if (connected_) {
1774    // It's possible that while sending the connection close packet, we get a
1775    // socket error and disconnect right then and there.  Avoid a double
1776    // disconnect in that case.
1777    CloseConnection(error, false);
1778  }
1779}
1780
1781void QuicConnection::SendConnectionClosePacket(QuicErrorCode error,
1782                                               const string& details) {
1783  DVLOG(1) << ENDPOINT << "Force closing " << connection_id()
1784           << " with error " << QuicUtils::ErrorToString(error)
1785           << " (" << error << ") " << details;
1786  ScopedPacketBundler ack_bundler(this, SEND_ACK);
1787  QuicConnectionCloseFrame* frame = new QuicConnectionCloseFrame();
1788  frame->error_code = error;
1789  frame->error_details = details;
1790  packet_generator_.AddControlFrame(QuicFrame(frame));
1791  packet_generator_.FlushAllQueuedFrames();
1792}
1793
1794void QuicConnection::CloseConnection(QuicErrorCode error, bool from_peer) {
1795  if (!connected_) {
1796    DLOG(DFATAL) << "Error: attempt to close an already closed connection"
1797                 << base::debug::StackTrace().ToString();
1798    return;
1799  }
1800  connected_ = false;
1801  if (debug_visitor_.get() != NULL) {
1802    debug_visitor_->OnConnectionClosed(error, from_peer);
1803  }
1804  visitor_->OnConnectionClosed(error, from_peer);
1805  // Cancel the alarms so they don't trigger any action now that the
1806  // connection is closed.
1807  ack_alarm_->Cancel();
1808  ping_alarm_->Cancel();
1809  resume_writes_alarm_->Cancel();
1810  retransmission_alarm_->Cancel();
1811  send_alarm_->Cancel();
1812  timeout_alarm_->Cancel();
1813}
1814
1815void QuicConnection::SendGoAway(QuicErrorCode error,
1816                                QuicStreamId last_good_stream_id,
1817                                const string& reason) {
1818  DVLOG(1) << ENDPOINT << "Going away with error "
1819           << QuicUtils::ErrorToString(error)
1820           << " (" << error << ")";
1821
1822  // Opportunistically bundle an ack with this outgoing packet.
1823  ScopedPacketBundler ack_bundler(this, BUNDLE_PENDING_ACK);
1824  packet_generator_.AddControlFrame(
1825      QuicFrame(new QuicGoAwayFrame(error, last_good_stream_id, reason)));
1826}
1827
1828void QuicConnection::CloseFecGroupsBefore(
1829    QuicPacketSequenceNumber sequence_number) {
1830  FecGroupMap::iterator it = group_map_.begin();
1831  while (it != group_map_.end()) {
1832    // If this is the current group or the group doesn't protect this packet
1833    // we can ignore it.
1834    if (last_header_.fec_group == it->first ||
1835        !it->second->ProtectsPacketsBefore(sequence_number)) {
1836      ++it;
1837      continue;
1838    }
1839    QuicFecGroup* fec_group = it->second;
1840    DCHECK(!fec_group->CanRevive());
1841    FecGroupMap::iterator next = it;
1842    ++next;
1843    group_map_.erase(it);
1844    delete fec_group;
1845    it = next;
1846  }
1847}
1848
1849size_t QuicConnection::max_packet_length() const {
1850  return packet_generator_.max_packet_length();
1851}
1852
1853void QuicConnection::set_max_packet_length(size_t length) {
1854  return packet_generator_.set_max_packet_length(length);
1855}
1856
1857bool QuicConnection::HasQueuedData() const {
1858  return pending_version_negotiation_packet_ ||
1859      !queued_packets_.empty() || packet_generator_.HasQueuedFrames();
1860}
1861
1862bool QuicConnection::CanWriteStreamData() {
1863  // Don't write stream data if there are negotiation or queued data packets
1864  // to send. Otherwise, continue and bundle as many frames as possible.
1865  if (pending_version_negotiation_packet_ || !queued_packets_.empty()) {
1866    return false;
1867  }
1868
1869  IsHandshake pending_handshake = visitor_->HasPendingHandshake() ?
1870      IS_HANDSHAKE : NOT_HANDSHAKE;
1871  // Sending queued packets may have caused the socket to become write blocked,
1872  // or the congestion manager to prohibit sending.  If we've sent everything
1873  // we had queued and we're still not blocked, let the visitor know it can
1874  // write more.
1875  return ShouldGeneratePacket(NOT_RETRANSMISSION, HAS_RETRANSMITTABLE_DATA,
1876                              pending_handshake);
1877}
1878
1879void QuicConnection::SetIdleNetworkTimeout(QuicTime::Delta timeout) {
1880  // Adjust the idle timeout on client and server to prevent clients from
1881  // sending requests to servers which have already closed the connection.
1882  if (is_server_) {
1883    timeout = timeout.Add(QuicTime::Delta::FromSeconds(1));
1884  } else if (timeout > QuicTime::Delta::FromSeconds(1)) {
1885    timeout = timeout.Subtract(QuicTime::Delta::FromSeconds(1));
1886  }
1887
1888  if (timeout < idle_network_timeout_) {
1889    idle_network_timeout_ = timeout;
1890    CheckForTimeout();
1891  } else {
1892    idle_network_timeout_ = timeout;
1893  }
1894}
1895
1896void QuicConnection::SetOverallConnectionTimeout(QuicTime::Delta timeout) {
1897  if (timeout < overall_connection_timeout_) {
1898    overall_connection_timeout_ = timeout;
1899    CheckForTimeout();
1900  } else {
1901    overall_connection_timeout_ = timeout;
1902  }
1903}
1904
1905bool QuicConnection::CheckForTimeout() {
1906  QuicTime now = clock_->ApproximateNow();
1907  QuicTime time_of_last_packet = max(time_of_last_received_packet_,
1908                                     time_of_last_sent_new_packet_);
1909
1910  // If no packets have been sent or received, then don't timeout.
1911  if (FLAGS_quic_timeouts_require_activity &&
1912      !time_of_last_packet.IsInitialized()) {
1913    timeout_alarm_->Cancel();
1914    timeout_alarm_->Set(now.Add(idle_network_timeout_));
1915    return false;
1916  }
1917
1918  // |delta| can be < 0 as |now| is approximate time but |time_of_last_packet|
1919  // is accurate time. However, this should not change the behavior of
1920  // timeout handling.
1921  QuicTime::Delta delta = now.Subtract(time_of_last_packet);
1922  DVLOG(1) << ENDPOINT << "last packet "
1923           << time_of_last_packet.ToDebuggingValue()
1924           << " now:" << now.ToDebuggingValue()
1925           << " delta:" << delta.ToMicroseconds()
1926           << " network_timeout: " << idle_network_timeout_.ToMicroseconds();
1927  if (delta >= idle_network_timeout_) {
1928    DVLOG(1) << ENDPOINT << "Connection timedout due to no network activity.";
1929    SendConnectionClose(QUIC_CONNECTION_TIMED_OUT);
1930    return true;
1931  }
1932
1933  // Next timeout delta.
1934  QuicTime::Delta timeout = idle_network_timeout_.Subtract(delta);
1935
1936  if (!overall_connection_timeout_.IsInfinite()) {
1937    QuicTime::Delta connected_time =
1938        now.Subtract(stats_.connection_creation_time);
1939    DVLOG(1) << ENDPOINT << "connection time: "
1940             << connected_time.ToMilliseconds() << " overall timeout: "
1941             << overall_connection_timeout_.ToMilliseconds();
1942    if (connected_time >= overall_connection_timeout_) {
1943      DVLOG(1) << ENDPOINT <<
1944          "Connection timedout due to overall connection timeout.";
1945      SendConnectionClose(QUIC_CONNECTION_OVERALL_TIMED_OUT);
1946      return true;
1947    }
1948
1949    // Take the min timeout.
1950    QuicTime::Delta connection_timeout =
1951        overall_connection_timeout_.Subtract(connected_time);
1952    if (connection_timeout < timeout) {
1953      timeout = connection_timeout;
1954    }
1955  }
1956
1957  timeout_alarm_->Cancel();
1958  timeout_alarm_->Set(now.Add(timeout));
1959  return false;
1960}
1961
1962void QuicConnection::SetPingAlarm() {
1963  if (is_server_) {
1964    // Only clients send pings.
1965    return;
1966  }
1967  if (!visitor_->HasOpenDataStreams()) {
1968    ping_alarm_->Cancel();
1969    // Don't send a ping unless there are open streams.
1970    return;
1971  }
1972  QuicTime::Delta ping_timeout = QuicTime::Delta::FromSeconds(kPingTimeoutSecs);
1973  ping_alarm_->Update(clock_->ApproximateNow().Add(ping_timeout),
1974                      QuicTime::Delta::FromSeconds(1));
1975}
1976
1977QuicConnection::ScopedPacketBundler::ScopedPacketBundler(
1978    QuicConnection* connection,
1979    AckBundling send_ack)
1980    : connection_(connection),
1981      already_in_batch_mode_(connection != NULL &&
1982                             connection->packet_generator_.InBatchMode()) {
1983  if (connection_  == NULL) {
1984    return;
1985  }
1986  // Move generator into batch mode. If caller wants us to include an ack,
1987  // check the delayed-ack timer to see if there's ack info to be sent.
1988  if (!already_in_batch_mode_) {
1989    DVLOG(1) << "Entering Batch Mode.";
1990    connection_->packet_generator_.StartBatchOperations();
1991  }
1992  // Bundle an ack if the alarm is set or with every second packet if we need to
1993  // raise the peer's least unacked.
1994  bool ack_pending =
1995      connection_->ack_alarm_->IsSet() || connection_->stop_waiting_count_ > 1;
1996  if (send_ack == SEND_ACK || (send_ack == BUNDLE_PENDING_ACK && ack_pending)) {
1997    DVLOG(1) << "Bundling ack with outgoing packet.";
1998    connection_->SendAck();
1999  }
2000}
2001
2002QuicConnection::ScopedPacketBundler::~ScopedPacketBundler() {
2003  if (connection_  == NULL) {
2004    return;
2005  }
2006  // If we changed the generator's batch state, restore original batch state.
2007  if (!already_in_batch_mode_) {
2008    DVLOG(1) << "Leaving Batch Mode.";
2009    connection_->packet_generator_.FinishBatchOperations();
2010  }
2011  DCHECK_EQ(already_in_batch_mode_,
2012            connection_->packet_generator_.InBatchMode());
2013}
2014
2015HasRetransmittableData QuicConnection::IsRetransmittable(
2016    const QueuedPacket& packet) {
2017  // Retransmitted packets retransmittable frames are owned by the unacked
2018  // packet map, but are not present in the serialized packet.
2019  if (packet.transmission_type != NOT_RETRANSMISSION ||
2020      packet.serialized_packet.retransmittable_frames != NULL) {
2021    return HAS_RETRANSMITTABLE_DATA;
2022  } else {
2023    return NO_RETRANSMITTABLE_DATA;
2024  }
2025}
2026
2027bool QuicConnection::IsConnectionClose(
2028    QueuedPacket packet) {
2029  RetransmittableFrames* retransmittable_frames =
2030      packet.serialized_packet.retransmittable_frames;
2031  if (!retransmittable_frames) {
2032    return false;
2033  }
2034  for (size_t i = 0; i < retransmittable_frames->frames().size(); ++i) {
2035    if (retransmittable_frames->frames()[i].type == CONNECTION_CLOSE_FRAME) {
2036      return true;
2037    }
2038  }
2039  return false;
2040}
2041
2042}  // namespace net
2043