Fix out of order SYN_STEAM frames.

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 <map>
 
 #include "base/basictypes.h"
 #include "base/bind.h"
@@ skipping 162 matching lines @@
 
 NextProto g_default_protocol = kProtoUnknown;
 size_t g_init_max_concurrent_streams = 100;
 size_t g_max_concurrent_stream_limit = 256;
 size_t g_default_initial_rcv_window_size = 10 * 1024 * 1024;  // 10MB
 bool g_enable_ping_based_connection_checking = true;
 
 }  // namespace
 
 // static
+void SpdySession::SpdyIOBufferProducer::ActivateStream(
+    SpdySession* spdy_session,
+    SpdyStream* spdy_stream) {
+  spdy_session->ActivateStream(spdy_stream);
+}
+
+// static
+SpdyIOBuffer* SpdySession::SpdyIOBufferProducer::CreateIOBuffer(
+    SpdyFrame* frame,
+    RequestPriority priority,
+    SpdyStream* stream) {
+  size_t size = frame->length() + SpdyFrame::kHeaderSize;
+  DCHECK_GT(size, 0u);
+
+  // TODO(mbelshe): We have too much copying of data here.
+  IOBufferWithSize* buffer = new IOBufferWithSize(size);
+  memcpy(buffer->data(), frame->data(), size);
+
+  return new SpdyIOBuffer(buffer, size, priority, stream);
+}
+
+// static
 void SpdySession::set_default_protocol(NextProto default_protocol) {
   g_default_protocol = default_protocol;
 }
 
 // static
 void SpdySession::set_max_concurrent_streams(size_t value) {
   g_max_concurrent_stream_limit = value;
 }
 
 // static
@@ skipping 155 matching lines @@
 
   SSLInfo ssl_info;
   bool was_npn_negotiated;
   NextProto protocol_negotiated = kProtoUnknown;
   if (!GetSSLInfo(&ssl_info, &was_npn_negotiated, &protocol_negotiated))
     return true;   // This is not a secure session, so all domains are okay.
 
   return !ssl_info.client_cert_sent && ssl_info.cert->VerifyNameMatch(domain);
 }
 
+void SpdySession::SetStreamHasWriteAvailable(SpdyStream* stream,
+                                             SpdyIOBufferProducer* producer) {
+  write_queue_.push(producer);
+  stream_producers_[producer] = stream;
+  WriteSocketLater();
+}
+
 int SpdySession::GetPushStream(
     const GURL& url,
     scoped_refptr<SpdyStream>* stream,
     const BoundNetLog& stream_net_log) {
   CHECK_NE(state_, CLOSED);
 
   *stream = NULL;
 
   // Don't allow access to secure push streams over an unauthenticated, but
   // encrypted SSL socket.
@@ skipping 18 matching lines @@
   return 0;
 }
 
 int SpdySession::CreateStream(
     const GURL& url,
     RequestPriority priority,
     scoped_refptr<SpdyStream>* spdy_stream,
     const BoundNetLog& stream_net_log,
     const CompletionCallback& callback) {
   if (!max_concurrent_streams_ ||
-      active_streams_.size() < max_concurrent_streams_) {
+      (active_streams_.size() + created_streams_.size()
+       < max_concurrent_streams_)) {
     return CreateStreamImpl(url, priority, spdy_stream, stream_net_log);

[ramant (doing other things), 2012/06/26 23:30:25]

nit: "When expressions are wrapped, the operator should be on the end of the
broken line, not the start of the new line"

http://dev.chromium.org/developers/coding-style

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

   }
 
   stalled_streams_++;
   net_log().AddEvent(NetLog::TYPE_SPDY_SESSION_STALLED_MAX_STREAMS);
   create_stream_queues_[priority].push(
       PendingCreateStream(url, priority, spdy_stream,
                           stream_net_log, callback));
   return ERR_IO_PENDING;
 }
 
@@ skipping 68 matching lines @@
     CloseSessionOnError(
         static_cast<net::Error>(certificate_error_code_),
         true,
         "Tried to create SPDY stream for secure content over an "
         "unauthenticated session.");
     return ERR_SPDY_PROTOCOL_ERROR;
   }
 
   const std::string& path = url.PathForRequest();
 
-  const SpdyStreamId stream_id = GetNewStreamId();
-
   *spdy_stream = new SpdyStream(this,
-                                stream_id,
                                 false,
                                 stream_net_log);
   const scoped_refptr<SpdyStream>& stream = *spdy_stream;
 
   stream->set_priority(priority);
   stream->set_path(path);
   stream->set_send_window_size(initial_send_window_size_);
   stream->set_recv_window_size(initial_recv_window_size_);
-  ActivateStream(stream);
+  created_streams_.insert(stream);
 
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyPriorityCount",
       static_cast<int>(priority), 0, 10, 11);
 
   // TODO(mbelshe): Optimize memory allocations
 
-  DCHECK_EQ(active_streams_[stream_id].get(), stream.get());
   return OK;
 }
 
 bool SpdySession::NeedsCredentials() const {
   if (!is_secure_)
     return false;
   SSLClientSocket* ssl_socket = GetSSLClientSocket();
   if (ssl_socket->GetNegotiatedProtocol() < kProtoSPDY3)
     return false;
   return ssl_socket->WasDomainBoundCertSent();
 }
 
 void SpdySession::AddPooledAlias(const HostPortProxyPair& alias) {
   pooled_aliases_.insert(alias);
 }
 
 int SpdySession::GetProtocolVersion() const {
   DCHECK(buffered_spdy_framer_.get());
   return buffered_spdy_framer_->protocol_version();
 }
 
-int SpdySession::WriteSynStream(
+SpdySynStreamControlFrame* SpdySession::CreateSynStream(
     SpdyStreamId stream_id,
     RequestPriority priority,
     uint8 credential_slot,
     SpdyControlFlags flags,
     const linked_ptr<SpdyHeaderBlock>& headers) {
-  // Find our stream
-  if (!IsStreamActive(stream_id))
-    return ERR_INVALID_SPDY_STREAM;
+  CHECK(IsStreamActive(stream_id));
   const scoped_refptr<SpdyStream>& stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
 
   SendPrefacePingIfNoneInFlight();
 
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdySynStreamControlFrame> syn_frame(
       buffered_spdy_framer_->CreateSynStream(
           stream_id, 0,
           ConvertRequestPriorityToSpdyPriority(priority, GetProtocolVersion()),
-          credential_slot, flags, false, headers.get()));
-  // We enqueue all SYN_STREAM frames at the same priority to ensure
-  // that we do not send them out-of-order.
-  // http://crbug.com/111708
-  QueueFrame(syn_frame.get(), HIGHEST, stream);
+          credential_slot, flags, true, headers.get()));
 
   base::StatsCounter spdy_requests("spdy.requests");
   spdy_requests.Increment();
   streams_initiated_count_++;
 
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SYN_STREAM,
         base::Bind(&NetLogSpdySynCallback, headers.get(), flags, stream_id, 0));
   }
 
-  return ERR_IO_PENDING;
+  return syn_frame.release();
 }
 
-int SpdySession::WriteCredentialFrame(const std::string& origin,
-                                      SSLClientCertType type,
-                                      const std::string& key,
-                                      const std::string& cert,
-                                      RequestPriority priority) {
+SpdyCredentialControlFrame* SpdySession::CreateCredentialFrame(
+    const std::string& origin,
+    SSLClientCertType type,
+    const std::string& key,
+    const std::string& cert,
+    RequestPriority priority) {
   DCHECK(is_secure_);
   unsigned char secret[32];  // 32 bytes from the spec
   GetSSLClientSocket()->ExportKeyingMaterial("SPDY certificate proof",
                                              true, origin,
                                              secret, arraysize(secret));
 
   // Convert the key string into a vector<unit8>
   std::vector<uint8> key_data;
   for (size_t i = 0; i < key.length(); i++) {
     key_data.push_back(key[i]);
@@ skipping 21 matching lines @@
   SpdyCredential credential;
   GURL origin_url(origin);
   credential.slot =
       credential_state_.SetHasCredential(origin_url);
   credential.certs.push_back(cert);
   credential.proof.assign(proof.begin(), proof.end());
 
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyCredentialControlFrame> credential_frame(
       buffered_spdy_framer_->CreateCredentialFrame(credential));
-  // We enqueue all SYN_STREAM frames at the same priority to ensure
-  // that we do not send them out-of-order, which means that we need
-  // to enqueue all CREDENTIAL frames at this priority to ensure that
-  // they are sent *before* the SYN_STREAM that references them.
-  // http://crbug.com/111708
-  QueueFrame(credential_frame.get(), HIGHEST, NULL);
 
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SEND_CREDENTIAL,
         base::Bind(&NetLogSpdyCredentialCallback, credential.slot, &origin));
   }
-  return ERR_IO_PENDING;
+  return credential_frame.release();
 }
 
-int SpdySession::WriteStreamData(SpdyStreamId stream_id,
-                                 net::IOBuffer* data, int len,
-                                 SpdyDataFlags flags) {
+SpdyDataFrame* SpdySession::CreateDataFrame(SpdyStreamId stream_id,
+                                            net::IOBuffer* data, int len,
+                                            SpdyDataFlags flags) {
   // Find our stream
   CHECK(IsStreamActive(stream_id));
   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
 
   if (len > kMaxSpdyFrameChunkSize) {
     len = kMaxSpdyFrameChunkSize;
     flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
   }
 
   // Obey send window size of the stream if flow control is enabled.
   if (flow_control_) {
     if (stream->send_window_size() <= 0) {
       // Because we queue frames onto the session, it is possible that
       // a stream was not flow controlled at the time it attempted the
       // write, but when we go to fulfill the write, it is now flow
       // controlled.  This is why we need the session to mark the stream
       // as stalled - because only the session knows for sure when the
       // stall occurs.
       stream->set_stalled_by_flow_control(true);
       net_log().AddEvent(
           NetLog::TYPE_SPDY_SESSION_STALLED_ON_SEND_WINDOW,
           NetLog::IntegerCallback("stream_id", stream_id));
-      return ERR_IO_PENDING;
+      return NULL;
     }
     int new_len = std::min(len, stream->send_window_size());
     if (new_len < len) {
       len = new_len;
       flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
     }
     stream->DecreaseSendWindowSize(len);
   }
 
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SEND_DATA,
         base::Bind(&NetLogSpdyDataCallback, stream_id, len, flags));
   }
 
   // Send PrefacePing for DATA_FRAMEs with nonzero payload size.
   if (len > 0)
     SendPrefacePingIfNoneInFlight();
 
   // TODO(mbelshe): reduce memory copies here.
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyDataFrame> frame(
       buffered_spdy_framer_->CreateDataFrame(
           stream_id, data->data(), len, flags));
-  QueueFrame(frame.get(), stream->priority(), stream);
 
-  return ERR_IO_PENDING;
+  return frame.release();
 }
 
 void SpdySession::CloseStream(SpdyStreamId stream_id, int status) {
+  DCHECK_NE(0u, stream_id);
   // TODO(mbelshe): We should send a RST_STREAM control frame here
   //                so that the server can cancel a large send.
 
   DeleteStream(stream_id, status);
 }
 
+void SpdySession::CloseCreatedStream(SpdyStream* stream, int status) {
+  DCHECK_EQ(0u, stream->stream_id());
+  created_streams_.erase(scoped_refptr<SpdyStream>(stream));
+}
+
 void SpdySession::ResetStream(SpdyStreamId stream_id,
                               SpdyStatusCodes status,
                               const std::string& description) {
   net_log().AddEvent(
       NetLog::TYPE_SPDY_SESSION_SEND_RST_STREAM,
       base::Bind(&NetLogSpdyRstCallback, stream_id, status, &description));
 
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyRstStreamControlFrame> rst_frame(
       buffered_spdy_framer_->CreateRstStream(stream_id, status));
 
   // Default to lowest priority unless we know otherwise.
   RequestPriority priority = net::IDLE;
   if(IsStreamActive(stream_id)) {
     scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
     priority = stream->priority();
   }
-  QueueFrame(rst_frame.get(), priority, NULL);
+  QueueFrame(rst_frame.release(), priority);
   RecordProtocolErrorHistogram(
       static_cast<SpdyProtocolErrorDetails>(status + STATUS_CODE_INVALID));
   DeleteStream(stream_id, ERR_SPDY_PROTOCOL_ERROR);
 }
 
 bool SpdySession::IsStreamActive(SpdyStreamId stream_id) const {
   return ContainsKey(active_streams_, stream_id);
 }
 
 LoadState SpdySession::GetLoadState() const {
@@ skipping 167 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).
   DCHECK(buffered_spdy_framer_.get());
-  while (in_flight_write_.buffer() || !queue_.empty()) {
+  while (in_flight_write_.buffer() || !write_queue_.empty()) {
     if (!in_flight_write_.buffer()) {
-      // Grab the next SpdyFrame to send.
-      SpdyIOBuffer next_buffer = queue_.top();
-      queue_.pop();
+      // Grab the next SpdyBuffer to send.
+      scoped_ptr<SpdyIOBufferProducer> producer(write_queue_.top());
+      scoped_ptr<SpdyIOBuffer> buffer(producer->ProduceNextBuffer(this));
+      stream_producers_.erase(producer.get());
+      write_queue_.pop();
+      // It is possible that a stream had data to write, but a
+      // WINDOW_UPDATE frame has been received which made that
+      // stream no longer writable.
+      // TODO(rch): consider handling that case by removing the
+      // stream from the writable queue?
+      if (buffer == NULL)
+        continue;
 
-      // 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.
-      SpdyFrame uncompressed_frame(next_buffer.buffer()->data(), false);
-      size_t size;
-      if (buffered_spdy_framer_->IsCompressible(uncompressed_frame)) {
-        DCHECK(uncompressed_frame.is_control_frame());
-        const SpdyControlFrame* uncompressed_control_frame =
-            reinterpret_cast<const SpdyControlFrame*>(&uncompressed_frame);
-        scoped_ptr<SpdyFrame> compressed_frame(
-            buffered_spdy_framer_->CompressControlFrame(
-                *uncompressed_control_frame));
-        if (!compressed_frame.get()) {
-          RecordProtocolErrorHistogram(
-              PROTOCOL_ERROR_SPDY_COMPRESSION_FAILURE);
-          CloseSessionOnError(
-              net::ERR_SPDY_PROTOCOL_ERROR, true, "SPDY Compression failure.");
-          return;
-        }
-
-        size = compressed_frame->length() + SpdyFrame::kHeaderSize;
-
-        DCHECK_GT(size, 0u);
-
-        if (uncompressed_control_frame->type() == SYN_STREAM) {
-          int uncompressed_size = uncompressed_control_frame->length();
-          int compressed_size = compressed_frame->length();
-          // Make sure we avoid early decimal truncation.
-          int compression_pct = 100 - (100* compressed_size)/uncompressed_size;
-          UMA_HISTOGRAM_PERCENTAGE("Net.SpdySynStreamCompressionPercentage",
-                                   compression_pct);
-        }
-
-        // 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, HIGHEST,
-                                        next_buffer.stream());
-      } else {
-        size = uncompressed_frame.length() + SpdyFrame::kHeaderSize;
-        in_flight_write_ = next_buffer;
-      }
+      in_flight_write_ = *buffer;
     } else {
       DCHECK(in_flight_write_.buffer()->BytesRemaining());
     }
 
     write_pending_ = true;
     int rv = connection_->socket()->Write(
         in_flight_write_.buffer(),
         in_flight_write_.buffer()->BytesRemaining(),
         base::Bind(&SpdySession::OnWriteComplete, base::Unretained(this)));
     if (rv == net::ERR_IO_PENDING)
@@ skipping 33 matching lines @@
   while (!active_streams_.empty()) {
     ActiveStreamMap::iterator it = active_streams_.begin();
     const scoped_refptr<SpdyStream>& stream = it->second;
     DCHECK(stream);
     std::string description = base::StringPrintf(
         "ABANDONED (stream_id=%d): ", stream->stream_id()) + stream->path();
     stream->LogStreamError(status, description);
     DeleteStream(stream->stream_id(), status);
   }
 
+  while (!created_streams_.empty()) {
+    CreatedStreamSet::iterator it = created_streams_.begin();
+    const scoped_refptr<SpdyStream>& stream = *it;
+    std::string description = base::StringPrintf(
+        "ABANDONED (stream_id=%d): ", stream->stream_id()) + stream->path();
+    stream->LogStreamError(status, description);
+    stream->OnClose(status);

[ramant (doing other things), 2012/06/26 23:30:25]

nit: consider sharing of code between lines 976-979 with 986-988.

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

+    created_streams_.erase(it);
+  }
+
   // We also need to drain the queue.
-  while (queue_.size())
-    queue_.pop();
+  while (!write_queue_.empty()) {
+    SpdyIOBufferProducer* producer = write_queue_.top();
+    stream_producers_.erase(producer);
+    delete producer;
+    write_queue_.pop();
+  }
 }
 
 int SpdySession::GetNewStreamId() {
   int id = stream_hi_water_mark_;
   stream_hi_water_mark_ += 2;
   if (stream_hi_water_mark_ > 0x7fff)
     stream_hi_water_mark_ = 1;
   return id;
 }
 
-void SpdySession::QueueFrame(SpdyFrame* frame,
-                             RequestPriority priority,
-                             SpdyStream* stream) {
-  int length = SpdyFrame::kHeaderSize + frame->length();
-  IOBuffer* buffer = new IOBuffer(length);
-  memcpy(buffer->data(), frame->data(), length);
-  queue_.push(SpdyIOBuffer(buffer, length, priority, stream));
-
-  WriteSocketLater();
-}
-
 void SpdySession::CloseSessionOnError(net::Error err,
                                       bool remove_from_pool,
                                       const std::string& description) {
   // Closing all streams can have a side-effect of dropping the last reference
   // to |this|.  Hold a reference through this function.
   scoped_refptr<SpdySession> self(this);
 
   DCHECK_LT(err, OK);
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_CLOSE,
@@ skipping 69 matching lines @@
   return connection_->socket()->GetPeerAddress(address);
 }
 
 int SpdySession::GetLocalAddress(IPEndPoint* address) const {
   if (!connection_->socket())
     return ERR_SOCKET_NOT_CONNECTED;
 
   return connection_->socket()->GetLocalAddress(address);
 }
 
+class SimpleSpdyIOBufferProducer : public SpdySession::SpdyIOBufferProducer {
+ public:
+  SimpleSpdyIOBufferProducer(SpdyFrame* frame,
+                          RequestPriority priority)
+      : frame_(frame),

[ramant (doing other things), 2012/06/26 23:30:25]

nit: indent "RequestPriority ..."

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

+        priority_(priority) {
+  }
+
+  virtual RequestPriority GetPriority() const OVERRIDE {
+    return priority_;
+  }
+
+  virtual SpdyIOBuffer* ProduceNextBuffer(SpdySession* session) {
+    return SpdySession::SpdyIOBufferProducer::CreateIOBuffer(
+        frame_, priority_, NULL);
+  }
+
+ private:
+  SpdyFrame* frame_;
+  RequestPriority priority_;
+};
+
+void SpdySession::QueueFrame(SpdyFrame* frame,
+                             RequestPriority priority) {
+  SimpleSpdyIOBufferProducer* producer
+      = new SimpleSpdyIOBufferProducer(frame, priority);
+  write_queue_.push(producer);

[ramant (doing other things), 2012/06/26 23:30:25]

nit: " = " should be in the previous line.

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

+  WriteSocketLater();
+}
+
 void SpdySession::ActivateStream(SpdyStream* stream) {
+  if (stream->stream_id() == 0) {
+    stream->set_stream_id(GetNewStreamId());
+    created_streams_.erase(scoped_refptr<SpdyStream>(stream));
+  }
   const SpdyStreamId id = stream->stream_id();
   DCHECK(!IsStreamActive(id));
 
   active_streams_[id] = stream;
 }
 
 void SpdySession::DeleteStream(SpdyStreamId id, int status) {
   // For push streams, if they are being deleted normally, we leave
   // the stream in the unclaimed_pushed_streams_ list.  However, if
   // the stream is errored out, clean it up entirely.
   if (status != OK) {
     PushedStreamMap::iterator it;
     for (it = unclaimed_pushed_streams_.begin();
          it != unclaimed_pushed_streams_.end(); ++it) {
       scoped_refptr<SpdyStream> curr = it->second;
       if (id == curr->stream_id()) {
         unclaimed_pushed_streams_.erase(it);
         break;
       }
     }
   }
 
   // The stream might have been deleted.
   ActiveStreamMap::iterator it2 = active_streams_.find(id);
   if (it2 == active_streams_.end())
     return;
 
+  // Possibly remove from the write queue.
+  WriteQueue old = write_queue_;
+  write_queue_ = WriteQueue();
+  while (!old.empty()) {
+    SpdyIOBufferProducer* producer = old.top();
+    StreamProducerMap::iterator it = stream_producers_.find(producer);
+    if (it == stream_producers_.end() || it->second->stream_id() != id)
+      write_queue_.push(producer);
+    else
+      delete producer;
+    old.pop();
+  }
+
   // 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 (spdy_session_pool_) {
@@ skipping 95 matching lines @@
       value);
   received_settings_ = true;
 
   // Log the setting.
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_RECV_SETTING,
       base::Bind(&NetLogSpdySettingCallback,
                  id, static_cast<SpdySettingsFlags>(flags), value));
 }
 
+void SpdySession::OnControlFrameCompressed(
+    const SpdyControlFrame& uncompressed_frame,
+    const SpdyControlFrame& compressed_frame) {
+  if (uncompressed_frame.type() == SYN_STREAM) {
+    int uncompressed_size = uncompressed_frame.length();
+    int compressed_size = compressed_frame.length();
+    // Make sure we avoid early decimal truncation.
+    int compression_pct = 100 - (100* compressed_size)/uncompressed_size;
+    UMA_HISTOGRAM_PERCENTAGE("Net.SpdySynStreamCompressionPercentage",
+                             compression_pct);
+  }
+}
+
+
 bool SpdySession::Respond(const SpdyHeaderBlock& headers,
                           const scoped_refptr<SpdyStream> stream) {
   int rv = OK;
   rv = stream->OnResponseReceived(headers);
   if (rv < 0) {
     DCHECK_NE(rv, ERR_IO_PENDING);
     const SpdyStreamId stream_id = stream->stream_id();
     DeleteStream(stream_id, rv);
     return false;
   }
@@ skipping 80 matching lines @@
   }
 
   // There should not be an existing pushed stream with the same path.
   PushedStreamMap::iterator it = unclaimed_pushed_streams_.find(url);
   if (it != unclaimed_pushed_streams_.end()) {
     ResetStream(stream_id, PROTOCOL_ERROR,
                 "Received duplicate pushed stream with url: " + url);
     return;
   }
 
-  scoped_refptr<SpdyStream> stream(
-      new SpdyStream(this, stream_id, true, net_log_));
+  scoped_refptr<SpdyStream> stream(new SpdyStream(this, true, net_log_));
+  stream->set_stream_id(stream_id);
 
   stream->set_path(gurl.PathForRequest());
   stream->set_send_window_size(initial_send_window_size_);
   stream->set_recv_window_size(initial_recv_window_size_);
 
   unclaimed_pushed_streams_[url] = stream;
 
   ActivateStream(stream);
   stream->set_response_received();
 
   // Parse the headers.
   if (!Respond(*headers, stream))
     return;
 
   base::StatsCounter push_requests("spdy.pushed_streams");
   push_requests.Increment();
 }
 
 void SpdySession::OnSynReply(const SpdySynReplyControlFrame& frame,
                              const linked_ptr<SpdyHeaderBlock>& headers) {
   SpdyStreamId stream_id = frame.stream_id();
-
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SYN_REPLY,
         base::Bind(&NetLogSpdySynCallback,
                    headers.get(), static_cast<SpdyControlFlags>(frame.flags()),
                    stream_id, 0));
   }
 
   if (!IsStreamActive(stream_id)) {
     // NOTE:  it may just be that the stream was cancelled.
@@ skipping 163 matching lines @@
   CHECK_EQ(stream->stream_id(), stream_id);
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_SENT_WINDOW_UPDATE,
       base::Bind(&NetLogSpdyWindowUpdateCallback,
                  stream_id, delta_window_size));
 
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyWindowUpdateControlFrame> window_update_frame(
       buffered_spdy_framer_->CreateWindowUpdate(stream_id, delta_window_size));
-  QueueFrame(window_update_frame.get(), stream->priority(), NULL);
+  QueueFrame(window_update_frame.release(), stream->priority());
 }
 
 // Given a cwnd that we would have sent to the server, modify it based on the
 // field trial policy.
 uint32 ApplyCwndFieldTrialPolicy(int cwnd) {
   base::FieldTrial* trial = base::FieldTrialList::Find("SpdyCwnd");
   if (!trial) {
       LOG(WARNING) << "Could not find \"SpdyCwnd\" in FieldTrialList";
       return cwnd;
   }
@@ skipping 19 matching lines @@
     // Create a new settings frame notifying the sever of our
     // max_concurrent_streams_ and initial window size.
     settings_map[SETTINGS_MAX_CONCURRENT_STREAMS] =
         SettingsFlagsAndValue(SETTINGS_FLAG_NONE, kInitialMaxConcurrentStreams);
     if (GetProtocolVersion() > 2 &&
         initial_recv_window_size_ != kSpdyStreamInitialWindowSize) {
       settings_map[SETTINGS_INITIAL_WINDOW_SIZE] =
           SettingsFlagsAndValue(SETTINGS_FLAG_NONE, initial_recv_window_size_);
     }
     sent_settings_ = true;
     SendSettings(settings_map);

[ramant (doing other things), 2012/06/26 23:30:25]

nit: delete "sent_settings_ = true;"?

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

   }
 
   // Next notify the server about the settings they have previously
   // told us to use when communicating with them.
   const SettingsMap& settings_map =
       http_server_properties_->GetSpdySettings(host_port_pair());
   if (settings_map.empty())
     return;
 
   // Record Histogram Data and Apply the SpdyCwnd FieldTrial if applicable.
@@ skipping 12 matching lines @@
   const SettingsMap& settings_map_new =
       http_server_properties_->GetSpdySettings(host_port_pair());
   for (SettingsMap::const_iterator i = settings_map_new.begin(),
            end = settings_map_new.end(); i != end; ++i) {
     const SpdySettingsIds new_id = i->first;
     const uint32 new_val = i->second.second;
     HandleSetting(new_id, new_val);
   }
 
   sent_settings_ = true;
   SendSettings(settings_map_new);

[ramant (doing other things), 2012/06/26 23:30:25]

nit: delete "sent_settings_ = true;"?

[Ryan Hamilton, 2012/06/27 16:58:28]

Done.

 }
 
 
 void SpdySession::SendSettings(const SettingsMap& settings) {
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_SEND_SETTINGS,
       base::Bind(&NetLogSpdySettingsCallback, &settings));
 
   // Create the SETTINGS frame and send it.
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdySettingsControlFrame> settings_frame(
       buffered_spdy_framer_->CreateSettings(settings));
-  QueueFrame(settings_frame.get(), HIGHEST, NULL);
+  sent_settings_ = true;
+  QueueFrame(settings_frame.release(), HIGHEST);
 }
 
 void SpdySession::HandleSetting(uint32 id, uint32 value) {
   switch (id) {
     case SETTINGS_MAX_CONCURRENT_STREAMS:
       max_concurrent_streams_ = std::min(static_cast<size_t>(value),
                                          g_max_concurrent_stream_limit);
       ProcessPendingCreateStreams();
       break;
     case SETTINGS_INITIAL_WINDOW_SIZE:
@@ skipping 14 matching lines @@
   }
 }
 
 void SpdySession::UpdateStreamsSendWindowSize(int32 delta_window_size) {
   ActiveStreamMap::iterator it;
   for (it = active_streams_.begin(); it != active_streams_.end(); ++it) {
     const scoped_refptr<SpdyStream>& stream = it->second;
     DCHECK(stream);
     stream->AdjustSendWindowSize(delta_window_size);
   }
+
+  CreatedStreamSet::iterator i;
+  for (i = created_streams_.begin(); i != created_streams_.end(); i++) {
+    const scoped_refptr<SpdyStream>& stream = *i;
+    stream->AdjustSendWindowSize(delta_window_size);
+  }
 }
 
 void SpdySession::SendPrefacePingIfNoneInFlight() {
   if (pings_in_flight_ || !g_enable_ping_based_connection_checking)
     return;
 
   base::TimeTicks now = base::TimeTicks::Now();
   // If there is no activity in the session, then send a preface-PING.
   if ((now - last_activity_time_) > connection_at_risk_of_loss_time_)
     SendPrefacePing();
 }
 
 void SpdySession::SendPrefacePing() {
   WritePingFrame(next_ping_id_);
 }
 
 void SpdySession::WritePingFrame(uint32 unique_id) {
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyPingControlFrame> ping_frame(
       buffered_spdy_framer_->CreatePingFrame(next_ping_id_));
-  QueueFrame(ping_frame.get(), HIGHEST, NULL);
+  QueueFrame(ping_frame.release(), HIGHEST);
 
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_PING,
         base::Bind(&NetLogSpdyPingCallback, next_ping_id_, "sent"));
   }
   if (unique_id % 2 != 0) {
     next_ping_id_ += 2;
     ++pings_in_flight_;
     PlanToCheckPingStatus();
@@ skipping 142 matching lines @@
 SSLClientSocket* SpdySession::GetSSLClientSocket() const {
   if (!is_secure_)
     return NULL;
   SSLClientSocket* ssl_socket =
       reinterpret_cast<SSLClientSocket*>(connection_->socket());
   DCHECK(ssl_socket);
   return ssl_socket;
 }
 
 }  // namespace net