Implement MAX_CONCURRENT_STREAMS SETTINGS header

net/spdy/spdy_session.cc

 // Copyright (c) 2010 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/basictypes.h"
 #include "base/linked_ptr.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
@@ skipping 138 matching lines @@
       connection_(new ClientSocketHandle),
       read_buffer_(new IOBuffer(kReadBufferSize)),
       read_pending_(false),
       stream_hi_water_mark_(1),  // Always start at 1 for the first stream id.
       write_pending_(false),
       delayed_write_pending_(false),
       is_secure_(false),
       certificate_error_code_(OK),
       error_(OK),
       state_(IDLE),
+      max_concurrent_streams_(kDefaultMaxConcurrentStreams),
       streams_initiated_count_(0),
       streams_pushed_count_(0),
       streams_pushed_and_claimed_count_(0),
       streams_abandoned_count_(0),
       sent_settings_(false),
       received_settings_(false),
       in_session_pool_(true),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)) {
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
@@ skipping 116 matching lines @@
     it->second = *stream;
     return OK;
   }
   return OK;
 }
 
 int SpdySession::CreateStream(
     const GURL& url,
     RequestPriority priority,
     scoped_refptr<SpdyStream>* spdy_stream,
+    const BoundNetLog& stream_net_log,
+    CompletionCallback* callback,
+    const SpdyHttpStream* spdy_http_stream) {
+  if (!max_concurrent_streams_ ||
+      active_streams_.size() < max_concurrent_streams_) {
+    return CreateStreamImpl(url, priority, spdy_stream, stream_net_log);
+  }
+
+  create_stream_queues_[priority].push(
+      PendingCreateStream(url, priority, spdy_stream,
+                          stream_net_log, callback, spdy_http_stream));
+  return ERR_IO_PENDING;
+}
+
+void SpdySession::ProcessPendingCreateStreams() {
+  while (!max_concurrent_streams_ ||
+         active_streams_.size() < max_concurrent_streams_) {
+    bool no_pending_create_streams = true;
+    for (int i = 0;i < NUM_PRIORITIES;++i) {
+      if (!create_stream_queues_[i].empty()) {
+        PendingCreateStream& pending_create = create_stream_queues_[i].front();
+        no_pending_create_streams = false;
+        int error = CreateStreamImpl(*pending_create.url,
+                                     pending_create.priority,
+                                     pending_create.spdy_stream,
+                                     *pending_create.stream_net_log);
+        pending_create.callback->Run(error);
+        create_stream_queues_[i].pop();
+        break;
+      }
+    }
+    if (no_pending_create_streams)
+      return;  // there were no streams in any queue
+  }
+}
+
+void SpdySession::CancelPendingCreateStreams(
+    const SpdyHttpStream *const spdy_http_stream) {
+  for (int i = 0;i < NUM_PRIORITIES;++i) {
+    PendingCreateStreamQueue tmp;
+    // Make a copy removing this trans
+    while (!create_stream_queues_[i].empty()) {
+      PendingCreateStream& pending_create = create_stream_queues_[i].front();
+      if (pending_create.spdy_http_stream != spdy_http_stream)
+        tmp.push(pending_create);
+      create_stream_queues_[i].pop();
+    }
+    // Now copy it back
+    while (!tmp.empty()) {
+      create_stream_queues_[i].push(tmp.front());
+      tmp.pop();
+    }
+  }
+}
+
+int SpdySession::CreateStreamImpl(
+    const GURL& url,
+    RequestPriority priority,
+    scoped_refptr<SpdyStream>* spdy_stream,
     const BoundNetLog& stream_net_log) {
   // Make sure that we don't try to send https/wss over an unauthenticated, but
   // encrypted SSL socket.
   if (is_secure_ && certificate_error_code_ != OK &&
       (url.SchemeIs("https") || url.SchemeIs("wss"))) {
     LOG(DFATAL) << "Tried to create spdy stream for secure content over an "
                 << "unauthenticated session.";
     return certificate_error_code_;
   }
 
@@ skipping 395 matching lines @@
   static StatsCounter abandoned_streams("spdy.abandoned_streams");
   static StatsCounter abandoned_push_streams("spdy.abandoned_push_streams");
 
   if (!active_streams_.empty())
     abandoned_streams.Add(active_streams_.size());
   if (!pushed_streams_.empty()) {
     streams_abandoned_count_ += pushed_streams_.size();
     abandoned_push_streams.Add(pushed_streams_.size());
   }
 
+  for (int i = 0;i < NUM_PRIORITIES;++i) {
+    while (!create_stream_queues_[i].empty()) {
+      PendingCreateStream& pending_create = create_stream_queues_[i].front();
+      pending_create.callback->Run(ERR_ABORTED);
+      create_stream_queues_[i].pop();
+    }
+  }
+
   while (!active_streams_.empty()) {
     ActiveStreamMap::iterator it = active_streams_.begin();
     const scoped_refptr<SpdyStream>& stream = it->second;
     DCHECK(stream);
     LOG(ERROR) << "ABANDONED (stream_id=" << stream->stream_id()
       << "): " << stream->path();
     DeleteStream(stream->stream_id(), status);
   }
 
   // TODO(erikchen): ideally stream->OnClose() is only ever called by
   // DeleteStream, but pending streams fall into their own category for now.
   PendingStreamMap::iterator it;
-  for (it = pending_streams_.begin(); it != pending_streams_.end(); ++it)
-  {
+  for (it = pending_streams_.begin(); it != pending_streams_.end(); ++it) {
     const scoped_refptr<SpdyStream>& stream = it->second;
     if (stream)
       stream->OnClose(ERR_ABORTED);
   }
   pending_streams_.clear();
 
   // We also need to drain the queue.
   while (queue_.size())
     queue_.pop();
 }
@@ skipping 58 matching lines @@
   // The stream might have been deleted.
   ActiveStreamMap::iterator it2 = active_streams_.find(id);
   if (it2 == active_streams_.end())
     return;
 
   // If this is an active stream, call the callback.
   const scoped_refptr<SpdyStream> stream(it2->second);
   active_streams_.erase(it2);
   if (stream)
     stream->OnClose(status);
+  ProcessPendingCreateStreams();
 }
 
 void SpdySession::RemoveFromPool() {
   if (in_session_pool_) {
     session_->spdy_session_pool()->Remove(this);
     in_session_pool_ = false;
   }
 }
 
 scoped_refptr<SpdyStream> SpdySession::GetActivePushStream(
@@ skipping 299 matching lines @@
   // |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::OnSettings(const spdy::SpdySettingsControlFrame& frame) {
   spdy::SpdySettings settings;
   if (spdy_framer_.ParseSettings(&frame, &settings)) {
+    HandleSettings(settings);
     SpdySettingsStorage* settings_storage = session_->mutable_spdy_settings();
     settings_storage->Set(host_port_pair_, settings);
   }
 
   received_settings_ = true;
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_RECV_SETTINGS,
       new NetLogSpdySettingsParameter(settings));
 }
 
 void SpdySession::SendSettings() {
   const SpdySettingsStorage& settings_storage = session_->spdy_settings();
   const spdy::SpdySettings& settings = settings_storage.Get(host_port_pair_);
   if (settings.empty())
     return;
+  HandleSettings(settings);
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_SEND_SETTINGS,
       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);
 }
 
+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_ = val;
+        ProcessPendingCreateStreams();
+        break;
+    }
+  }
+}
+
 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);
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsAbandonedPerSession",
                               streams_abandoned_count_,
                               0, 300, 50);
   UMA_HISTOGRAM_ENUMERATION("Net.SpdySettingsSent",
                             sent_settings_ ? 1 : 0, 2);
   UMA_HISTOGRAM_ENUMERATION("Net.SpdySettingsReceived",
                             received_settings_ ? 1 : 0, 2);
 
   if (received_settings_) {
     // Enumerate the saved settings, and set histograms for it.
     const SpdySettingsStorage& settings_storage = session_->spdy_settings();
     const spdy::SpdySettings& settings = settings_storage.Get(host_port_pair_);
-    if (settings.empty()) {
-      NOTREACHED();  // If we lost our settings already, something is wrong!
-      return;
-    }
 
     spdy::SpdySettings::const_iterator it;
     for (it = settings.begin(); it != settings.end(); ++it) {
       const spdy::SpdySetting setting = *it;
       switch (setting.first.id()) {
         case spdy::SETTINGS_CURRENT_CWND:
           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd",
                                       setting.second,
                                       1, 200, 100);
           break;
         case spdy::SETTINGS_ROUND_TRIP_TIME:
           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRTT",
                                       setting.second,
                                       1, 1200, 100);
           break;
         case spdy::SETTINGS_DOWNLOAD_RETRANS_RATE:
           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRetransRate",
                                       setting.second,
                                       1, 100, 50);
           break;
       }
     }
   }
 }
 
 }  // namespace net