Send PING to check the status of the SPDY connection.

net/spdy/spdy_session.cc

 // Copyright (c) 2011 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/spdy/spdy_session.h"
 
 #include "base/basictypes.h"
 #include "base/logging.h"
 #include "base/memory/linked_ptr.h"
 #include "base/message_loop.h"
@@ skipping 154 matching lines @@
   }
 
  private:
   ~NetLogSpdyRstParameter() {}
   const spdy::SpdyStreamId stream_id_;
   const int status_;
 
   DISALLOW_COPY_AND_ASSIGN(NetLogSpdyRstParameter);
 };
 
+class NetLogSpdyPingParameter : public NetLog::EventParameters {
+ public:
+  explicit NetLogSpdyPingParameter(uint32 unique_id) : unique_id_(unique_id) {}
+
+  virtual Value* ToValue() const {
+    DictionaryValue* dict = new DictionaryValue();
+    dict->SetInteger("unique_id", unique_id_);
+    return dict;
+  }
+
+ private:
+  ~NetLogSpdyPingParameter() {}
+  const uint32 unique_id_;
+
+  DISALLOW_COPY_AND_ASSIGN(NetLogSpdyPingParameter);
+};
+
 class NetLogSpdyGoAwayParameter : public NetLog::EventParameters {
  public:
   NetLogSpdyGoAwayParameter(spdy::SpdyStreamId last_stream_id,
                             int active_streams,
                             int unclaimed_streams)
       : last_stream_id_(last_stream_id),
         active_streams_(active_streams),
         unclaimed_streams_(unclaimed_streams) {}
 
   virtual Value* ToValue() const {
@@ skipping 21 matching lines @@
 
 // static
 bool SpdySession::use_flow_control_ = false;
 
 // static
 size_t SpdySession::init_max_concurrent_streams_ = 10;
 
 // static
 size_t SpdySession::max_concurrent_stream_limit_ = 256;
 
+// static
+bool SpdySession::enable_ping_based_connection_checking_ = true;
+
+// static
+int SpdySession::connection_at_risk_of_loss_ms_ = 0;
+
+// static
+int SpdySession::trailing_ping_delay_time_ms_ = 1000;
+
+// static
+int SpdySession::hung_interval_ms_ = 10000;
+
 SpdySession::SpdySession(const HostPortProxyPair& host_port_proxy_pair,
                          SpdySessionPool* spdy_session_pool,
                          SpdySettingsStorage* spdy_settings,
                          bool verify_domain_authentication,
                          NetLog* net_log)
     : ALLOW_THIS_IN_INITIALIZER_LIST(
           read_callback_(this, &SpdySession::OnReadComplete)),
       ALLOW_THIS_IN_INITIALIZER_LIST(
           write_callback_(this, &SpdySession::OnWriteComplete)),
       ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)),
@@ skipping 13 matching lines @@
       max_concurrent_streams_(init_max_concurrent_streams_),
       streams_initiated_count_(0),
       streams_pushed_count_(0),
       streams_pushed_and_claimed_count_(0),
       streams_abandoned_count_(0),
       frames_received_(0),
       bytes_received_(0),
       sent_settings_(false),
       received_settings_(false),
       stalled_streams_(0),
+      pings_in_flight_(0),
+      next_ping_id_(1),
+      received_data_time_(base::TimeTicks::Now()),
+      trailing_ping_pending_(false),
+      check_ping_status_pending_(false),
+      last_sent_was_ping_(false),
       initial_send_window_size_(spdy::kSpdyStreamInitialWindowSize),
       initial_recv_window_size_(spdy::kSpdyStreamInitialWindowSize),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)),
       verify_domain_authentication_(verify_domain_authentication) {
   DCHECK(HttpStreamFactory::spdy_enabled());
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       make_scoped_refptr(
           new NetLogSpdySessionParameter(host_port_proxy_pair_)));
 
@@ skipping 208 matching lines @@
     spdy::SpdyStreamId stream_id,
     RequestPriority priority,
     spdy::SpdyControlFlags flags,
     const linked_ptr<spdy::SpdyHeaderBlock>& headers) {
   // Find our stream
   if (!IsStreamActive(stream_id))
     return ERR_INVALID_SPDY_STREAM;
   const scoped_refptr<SpdyStream>& stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
 
+  SendPrefacePingIfNoneInFlight();
+
   scoped_ptr<spdy::SpdySynStreamControlFrame> syn_frame(
       spdy_framer_.CreateSynStream(
           stream_id, 0,
           ConvertRequestPriorityToSpdyPriority(priority),
           flags, false, headers.get()));
   QueueFrame(syn_frame.get(), priority, stream);
 
   base::StatsCounter spdy_requests("spdy.requests");
   spdy_requests.Increment();
   streams_initiated_count_++;
 
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SYN_STREAM,
         make_scoped_refptr(
             new NetLogSpdySynParameter(headers, flags, stream_id, 0)));
   }
+  last_sent_was_ping_ = false;
 
   return ERR_IO_PENDING;
 }
 
 int SpdySession::WriteStreamData(spdy::SpdyStreamId stream_id,
                                  net::IOBuffer* data, int len,
                                  spdy::SpdyDataFlags flags) {
   // Find our stream
   DCHECK(IsStreamActive(stream_id));
   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
   if (!stream)
     return ERR_INVALID_SPDY_STREAM;
 
+  SendPrefacePingIfNoneInFlight();
+
   if (len > kMaxSpdyFrameChunkSize) {
     len = kMaxSpdyFrameChunkSize;
     flags = static_cast<spdy::SpdyDataFlags>(flags & ~spdy::DATA_FLAG_FIN);
   }
 
   // Obey send window size of the stream if flow control is enabled.
   if (use_flow_control_) {
     if (stream->send_window_size() <= 0) {
       // Because we queue frames onto the session, it is possible that
       // a stream was not flow controlled at the time it attempted the
@@ skipping 19 matching lines @@
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SEND_DATA,
         make_scoped_refptr(new NetLogSpdyDataParameter(stream_id, len, flags)));
   }
 
   // TODO(mbelshe): reduce memory copies here.
   scoped_ptr<spdy::SpdyDataFrame> frame(
       spdy_framer_.CreateDataFrame(stream_id, data->data(), len, flags));
   QueueFrame(frame.get(), stream->priority(), stream);
+  last_sent_was_ping_ = false;
   return ERR_IO_PENDING;
 }
 
 void SpdySession::CloseStream(spdy::SpdyStreamId stream_id, int status) {
   // TODO(mbelshe): We should send a RST_STREAM control frame here
   //                so that the server can cancel a large send.
 
   DeleteStream(stream_id, status);
 }
 
 void SpdySession::ResetStream(
     spdy::SpdyStreamId stream_id, spdy::SpdyStatusCodes status) {
 
   net_log().AddEvent(
       NetLog::TYPE_SPDY_SESSION_SEND_RST_STREAM,
       make_scoped_refptr(new NetLogSpdyRstParameter(stream_id, status)));
 
   scoped_ptr<spdy::SpdyRstStreamControlFrame> rst_frame(
       spdy_framer_.CreateRstStream(stream_id, status));
 
   // Default to lowest priority unless we know otherwise.
   int priority = 3;
   if(IsStreamActive(stream_id)) {
     scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
     priority = stream->priority();
   }
   QueueFrame(rst_frame.get(), priority, NULL);
-
+  last_sent_was_ping_ = false;
   DeleteStream(stream_id, ERR_SPDY_PROTOCOL_ERROR);
 }
 
 bool SpdySession::IsStreamActive(spdy::SpdyStreamId stream_id) const {
   return ContainsKey(active_streams_, stream_id);
 }
 
 LoadState SpdySession::GetLoadState() const {
   // NOTE: The application only queries the LoadState via the
   //       SpdyNetworkTransaction, and details are only needed when
@@ skipping 20 matching lines @@
     // Session is tearing down.
     net::Error error = static_cast<net::Error>(bytes_read);
     if (bytes_read == 0)
       error = ERR_CONNECTION_CLOSED;
     CloseSessionOnError(error, true);
     return;
   }
 
   bytes_received_ += bytes_read;
 
+  received_data_time_ = base::TimeTicks::Now();
+
   // The SpdyFramer will use callbacks onto |this| as it parses frames.
   // When errors occur, those callbacks can lead to teardown of all references
   // to |this|, so maintain a reference to self during this call for safe
   // cleanup.
   scoped_refptr<SpdySession> self(this);
 
   char *data = read_buffer_->data();
   while (bytes_read &&
          spdy_framer_.error_code() == spdy::SpdyFramer::SPDY_NO_ERROR) {
     uint32 bytes_processed = spdy_framer_.ProcessInput(data, bytes_read);
@@ skipping 597 matching lines @@
       return;
     }
   }
 
   frames_received_++;
 
   switch (type) {
     case spdy::GOAWAY:
       OnGoAway(*reinterpret_cast<const spdy::SpdyGoAwayControlFrame*>(frame));
       break;
+    case spdy::PING:
+      OnPing(*reinterpret_cast<const spdy::SpdyPingControlFrame*>(frame));
+      break;
     case spdy::SETTINGS:
       OnSettings(
           *reinterpret_cast<const spdy::SpdySettingsControlFrame*>(frame));
       break;
     case spdy::RST_STREAM:
       OnRst(*reinterpret_cast<const spdy::SpdyRstStreamControlFrame*>(frame));
       break;
     case spdy::SYN_STREAM:
       OnSyn(*reinterpret_cast<const spdy::SpdySynStreamControlFrame*>(frame),
             headers);
@@ skipping 68 matching lines @@
   CloseAllStreams(net::ERR_ABORTED);
 
   // TODO(willchan): Cancel any streams that are past the GoAway frame's
   // |last_accepted_stream_id|.
 
   // Don't bother killing any streams that are still reading.  They'll either
   // complete successfully or get an ERR_CONNECTION_CLOSED when the socket is
   // closed.
 }
 
+void SpdySession::OnPing(const spdy::SpdyPingControlFrame& frame) {
+  net_log_.AddEvent(
+      NetLog::TYPE_SPDY_SESSION_PING,
+      make_scoped_refptr(new NetLogSpdyPingParameter(frame.unique_id())));
+
+  // Send response to a PING from server.
+  if (frame.unique_id() % 2 == 0) {
+    WritePingFrame(frame.unique_id());
+    return;
+  }
+
+  --pings_in_flight_;
+  if (pings_in_flight_ < 0) {
+    CloseSessionOnError(net::ERR_SPDY_PROTOCOL_ERROR, true);
+    return;
+  }
+
+  if (pings_in_flight_ > 0)
+    return;
+
+  if (last_sent_was_ping_)
+    return;
+
+  PlanToSendTrailingPing();
+}
+
 void SpdySession::OnSettings(const spdy::SpdySettingsControlFrame& frame) {
   spdy::SpdySettings settings;
   if (spdy_framer_.ParseSettings(&frame, &settings)) {
     HandleSettings(settings);
     spdy_settings_->Set(host_port_pair(), settings);
   }
 
   received_settings_ = true;
 
   net_log_.AddEvent(
@@ skipping 37 matching lines @@
   CHECK_EQ(stream->stream_id(), stream_id);
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_RECV_WINDOW_UPDATE,
       make_scoped_refptr(new NetLogSpdyWindowUpdateParameter(
           stream_id, delta_window_size, stream->recv_window_size())));
 
   scoped_ptr<spdy::SpdyWindowUpdateControlFrame> window_update_frame(
       spdy_framer_.CreateWindowUpdate(stream_id, delta_window_size));
   QueueFrame(window_update_frame.get(), stream->priority(), stream);
+  last_sent_was_ping_ = false;
 }
 
 // Given a cwnd that we would have sent to the server, modify it based on the
 // field trial policy.
 uint32 ApplyCwndFieldTrialPolicy(int cwnd) {
   base::FieldTrial* trial = base::FieldTrialList::Find("SpdyCwnd");
   if (!trial) {
       LOG(WARNING) << "Could not find \"SpdyCwnd\" in FieldTrialList";
       return cwnd;
   }
@@ skipping 44 matching lines @@
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_SEND_SETTINGS,
       make_scoped_refptr(new NetLogSpdySettingsParameter(settings)));
 
   // Create the SETTINGS frame and send it.
   scoped_ptr<spdy::SpdySettingsControlFrame> settings_frame(
       spdy_framer_.CreateSettings(settings));
   sent_settings_ = true;
   QueueFrame(settings_frame.get(), 0, NULL);
+  last_sent_was_ping_ = false;
 }
 
 void SpdySession::HandleSettings(const spdy::SpdySettings& settings) {
   for (spdy::SpdySettings::const_iterator i = settings.begin(),
            end = settings.end(); i != end; ++i) {
     const uint32 id = i->first.id();
     const uint32 val = i->second;
     switch (id) {
       case spdy::SETTINGS_MAX_CONCURRENT_STREAMS:
         max_concurrent_streams_ = std::min(static_cast<size_t>(val),
                                            max_concurrent_stream_limit_);
         ProcessPendingCreateStreams();
         break;
     }
   }
 }
 
+void SpdySession::SendPrefacePingIfNoneInFlight() {
+  if (pings_in_flight_ || trailing_ping_pending_ ||
+      !enable_ping_based_connection_checking_)
+    return;
+
+  const base::TimeDelta kConnectionAtRiskOfLoss =
+      base::TimeDelta::FromMilliseconds(connection_at_risk_of_loss_ms_);
+
+  base::TimeTicks now = base::TimeTicks::Now();
+  // If we haven't heard from server, then send a preface-PING.
+  if ((now - received_data_time_) > kConnectionAtRiskOfLoss)
+    SendPrefacePing();
+
+  PlanToSendTrailingPing();
+}
+
+void SpdySession::SendPrefacePing() {
+  WritePingFrame(next_ping_id_);
+}
+
+void SpdySession::PlanToSendTrailingPing() {
+  if (trailing_ping_pending_)
+    return;
+
+  trailing_ping_pending_ = true;
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(&SpdySession::SendTrailingPing),
+      trailing_ping_delay_time_ms_);
+}
+
+void SpdySession::SendTrailingPing() {
+  DCHECK(trailing_ping_pending_);
+  trailing_ping_pending_ = false;
+  WritePingFrame(next_ping_id_);
+}
+
+void SpdySession::WritePingFrame(uint32 unique_id) {
+  scoped_ptr<spdy::SpdyPingControlFrame> ping_frame(
+      spdy_framer_.CreatePingFrame(next_ping_id_));
+  QueueFrame(ping_frame.get(), SPDY_PRIORITY_HIGHEST, NULL);
+
+  if (net_log().IsLoggingAllEvents()) {
+    net_log().AddEvent(
+        NetLog::TYPE_SPDY_SESSION_PING,
+        make_scoped_refptr(new NetLogSpdyPingParameter(next_ping_id_)));
+  }
+  if (unique_id % 2 != 0) {
+    next_ping_id_ += 2;
+    ++pings_in_flight_;
+    last_sent_was_ping_ = true;
+    PlanToCheckPingStatus();
+  }
+}
+
+void SpdySession::PlanToCheckPingStatus() {
+  if (check_ping_status_pending_)
+    return;
+
+  check_ping_status_pending_ = true;
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &SpdySession::CheckPingStatus, base::TimeTicks::Now()),
+      hung_interval_ms_);
+}
+
+void SpdySession::CheckPingStatus(base::TimeTicks last_check_time) {
+  // Check if we got a response back for all PINGs we had sent.
+  if (pings_in_flight_ == 0) {
+    check_ping_status_pending_ = false;
+    return;
+  }
+
+  DCHECK(check_ping_status_pending_);
+
+  const base::TimeDelta kHungInterval =
+      base::TimeDelta::FromMilliseconds(hung_interval_ms_);
+
+  base::TimeTicks now = base::TimeTicks::Now();
+  base::TimeDelta delay = kHungInterval - (now - received_data_time_);
+
+  if (delay.InMilliseconds() < 0 || received_data_time_ < last_check_time) {
+    DCHECK(now - received_data_time_ > kHungInterval);
+    CloseSessionOnError(net::ERR_SPDY_PING_FAILED, true);
+    return;
+  }
+
+  // Check the status of connection after a delay.
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(&SpdySession::CheckPingStatus, now),
+      delay.InMilliseconds());
+}
+
 void SpdySession::RecordHistograms() {
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPerSession",
                               streams_initiated_count_,
                               0, 300, 50);
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPushedPerSession",
                               streams_pushed_count_,
                               0, 300, 50);
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPushedAndClaimedPerSession",
                               streams_pushed_and_claimed_count_,
                               0, 300, 50);
@@ skipping 68 matching lines @@
   if (it == pending_callback_map_.end())
     return;
 
   OldCompletionCallback* callback = it->second.callback;
   int result = it->second.result;
   pending_callback_map_.erase(it);
   callback->Run(result);
 }
 
 }  // namespace net