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>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/logging.h"
+#include "base/memory/ptr_util.h"
 #include "base/metrics/histogram_macros.h"
 #include "base/metrics/sparse_histogram.h"
 #include "base/profiler/scoped_tracker.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/strings/utf_string_conversions.h"
 #include "base/threading/thread_task_runner_handle.h"
@@ skipping 232 matching lines @@
 std::unique_ptr<base::Value> NetLogSpdyAdoptedPushStreamCallback(
     SpdyStreamId stream_id,
     const GURL* url,
     NetLogCaptureMode capture_mode) {
   std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());
   dict->SetInteger("stream_id", stream_id);
   dict->SetString("url", url->spec());
   return std::move(dict);
 }
 
+std::unique_ptr<base::Value> NetLogSpdyPriorityCallback(
+    SpdyStreamId stream_id,
+    SpdyStreamId parent_stream_id,
+    int weight,
+    bool exclusive,
+    NetLogCaptureMode capture_mode) {
+  std::unique_ptr<base::DictionaryValue> dict(new base::DictionaryValue());

[Bence, 2017/01/05 17:14:46]

Please use MakeUnique in new code, see
https://cs.chromium.org/chromium/src/base/memory/ptr_util.h?l=43.

[Tom Bergan, 2017/01/06 00:08:41]

Done. Also did a search/replace on the other NetLog callbacks above.

+  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 487 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 205 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_UPDATE_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 1118 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 394 matching lines @@
     if (!queue->empty()) {
       SpdyStreamId stream_id = queue->front();
       queue->pop_front();
       return stream_id;
     }
   }
   return 0;
 }
 
 }  // namespace net