Initial implementation of RequestPriority-based HTTP/2 dependencies.

net/spdy/spdy_session.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/spdy/spdy_session.h"
 
 #include <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
@@ skipping 41 matching lines @@
 
 namespace {
 
 const int kReadBufferSize = 8 * 1024;
 const int kDefaultConnectionAtRiskOfLossSeconds = 10;
 const int kHungIntervalSeconds = 10;
 
 // Minimum seconds that unclaimed pushed streams will be kept in memory.
 const int kMinPushedStreamLifetimeSeconds = 300;
 
+// Field trial constants
+const char kSpdyDependenciesFieldTrial[] = "SpdyEnableDependencies";
+const char kSpdyDepencenciesFieldTrialEnable[] = "Enable";
+
+// Whether the creation of SPDY dependencies based on priority is
+// enabled by default.
+static bool priority_dependency_enabled_default = false;
+
 scoped_ptr<base::ListValue> SpdyHeaderBlockToListValue(
     const SpdyHeaderBlock& headers,
     NetLogCaptureMode capture_mode) {
   scoped_ptr<base::ListValue> headers_list(new base::ListValue());
   for (SpdyHeaderBlock::const_iterator it = headers.begin();
        it != headers.end(); ++it) {
     headers_list->AppendString(
         it->first.as_string() + ": " +
         ElideHeaderValueForNetLog(capture_mode, it->first.as_string(),
                                   it->second.as_string()));
@@ skipping 612 matching lines @@
       enable_sending_initial_data_(enable_sending_initial_data),
       enable_compression_(enable_compression),
       enable_ping_based_connection_checking_(
           enable_ping_based_connection_checking),
       protocol_(default_protocol),
       connection_at_risk_of_loss_time_(
           base::TimeDelta::FromSeconds(kDefaultConnectionAtRiskOfLossSeconds)),
       hung_interval_(base::TimeDelta::FromSeconds(kHungIntervalSeconds)),
       trusted_spdy_proxy_(trusted_spdy_proxy),
       time_func_(time_func),
+      send_priority_dependency_(priority_dependency_enabled_default),
       weak_factory_(this) {
   DCHECK_GE(protocol_, kProtoSPDYMinimumVersion);
   DCHECK_LE(protocol_, kProtoSPDYMaximumVersion);
   DCHECK(HttpStreamFactory::spdy_enabled());
   net_log_.BeginEvent(
       NetLog::TYPE_HTTP2_SESSION,
       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair()));
   next_unclaimed_push_stream_sweep_time_ = time_func_() +
       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
+  if (base::FieldTrialList::FindFullName(kSpdyDependenciesFieldTrial) ==
+      kSpdyDepencenciesFieldTrialEnable) {
+    send_priority_dependency_ = true;
+  }
   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 }
 
 SpdySession::~SpdySession() {
   CHECK(!in_io_loop_);
   DcheckDraining();
 
   // TODO(akalin): Check connection->is_initialized() instead. This
   // requires re-working CreateFakeSpdySession(), though.
   DCHECK(connection_->socket());
@@ skipping 325 matching lines @@
 bool SpdySession::CloseOneIdleConnection() {
   CHECK(!in_io_loop_);
   DCHECK(pool_);
   if (active_streams_.empty()) {
     DoDrainSession(ERR_CONNECTION_CLOSED, "Closing idle connection.");
   }
   // Return false as the socket wasn't immediately closed.
   return false;
 }
 
+// static
+void SpdySession::SetPriorityDependencyDefaultForTesting(bool enable) {
+  priority_dependency_enabled_default = enable;
+}
+
 void SpdySession::EnqueueStreamWrite(
     const base::WeakPtr<SpdyStream>& stream,
     SpdyFrameType frame_type,
     scoped_ptr<SpdyBufferProducer> producer) {
   DCHECK(frame_type == HEADERS ||
          frame_type == DATA ||
          frame_type == CREDENTIAL ||
          frame_type == SYN_STREAM);
   EnqueueWrite(stream->priority(), frame_type, producer.Pass(), stream);
 }
@@ skipping 20 matching lines @@
     syn_stream.set_associated_to_stream_id(0);
     syn_stream.set_priority(spdy_priority);
     syn_stream.set_fin((flags & CONTROL_FLAG_FIN) != 0);
     syn_stream.set_unidirectional((flags & CONTROL_FLAG_UNIDIRECTIONAL) != 0);
     syn_stream.set_header_block(block);
     syn_frame.reset(buffered_spdy_framer_->SerializeFrame(syn_stream));
   } else {
     SpdyHeadersIR headers(stream_id);
     headers.set_priority(spdy_priority);
     headers.set_has_priority(true);
+
+    if (send_priority_dependency_) {
+      // Set dependencies to reflect request priority.  A newly created
+      // stream should be dependent on the most recent previously created
+      // stream of the same priority level.  The newly created stream
+      // should also have all streams of a lower priority level dependent
+      // on it, which is guaranteed by setting the exclusive bit.
+      //
+      // Note that this depends on stream ids being allocated in a monotonically
+      // increasing fashion, and on all streams in
+      // active_streams_{,by_priority_} having stream ids set.
+      for (int i = priority; i >= IDLE; --i) {
+        if (active_streams_by_priority_[i].empty())
+          continue;
+
+        auto candidate_it = active_streams_by_priority_[i].rbegin();
+
+        // |active_streams_by_priority_| is updated before the
+        // SYN stream frame is created, so the current streams
+        // id is already on the list.  Skip over it, skipping this
+        // priority level if it's singular.
+        if (candidate_it->second->stream_id() == stream_id)
+          ++candidate_it;
+        if (candidate_it == active_streams_by_priority_[i].rend())
+          continue;
+
+        headers.set_parent_stream_id(candidate_it->second->stream_id());
+        break;
+      }
+
+      // If there are no streams of priority <= the current stream, the
+      // current stream will default to a child of the idle node (0).
+      headers.set_exclusive(true);
+    }
+
     headers.set_fin((flags & CONTROL_FLAG_FIN) != 0);
     headers.set_header_block(block);
     syn_frame.reset(buffered_spdy_framer_->SerializeFrame(headers));
   }
 
   streams_initiated_count_++;
 
   if (net_log().IsCapturing()) {
     const NetLog::EventType type =
         (GetProtocolVersion() <= SPDY3)
@@ skipping 186 matching lines @@
   return LOAD_STATE_IDLE;
 }
 
 void SpdySession::CloseActiveStreamIterator(ActiveStreamMap::iterator it,
                                             int status) {
   // TODO(mbelshe): We should send a RST_STREAM control frame here
   //                so that the server can cancel a large send.
 
   scoped_ptr<SpdyStream> owned_stream(it->second.stream);
   active_streams_.erase(it);
+  active_streams_by_priority_[owned_stream->priority()].erase(
+      owned_stream->stream_id());
 
   // TODO(akalin): When SpdyStream was ref-counted (and
   // |unclaimed_pushed_streams_| held scoped_refptr<SpdyStream>), this
   // was only done when status was not OK. This meant that pushed
   // streams can still be claimed after they're closed. This is
   // probably something that we still want to support, although server
   // push is hardly used. Write tests for this and fix this. (See
   // http://crbug.com/261712 .)
   if (owned_stream->type() == SPDY_PUSH_STREAM) {
     unclaimed_pushed_streams_.erase(owned_stream->url());
@@ skipping 641 matching lines @@
   created_streams_.erase(stream);
   return owned_stream.Pass();
 }
 
 void SpdySession::InsertActivatedStream(scoped_ptr<SpdyStream> stream) {
   SpdyStreamId stream_id = stream->stream_id();
   CHECK_NE(stream_id, 0u);
   std::pair<ActiveStreamMap::iterator, bool> result =
       active_streams_.insert(
           std::make_pair(stream_id, ActiveStreamInfo(stream.get())));
+  active_streams_by_priority_[stream->priority()].insert(
+      std::make_pair(stream_id, stream.get()));
   CHECK(result.second);
   ignore_result(stream.release());
 }
 
 void SpdySession::DeleteStream(scoped_ptr<SpdyStream> stream, int status) {
   if (in_flight_write_stream_.get() == stream.get()) {
     // If we're deleting the stream for the in-flight write, we still
     // need to let the write complete, so we clear
     // |in_flight_write_stream_| and let the write finish on its own
     // without notifying |in_flight_write_stream_|.
@@ skipping 1312 matching lines @@
     if (!queue->empty()) {
       SpdyStreamId stream_id = queue->front();
       queue->pop_front();
       return stream_id;
     }
   }
   return 0;
 }
 
 }  // namespace net