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() {}
+  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 51 matching lines @@
       max_concurrent_streams_(kDefaultMaxConcurrentStreams),
       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),
+      follower_ping_pending_(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);
 
+  int result = SendPing(stream_id);

[willchan no longer on Chromium, 2011/10/13 06:38:39]

You should queue this _after_ the SYN_STREAM. We don't want to slow down the
SYN_STREAM by the serialization latency of the PING.

And is there a reason you aren't doing this in SpdyHttpStream like you did
previously?

[willchan no longer on Chromium, 2011/10/13 15:47:17]

You didn't address this comment.
You didn't address this question.

[ramant (doing other things), 2011/10/13 21:41:14]

Jim said Roberto wanted us to send the pre-ping request early. Would be great if
you, Jim and Roberto can have a quick meeting. It is a very small change to move
the call into spdy_http_stream. thanks much.

+  if (result != ERR_IO_PENDING)
+    return result;
+
   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_++;
@@ skipping 118 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);
     bytes_read -= bytes_processed;
     data += bytes_processed;
     if (spdy_framer_.state() == spdy::SpdyFramer::SPDY_DONE)
       spdy_framer_.Reset();
   }
 
   if (state_ != CLOSED)
     ReadSocket();
 }
 
 void SpdySession::OnWriteComplete(int result) {
+  OnWriteCompleteInternal(result, false);
+}
+
+void SpdySession::OnWriteCompleteInternal(int result, bool ping_frame) {

[willchan no longer on Chromium, 2011/10/13 06:38:39]

You shouldn't need this extra bool parameter.

[ramant (doing other things), 2011/10/13 21:41:14]

Done.

   DCHECK(write_pending_);
   DCHECK(in_flight_write_.size());
 
   write_pending_ = false;
 
   scoped_refptr<SpdyStream> stream = in_flight_write_.stream();
 
   if (result >= 0) {
     // It should not be possible to have written more bytes than our
     // in_flight_write_.
     DCHECK_LE(result, in_flight_write_.buffer()->BytesRemaining());
 
     in_flight_write_.buffer()->DidConsume(result);
 
     // We only notify the stream when we've fully written the pending frame.
     if (!in_flight_write_.buffer()->BytesRemaining()) {
       if (stream) {
         // Report the number of bytes written to the caller, but exclude the
         // frame size overhead.  NOTE: if this frame was compressed the
         // reported bytes written is the compressed size, not the original
         // size.
         if (result > 0) {
           result = in_flight_write_.buffer()->size();
           DCHECK_GE(result, static_cast<int>(spdy::SpdyFrame::size()));
           result -= static_cast<int>(spdy::SpdyFrame::size());
         }
 
         // It is possible that the stream was cancelled while we were writing
         // to the socket.
-        if (!stream->cancelled())
+        if (!stream->cancelled() && !ping_frame)
           stream->OnWriteComplete(result);
       }
 
       // Cleanup the write which just completed.
       in_flight_write_.release();
     }
 
     // Write more data.  We're already in a continuation, so we can
     // go ahead and write it immediately (without going back to the
     // message loop).
@@ skipping 66 matching lines @@
   // closed, just return.
   if (state_ < CONNECTED || state_ == CLOSED)
     return;
 
   if (write_pending_)   // Another write is in progress still.
     return;
 
   // Loop sending frames until we've sent everything or until the write
   // returns error (or ERR_IO_PENDING).
   while (in_flight_write_.buffer() || !queue_.empty()) {
+    bool ping_frame = false;

[willchan no longer on Chromium, 2011/10/13 06:38:39]

You shouldn't need this.

[ramant (doing other things), 2011/10/13 21:41:14]

Done.

     if (!in_flight_write_.buffer()) {
       // Grab the next SpdyFrame to send.
       SpdyIOBuffer next_buffer = queue_.top();
       queue_.pop();
 
       // We've deferred compression until just before we write it to the socket,
       // which is now.  At this time, we don't compress our data frames.
       spdy::SpdyFrame uncompressed_frame(next_buffer.buffer()->data(), false);
       size_t size;
       if (spdy_framer_.IsCompressible(uncompressed_frame)) {
@@ skipping 11 matching lines @@
 
         // TODO(mbelshe): We have too much copying of data here.
         IOBufferWithSize* buffer = new IOBufferWithSize(size);
         memcpy(buffer->data(), compressed_frame->data(), size);
 
         // Attempt to send the frame.
         in_flight_write_ = SpdyIOBuffer(buffer, size, 0, next_buffer.stream());
       } else {
         size = uncompressed_frame.length() + spdy::SpdyFrame::size();
         in_flight_write_ = next_buffer;
+        if (spdy_framer_.IsPingFrame(uncompressed_frame))
+          ping_frame = true;
       }
     } else {
       DCHECK(in_flight_write_.buffer()->BytesRemaining());
     }
 
     write_pending_ = true;
     int rv = connection_->socket()->Write(in_flight_write_.buffer(),
         in_flight_write_.buffer()->BytesRemaining(), &write_callback_);
     if (rv == net::ERR_IO_PENDING)
       break;
 
     // We sent the frame successfully.
-    OnWriteComplete(rv);
+    OnWriteCompleteInternal(rv, ping_frame);
 
     // TODO(mbelshe):  Test this error case.  Maybe we should mark the socket
     //                 as in an error state.
     if (rv < 0)
       break;
   }
 }
 
 void SpdySession::CloseAllStreams(net::Error status) {
   base::StatsCounter abandoned_streams("spdy.abandoned_streams");
@@ skipping 416 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) {
+  --pings_in_flight_;
+  net_log_.AddEvent(
+      NetLog::TYPE_SPDY_SESSION_PING,
+      make_scoped_refptr(new NetLogSpdyPingParameter(frame.unique_id())));
+}
+
 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 118 matching lines @@
     switch (id) {
       case spdy::SETTINGS_MAX_CONCURRENT_STREAMS:
         max_concurrent_streams_ = std::min(static_cast<size_t>(val),
                                            max_concurrent_stream_limit_);
         ProcessPendingCreateStreams();
         break;
     }
   }
 }
 
+int SpdySession::SendPing(spdy::SpdyStreamId stream_id) {
+  const base::TimeDelta kInterval = base::TimeDelta::FromMilliseconds(10000);

[willchan no longer on Chromium, 2011/10/13 06:38:39]

Wow, this logic is not what we had discussed in the meeting. Why is there all
this pre-PING and post-PING stuff? Is this a new change after Jim and Roberto
talked? Can someone tell me what's going on?

[willchan no longer on Chromium, 2011/10/13 15:10:53]

From Raman:
"""
pre-PING and post-PING came after Jim and Roberto talked. 

Before we send the request, send a pre-PING if there are no ping's in flight.
Send a post-PING at the end of the request (if we didn't post a post-PING
already). Send post-PING's once in ?? (5) secs.
"""

Raman, just reply here directly so I don't have to copy / paste it in.

[ramant (doing other things), 2011/10/13 21:41:14]

Sorry about that.

+
+  base::TimeTicks now = base::TimeTicks::Now();
+  // If there are no PINGs in flight and we haven't heard from server, then
+  // send a pre-PING.
+  if ((pings_in_flight_ == 0) && ((now - received_data_time_) > kInterval)) {
+    int result = SendPrePing(stream_id);
+    if (result != ERR_IO_PENDING)
+      return result;
+  }
+
+  // Send a post-PING after a delay to make sure request has been received by
+  // the server.
+  const int kRequestTimeMs = 5000;
+  if (!follower_ping_pending_) {
+    follower_ping_pending_ = true;
+    MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &SpdySession::SendPostPing, stream_id),
+        kRequestTimeMs);
+
+    // Post a task to check the status of the connection.
+    const int kCheckStatusTimeMs = 10000;
+    MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        method_factory_.NewRunnableMethod(
+            &SpdySession::CheckStatus, stream_id, now),
+        kCheckStatusTimeMs);
+  }
+  return ERR_IO_PENDING;
+}
+
+int SpdySession::SendPrePing(spdy::SpdyStreamId stream_id) {
+  // Delay the writing of the PING frame so that it goes along with other
+  // frames.
+  AutoReset<bool> reset(&delayed_write_pending_, true);
+  return WritePingFrame(stream_id);
+}
+
+int SpdySession::SendPostPing(spdy::SpdyStreamId stream_id) {
+  DCHECK(follower_ping_pending_);
+  follower_ping_pending_ = false;
+  return WritePingFrame(stream_id);
+}
+
+int SpdySession::WritePingFrame(spdy::SpdyStreamId stream_id) {
+  // 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);
+
+  scoped_ptr<spdy::SpdyPingControlFrame> ping_frame(
+      spdy_framer_.CreatePingFrame(unique_id_counter_));
+  QueueFrame(ping_frame.get(), SPDY_PRIORITY_HIGHEST, stream);

[willchan no longer on Chromium, 2011/10/13 06:38:39]

Here's the problem. Pass a NULL stream to QueueFrame(). A PING frame shouldn't
be tied to a stream.

[ramant (doing other things), 2011/10/13 21:41:14]

Done.

+
+  if (net_log().IsLoggingAllEvents()) {
+    net_log().AddEvent(
+        NetLog::TYPE_SPDY_SESSION_PING,
+        make_scoped_refptr(new NetLogSpdyPingParameter(unique_id_counter_)));
+  }
+
+  ++pings_in_flight_;
+  unique_id_counter_ += 2;
+  return ERR_IO_PENDING;
+}
+
+void SpdySession::CheckStatus(spdy::SpdyStreamId stream_id,
+                              base::TimeTicks last_check_time) {
+  // Find our stream.
+  if (!IsStreamActive(stream_id))
+    return;
+  const scoped_refptr<SpdyStream>& stream = active_streams_[stream_id];
+  CHECK_EQ(stream->stream_id(), stream_id);
+
+  // Check if we got a response back for all PINGs we had sent.
+  if (pings_in_flight_ == 0)
+    return;
+
+  // Check if we haven't received any data in |kHungInterval|.
+  const base::TimeDelta kHungInterval = base::TimeDelta::FromSeconds(10);
+  if (received_data_time_ < last_check_time) {
+    DCHECK(base::TimeTicks::Now() - received_data_time_ > kHungInterval);
+    DeleteStream(stream_id, ERR_SPDY_PING_FAILED);
+    return;
+  }
+
+  // Check the status of connection after a delay.
+  base::TimeTicks now = base::TimeTicks::Now();
+  base::TimeDelta delay = now - received_data_time_ + kHungInterval;
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      method_factory_.NewRunnableMethod(
+          &SpdySession::CheckStatus, stream_id, 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