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/auto_reset.h"
 #include "base/basictypes.h"
 #include "base/logging.h"
 #include "base/memory/linked_ptr.h"
 #include "base/message_loop.h"
 #include "base/metrics/field_trial.h"
 #include "base/metrics/stats_counters.h"
 #include "base/stl_util.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
 #include "base/stringprintf.h"
@@ skipping 148 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() {}

[jar (doing other things), 2011/10/14 19:59:07]

Question for WillChan: Does making this private preclude every destroying an
instance?  (since this class doesn't do the deed).  Does this me we (by design)
leak all such created instances?

[willchan no longer on Chromium, 2011/10/14 22:49:21]

Refcounted

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+  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::send_ping_for_every_request_ = false;
+
+// static
+int SpdySession::post_ping_delay_time_ms_ = 1000;
+
+// static
+int SpdySession::check_status_delay_time_ms_ = 10000;
+
+// 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),
+      unique_id_counter_(1),
+      post_ping_pending_(false),
+      check_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) {

[jar (doing other things), 2011/10/14 19:59:07]

Since we just got data over the underlying TCP/IP connection to set this up....
please initialize received_data_ to TimeTicks::Now().

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

   DCHECK(HttpStreamFactory::spdy_enabled());
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       make_scoped_refptr(
           new NetLogSpdySessionParameter(host_port_proxy_pair_)));
 
   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 
   spdy_framer_.set_visitor(this);
 
@@ skipping 204 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);
 
+  SendPing();

[jar (doing other things), 2011/10/14 19:59:07]

We need a better name, since this confused me into thinking ew always send a
ping in front of every message.  Perhaps:

SendPrefacePingIfNoneInFlight()

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+
   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

[jar (doing other things), 2011/10/14 19:59:07]

We should probably (just for safety and correctness) call:

SendPrefacePingIfNoneInFlight()

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

   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;
 
   if (len > kMaxSpdyFrameChunkSize) {
     len = kMaxSpdyFrameChunkSize;
     flags = static_cast<spdy::SpdyDataFlags>(flags & ~spdy::DATA_FLAG_FIN);
   }
@@ skipping 25 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())));

[jar (doing other things), 2011/10/14 19:59:07]

We probably need to condition this on logging... especially if the new'ed item
would leak otherwise.

WillChan: What am I missing??

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+
+  // Send reponse to a PING from Server.
+  if (frame.unique_id() % 2 == 0) {
+    WritePingFrame();

[jar (doing other things), 2011/10/14 19:59:07]

You need to pass the frame.unique_id

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+    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;
+
+  PlanToSendPostPing();
+}
+
 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;

[willchan no longer on Chromium, 2011/10/14 22:49:21]

Why don't we just call this in QueueFrame()? You can identify the frame type
there.

[ramant (doing other things), 2011/10/14 23:42:45]

We don't want to set last_sent_was_ping_ to true when we send ping response for
the ping request from server. Wasn't sure how to differentiate if a ping frame
is client generated vs a response to server.

 }
 
 // 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::SendPing() {
+  if (pings_in_flight_ || post_ping_pending_ || !send_ping_for_every_request_)

[willchan no longer on Chromium, 2011/10/14 22:49:21]

So, if we are sending a ping for every request, how does that get sent out?

[ramant (doing other things), 2011/10/14 23:42:45]

The variable name is wrong. We don't send ping for every request.

Done.

+    return;
+
+  const base::TimeDelta kInterval =
+      base::TimeDelta::FromMilliseconds(hung_interval_ms_);

[jar (doing other things), 2011/10/14 19:59:07]

This needs a different timer name.  It is really meant to be an optimization to
avoid sending wasteful preface pings (when we just got some data). Perhaps we
could call it:

connection_at_risk_of_loss_ms


If it is zero (the most conservative figure), then we always send the preface
ping (when none are in flight).

It is common for TCP/IP sessions to time out in about 3-5 minutes.  Certainly if
it has been more than 3 minutes, we do want to send a preface ping.

I don't think any connection will time out in under about 10 seconds.  So this
might as well be set to something conservative like 10 seconds.  Later, we could
adjust it to send fewer pings perhaps.  

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+
+  base::TimeTicks now = base::TimeTicks::Now();
+  // If we haven't heard from server, then send a pre-PING.
+  if ((now - received_data_time_) > kInterval)
+    SendPrePing();
+
+  PlanToSendPostPing();
+}
+
+void SpdySession::SendPrePing() {
+  AutoReset<bool> reset(&delayed_write_pending_, true);

[willchan no longer on Chromium, 2011/10/14 22:49:21]

Please explain how this works. How does this coalesce the PING and SYN_STREAM
into the same packet? The SpdySession::WriteSocket() function is still only
calling Socket::Write() with a buffer for a single frame.

[ramant (doing other things), 2011/10/14 23:42:45]

Deleted the AutoReset.  You are right, we are writing into socket for every
frame.

+  WritePingFrame();
+  PlanToCheckStatus();
+  ++pings_in_flight_;

[jar (doing other things), 2011/10/14 19:59:07]

You should pass the next_ping_id_ to WritePingFrame.  Then you can move
++pings_in_flight_ into WritePingFrame(n), and condition it on incrementing only
if we have an odd number.

Similarly, last_sent_was_ping_ can be set in that function (when it is an odd
number).

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+  last_sent_was_ping_ = true;
+}
+
+void SpdySession::SendPostPing() {
+  DCHECK(post_ping_pending_);
+  post_ping_pending_ = false;
+  WritePingFrame();
+  PlanToCheckStatus();
+  ++pings_in_flight_;
+  last_sent_was_ping_ = true;
+}
+
+void SpdySession::PlanToSendPostPing() {
+  if (post_ping_pending_)
+    return;
+
+  post_ping_pending_ = true;
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(&SpdySession::SendPostPing),
+      post_ping_delay_time_ms_);
+}
+
+void SpdySession::PlanToCheckStatus() {
+  if (check_status_pending_)
+    return;
+
+  check_status_pending_ = true;
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &SpdySession::CheckStatus, base::TimeTicks::Now()),
+      check_status_delay_time_ms_);

[jar (doing other things), 2011/10/14 19:59:07]

This should be our hung_interval_ms (or newer name)

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+}
+
+void SpdySession::WritePingFrame() {
+  scoped_ptr<spdy::SpdyPingControlFrame> ping_frame(
+      spdy_framer_.CreatePingFrame(unique_id_counter_));
+  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(unique_id_counter_)));
+  }
+  unique_id_counter_ += 2;

[jar (doing other things), 2011/10/14 19:59:07]

This is only done if we were sending (an odd number) which will come as an
argument.

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+}
+
+void SpdySession::CheckStatus(base::TimeTicks last_check_time) {
+  // Check if we got a response back for all PINGs we had sent.
+  if (pings_in_flight_ == 0) {
+    check_status_pending_ = false;
+    return;
+  }
+
+  DCHECK(check_status_pending_);
+
+  const base::TimeDelta kHungInterval =
+      base::TimeDelta::FromMilliseconds(hung_interval_ms_);
+  if (received_data_time_ < last_check_time) {
+    DCHECK(base::TimeTicks::Now() - received_data_time_ > kHungInterval);

[jar (doing other things), 2011/10/14 19:59:07]

I think this should be >= kHungInterval

[willchan no longer on Chromium, 2011/10/14 23:13:11]

Can you use DCHECK_GT?

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

[ramant (doing other things), 2011/10/14 23:42:45]

DCHECK_GE was giving errors for TimeDelta arguments (logging.h was trying to
create a string).

+    CloseSessionOnError(net::ERR_SPDY_PING_FAILED, true);
+    return;
+  }
+
+  // Check the status of connection after a delay.
+  base::TimeTicks now = base::TimeTicks::Now();

[jar (doing other things), 2011/10/14 19:59:07]

You mentioned this code might be clearer if you just did the now call up around
line 1606.  You can then calculate time since received_data_time, and subtract
kHungInterval (which is the amount of time you still need to wait).

If that is negative, then we fail and close connection.

If it is positive, then we post delayed task for that amount of tim.

[ramant (doing other things), 2011/10/14 23:42:45]

Done.

+  base::TimeDelta delay = kHungInterval - (now - received_data_time_);
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(&SpdySession::CheckStatus, 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