http2: Update priorities of pushed streams

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 <limits>
 #include <map>
 #include <utility>
@@ skipping 89 matching lines @@
 
 std::unique_ptr<base::Value> NetLogSpdyHeadersSentCallback(
     const SpdyHeaderBlock* headers,
     bool fin,
     SpdyStreamId stream_id,
     bool has_priority,
     int weight,
     SpdyStreamId parent_stream_id,
     bool exclusive,
     NetLogCaptureMode capture_mode) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->Set("headers", ElideSpdyHeaderBlockForNetLog(*headers, capture_mode));
   dict->SetBoolean("fin", fin);
   dict->SetInteger("stream_id", stream_id);
   dict->SetBoolean("has_priority", has_priority);
   if (has_priority) {
     dict->SetInteger("parent_stream_id", parent_stream_id);
     dict->SetInteger("weight", weight);
     dict->SetBoolean("exclusive", exclusive);
   }
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyHeadersReceivedCallback(
     const SpdyHeaderBlock* headers,
     bool fin,
     SpdyStreamId stream_id,
     NetLogCaptureMode capture_mode) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->Set("headers", ElideSpdyHeaderBlockForNetLog(*headers, capture_mode));
   dict->SetBoolean("fin", fin);
   dict->SetInteger("stream_id", stream_id);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySessionCloseCallback(
     int net_error,
     const std::string* description,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("net_error", net_error);
   dict->SetString("description", *description);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySessionCallback(
     const HostPortProxyPair* host_pair,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetString("host", host_pair->first.ToString());
   dict->SetString("proxy", host_pair->second.ToPacString());
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyInitializedCallback(
     NetLogSource source,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   if (source.IsValid()) {
     source.AddToEventParameters(dict.get());
   }
   dict->SetString("protocol", NextProtoToString(kProtoHTTP2));
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySettingsCallback(
     const HostPortPair& host_port_pair,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetString("host", host_port_pair.ToString());
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySettingCallback(
     SpdySettingsIds id,
     uint32_t value,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("id", id);
   dict->SetInteger("value", value);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySendSettingsCallback(
     const SettingsMap* settings,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
-  std::unique_ptr<base::ListValue> settings_list(new base::ListValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
+  auto settings_list = base::MakeUnique<base::ListValue>();
   for (SettingsMap::const_iterator it = settings->begin();
        it != settings->end(); ++it) {
     const SpdySettingsIds id = it->first;
     const uint32_t value = it->second;
     settings_list->AppendString(
         base::StringPrintf("[id:%u value:%u]", id, value));
   }
   dict->Set("settings", std::move(settings_list));
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyWindowUpdateFrameCallback(
     SpdyStreamId stream_id,
     uint32_t delta,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("stream_id", static_cast<int>(stream_id));
   dict->SetInteger("delta", delta);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySessionWindowUpdateCallback(
     int32_t delta,
     int32_t window_size,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("delta", delta);
   dict->SetInteger("window_size", window_size);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyDataCallback(
     SpdyStreamId stream_id,
     int size,
     bool fin,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("stream_id", static_cast<int>(stream_id));
   dict->SetInteger("size", size);
   dict->SetBoolean("fin", fin);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyRstCallback(
     SpdyStreamId stream_id,
     int status,
     const std::string* description,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("stream_id", static_cast<int>(stream_id));
   dict->SetInteger("status", status);
   dict->SetString("description", *description);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyPingCallback(
     SpdyPingId unique_id,
     bool is_ack,
     const char* type,
     NetLogCaptureMode /* capture_mode */) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("unique_id", static_cast<int>(unique_id));
   dict->SetString("type", type);
   dict->SetBoolean("is_ack", is_ack);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyGoAwayCallback(
     SpdyStreamId last_stream_id,
     int active_streams,
     int unclaimed_streams,
     SpdyGoAwayStatus status,
     base::StringPiece debug_data,
     NetLogCaptureMode capture_mode) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("last_accepted_stream_id",
                    static_cast<int>(last_stream_id));
   dict->SetInteger("active_streams", active_streams);
   dict->SetInteger("unclaimed_streams", unclaimed_streams);
   dict->SetInteger("status", static_cast<int>(status));
   dict->SetString("debug_data",
                   ElideGoAwayDebugDataForNetLog(capture_mode, debug_data));
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyPushPromiseReceivedCallback(
     const SpdyHeaderBlock* headers,
     SpdyStreamId stream_id,
     SpdyStreamId promised_stream_id,
     NetLogCaptureMode capture_mode) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->Set("headers", ElideSpdyHeaderBlockForNetLog(*headers, capture_mode));
   dict->SetInteger("id", stream_id);
   dict->SetInteger("promised_stream_id", promised_stream_id);
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdyAdoptedPushStreamCallback(
     SpdyStreamId stream_id,
     const GURL* url,
     NetLogCaptureMode capture_mode) {
-  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
+  auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("stream_id", stream_id);
   dict->SetString("url", url->spec());
   return std::move(dict);
 }
 
 std::unique_ptr<base::Value> NetLogSpdySessionStalledCallback(
     size_t num_active_streams,
     size_t num_created_streams,
     size_t num_pushed_streams,
     size_t max_concurrent_streams,
     const std::string& url,
     NetLogCaptureMode capture_mode) {
   auto dict = base::MakeUnique<base::DictionaryValue>();
   dict->SetInteger("num_active_streams", num_active_streams);
   dict->SetInteger("num_created_streams", num_created_streams);
   dict->SetInteger("num_pushed_streams", num_pushed_streams);
   dict->SetInteger("max_concurrent_streams", max_concurrent_streams);
   dict->SetString("url", url);
   return std::move(dict);
 }
 
+std::unique_ptr<base::Value> NetLogSpdyPriorityCallback(
+    SpdyStreamId stream_id,
+    SpdyStreamId parent_stream_id,
+    int weight,
+    bool exclusive,
+    NetLogCaptureMode capture_mode) {
+  auto dict = base::MakeUnique<base::DictionaryValue>();
+  dict->SetInteger("stream_id", stream_id);
+  dict->SetInteger("parent_stream_id", parent_stream_id);
+  dict->SetInteger("weight", weight);
+  dict->SetBoolean("exclusive", exclusive);
+  return std::move(dict);
+}
+
 // Helper function to return the total size of an array of objects
 // with .size() member functions.
 template <typename T, size_t N> size_t GetTotalSize(const T (&arr)[N]) {
   size_t total_size = 0;
   for (size_t i = 0; i < N; ++i) {
     total_size += arr[i].size();
   }
   return total_size;
 }
 
@@ skipping 497 matching lines @@
 
   SSLInfo ssl_info;
   if (!GetSSLInfo(&ssl_info))
     return true;   // This is not a secure session, so all domains are okay.
 
   return CanPool(transport_security_state_, ssl_info,
                  host_port_pair().host(), domain);
 }
 
 int SpdySession::GetPushStream(const GURL& url,
+                               RequestPriority priority,
                                base::WeakPtr<SpdyStream>* stream,
                                const NetLogWithSource& stream_net_log) {
   CHECK(!in_io_loop_);
 
   stream->reset();
 
   if (availability_state_ == STATE_DRAINING)
     return ERR_CONNECTION_CLOSED;
 
   *stream = GetActivePushStream(url);
   if (*stream) {
     DCHECK_LT(streams_pushed_and_claimed_count_, streams_pushed_count_);
     streams_pushed_and_claimed_count_++;
+
+    // If the stream is still open, update its priority to match
+    // the priority of the matching request.
+    if (!(*stream)->IsClosed() && (*stream)->priority() != priority) {
+      (*stream)->set_priority(priority);
+
+      // Send PRIORITY updates.
+      auto updates = priority_dependency_state_.OnStreamUpdate(
+          (*stream)->stream_id(),
+          ConvertRequestPriorityToSpdyPriority(priority));
+      for (auto u : updates) {
+        ActiveStreamMap::iterator it = active_streams_.find(u.id);
+        DCHECK(it != active_streams_.end());
+        int weight = Spdy3PriorityToHttp2Weight(
+            ConvertRequestPriorityToSpdyPriority(it->second->priority()));
+        EnqueuePriorityFrame(u.id, u.dependent_stream_id, weight, u.exclusive);
+      }
+    }
   }
+
   return OK;
 }
 
 void SpdySession::CancelPush(const GURL& url) {
   UnclaimedPushedStreamContainer::const_iterator unclaimed_it =
       unclaimed_pushed_streams_.find(url);
   if (unclaimed_it == unclaimed_pushed_streams_.end())
     return;
 
   SpdyStreamId stream_id = unclaimed_it->second.stream_id;
@@ skipping 211 matching lines @@
 
   DCHECK(buffered_spdy_framer_.get());
   SpdyPriority spdy_priority = ConvertRequestPriorityToSpdyPriority(priority);
 
   std::unique_ptr<SpdySerializedFrame> syn_frame;
   bool has_priority = true;
   int weight = Spdy3PriorityToHttp2Weight(spdy_priority);
   SpdyStreamId dependent_stream_id = 0;
   bool exclusive = false;
 
-  priority_dependency_state_.OnStreamSynSent(stream_id, spdy_priority,
-                                             &dependent_stream_id, &exclusive);
+  priority_dependency_state_.OnStreamCreation(stream_id, spdy_priority,
+                                              &dependent_stream_id, &exclusive);
 
   if (net_log().IsCapturing()) {
     net_log().AddEvent(
         NetLogEventType::HTTP2_SESSION_SEND_HEADERS,
         base::Bind(&NetLogSpdyHeadersSentCallback, &block,
                    (flags & CONTROL_FLAG_FIN) != 0, stream_id, has_priority,
                    weight, dependent_stream_id, exclusive));
   }
 
   SpdyHeadersIR headers(stream_id, std::move(block));
@@ skipping 237 matching lines @@
       base::Bind(&NetLogSpdyRstCallback, stream_id, status, &description));
 
   DCHECK(buffered_spdy_framer_.get());
   std::unique_ptr<SpdySerializedFrame> rst_frame(
       buffered_spdy_framer_->CreateRstStream(stream_id, status));
 
   EnqueueSessionWrite(priority, RST_STREAM, std::move(rst_frame));
   RecordProtocolErrorHistogram(MapRstStreamStatusToProtocolError(status));
 }
 
+void SpdySession::EnqueuePriorityFrame(SpdyStreamId stream_id,
+                                       SpdyStreamId dependency_id,
+                                       int weight,
+                                       bool exclusive) {
+  net_log().AddEvent(NetLogEventType::HTTP2_STREAM_SEND_PRIORITY,
+                     base::Bind(&NetLogSpdyPriorityCallback, stream_id,
+                                dependency_id, weight, exclusive));
+
+  DCHECK(buffered_spdy_framer_.get());
+  std::unique_ptr<SpdySerializedFrame> frame(
+      buffered_spdy_framer_->CreatePriority(stream_id, dependency_id, weight,
+                                            exclusive));
+
+  // PRIORITY frames describe sequenced updates to the tree, so they must
+  // be serialized. We do this by queueing all PRIORITY frames at HIGHEST
+  // priority.
+  EnqueueWrite(HIGHEST, PRIORITY,
+               base::MakeUnique<SimpleBufferProducer>(
+                   base::MakeUnique<SpdyBuffer>(std::move(frame))),
+               base::WeakPtr<SpdyStream>());
+}
+
 void SpdySession::PumpReadLoop(ReadState expected_read_state, int result) {
   CHECK(!in_io_loop_);
   if (availability_state_ == STATE_DRAINING) {
     return;
   }
   ignore_result(DoReadLoop(expected_read_state, result));
 }
 
 int SpdySession::DoReadLoop(ReadState expected_read_state, int result) {
   CHECK(!in_io_loop_);
@@ skipping 1117 matching lines @@
       return;
     }
 
     CHECK_EQ(it->second->stream_id(), stream_id);
     it->second->IncreaseSendWindowSize(delta_window_size);
   }
 }
 
 void SpdySession::TryCreatePushStream(SpdyStreamId stream_id,
                                       SpdyStreamId associated_stream_id,
-                                      SpdyPriority priority,
                                       SpdyHeaderBlock headers) {
   // Server-initiated streams should have even sequence numbers.
   if ((stream_id & 0x1) != 0) {
     LOG(WARNING) << "Received invalid push stream id " << stream_id;
     CloseSessionOnError(ERR_SPDY_PROTOCOL_ERROR, "Odd push stream id.");
     return;
   }
 
   // Server-initiated streams must be associated with client-initiated streams.
   if ((associated_stream_id & 0x1) != 1) {
@@ skipping 14 matching lines @@
 
   if (IsStreamActive(stream_id)) {
     // We should not get here, we'll start going away earlier on
     // |last_seen_push_stream_id_| check.
     LOG(WARNING) << "Received push for active stream " << stream_id;
     return;
   }
 
   last_accepted_push_stream_id_ = stream_id;
 
-  RequestPriority request_priority =
-      ConvertSpdyPriorityToRequestPriority(priority);
+  // Pushed streams are speculative, so they start at an IDLE priority.
+  const RequestPriority request_priority = IDLE;
 
   if (availability_state_ == STATE_GOING_AWAY) {
     // TODO(akalin): This behavior isn't in the SPDY spec, although it
     // probably should be.
     EnqueueResetStreamFrame(stream_id,
                             request_priority,
                             RST_STREAM_REFUSED_STREAM,
                             "push stream request received when going away");
     return;
   }
@@ skipping 96 matching lines @@
         "Received duplicate pushed stream with url: " + gurl.spec());
     return;
   }
 
   std::unique_ptr<SpdyStream> stream(
       new SpdyStream(SPDY_PUSH_STREAM, GetWeakPtr(), gurl, request_priority,
                      stream_initial_send_window_size_,
                      stream_max_recv_window_size_, net_log_));
   stream->set_stream_id(stream_id);
 
+  // Convert RequestPriority to a SpdyPriority to send in a PRIORITY frame.
+  SpdyPriority spdy_priority =
+      ConvertRequestPriorityToSpdyPriority(request_priority);
+  SpdyStreamId dependency_id = 0;
+  bool exclusive = false;
+  priority_dependency_state_.OnStreamCreation(stream_id, spdy_priority,
+                                              &dependency_id, &exclusive);
+  EnqueuePriorityFrame(stream_id, dependency_id,
+                       Spdy3PriorityToHttp2Weight(spdy_priority), exclusive);
+
   // PUSH_PROMISE arrives on associated stream.
   associated_it->second->AddRawReceivedBytes(last_compressed_frame_len_);
   last_compressed_frame_len_ = 0;
 
   UnclaimedPushedStreamContainer::const_iterator inserted_pushed_it =
       unclaimed_pushed_streams_.insert(pushed_it, gurl, stream_id,
                                        time_func_());
   DCHECK(inserted_pushed_it != pushed_it);
   DeleteExpiredPushedStreams();
 
@@ skipping 21 matching lines @@
                                 SpdyStreamId promised_stream_id,
                                 SpdyHeaderBlock headers) {
   CHECK(in_io_loop_);
 
   if (net_log_.IsCapturing()) {
     net_log_.AddEvent(NetLogEventType::HTTP2_SESSION_RECV_PUSH_PROMISE,
                       base::Bind(&NetLogSpdyPushPromiseReceivedCallback,
                                  &headers, stream_id, promised_stream_id));
   }
 
-  // Any priority will do.  TODO(baranovich): Pass parent stream id priority?
-  TryCreatePushStream(promised_stream_id, stream_id, 0, std::move(headers));
+  TryCreatePushStream(promised_stream_id, stream_id, std::move(headers));
 }
 
 void SpdySession::SendStreamWindowUpdate(SpdyStreamId stream_id,
                                          uint32_t delta_window_size) {
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   CHECK(it != active_streams_.end());
   CHECK_EQ(it->second->stream_id(), stream_id);
   SendWindowUpdateFrame(stream_id, delta_window_size, it->second->priority());
 }
 
@@ skipping 392 matching lines @@
     if (!queue->empty()) {
       SpdyStreamId stream_id = queue->front();
       queue->pop_front();
       return stream_id;
     }
   }
   return 0;
 }
 
 }  // namespace net