[SPDY] Fix SpdySession's handling of SYN_REPLY 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 <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
@@ skipping 293 matching lines @@
 void SpdyStreamRequest::Reset() {
   type_ = SPDY_BIDIRECTIONAL_STREAM;
   session_ = NULL;
   stream_.reset();
   url_ = GURL();
   priority_ = MINIMUM_PRIORITY;
   net_log_ = BoundNetLog();
   callback_.Reset();
 }
 
+SpdySession::ActiveStreamInfo::ActiveStreamInfo()
+    : stream(NULL),
+      waiting_for_syn_reply(false) {}
+
+SpdySession::ActiveStreamInfo::ActiveStreamInfo(SpdyStream* stream)
+    : stream(stream),
+      waiting_for_syn_reply(stream->type() != SPDY_PUSH_STREAM) {}
+
+SpdySession::ActiveStreamInfo::~ActiveStreamInfo() {}
+
+SpdySession::PushedStreamInfo::PushedStreamInfo() : stream_id(0) {}
+
+SpdySession::PushedStreamInfo::PushedStreamInfo(
+    SpdyStreamId stream_id,
+    base::TimeTicks creation_time)
+    : stream_id(stream_id),
+      creation_time(creation_time) {}
+
+SpdySession::PushedStreamInfo::~PushedStreamInfo() {}
+
 SpdySession::SpdySession(const SpdySessionKey& spdy_session_key,
                          SpdySessionPool* spdy_session_pool,
                          HttpServerProperties* http_server_properties,
                          bool verify_domain_authentication,
                          bool enable_sending_initial_settings,
                          bool enable_credential_frames,
                          bool enable_compression,
                          bool enable_ping_based_connection_checking,
                          NextProto default_protocol,
                          size_t stream_initial_recv_window_size,
@@ skipping 374 matching lines @@
 }
 
 scoped_ptr<SpdyFrame> SpdySession::CreateSynStream(
     SpdyStreamId stream_id,
     RequestPriority priority,
     uint8 credential_slot,
     SpdyControlFlags flags,
     const SpdyHeaderBlock& headers) {
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   CHECK(it != active_streams_.end());
-  CHECK_EQ(it->second->stream_id(), stream_id);
+  CHECK_EQ(it->second.stream->stream_id(), stream_id);
 
   SendPrefacePingIfNoneInFlight();
 
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyFrame> syn_frame(
       buffered_spdy_framer_->CreateSynStream(
           stream_id, 0,
           ConvertRequestPriorityToSpdyPriority(priority, GetProtocolVersion()),
           credential_slot, flags, enable_compression_, &headers));
 
@@ skipping 46 matching lines @@
   }
   return OK;
 }
 
 scoped_ptr<SpdyBuffer> SpdySession::CreateDataBuffer(SpdyStreamId stream_id,
                                                      IOBuffer* data,
                                                      int len,
                                                      SpdyDataFlags flags) {
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   CHECK(it != active_streams_.end());
-  SpdyStream* stream = it->second;
+  SpdyStream* stream = it->second.stream;
   CHECK_EQ(stream->stream_id(), stream_id);
 
   if (len < 0) {
     NOTREACHED();
     return scoped_ptr<SpdyBuffer>();
   }
 
   int effective_len = std::min(len, kMaxSpdyFrameChunkSize);
 
   bool send_stalled_by_stream =
@@ skipping 107 matching lines @@
 
   // TODO(mbelshe): We should send a RST_STREAM control frame here
   //                so that the server can cancel a large send.
 
   // 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) {
     for (PushedStreamMap::iterator it = unclaimed_pushed_streams_.begin();
          it != unclaimed_pushed_streams_.end(); ++it) {
-      if (stream_id == it->second.first->stream_id()) {
+      if (stream_id == it->second.stream_id) {
         unclaimed_pushed_streams_.erase(it);
         break;
       }
     }
   }
 
   // The stream might have been deleted.
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end())
     return;
 
-  scoped_ptr<SpdyStream> owned_stream(it->second);
+  scoped_ptr<SpdyStream> owned_stream(it->second.stream);
   active_streams_.erase(it);
 
   DeleteStream(owned_stream.Pass(), status);
 }
 
 void SpdySession::CloseCreatedStream(
     const base::WeakPtr<SpdyStream>& stream, int status) {
   DCHECK_EQ(0u, stream->stream_id());
 
   scoped_ptr<SpdyStream> owned_stream(stream.get());
@@ skipping 307 matching lines @@
   // The loops below are carefully written to avoid problems with
   // streams closing other streams.
 
   while (true) {
     ActiveStreamMap::iterator it =
         active_streams_.lower_bound(last_good_stream_id + 1);
     if (it == active_streams_.end())
       break;
     SpdyStreamId stream_id = it->first;
     streams_abandoned_count_++;
-    LogAbandonedStream(it->second, status);
+    LogAbandonedStream(it->second.stream, status);
     CloseActiveStream(stream_id, status);
   }
 
   while (!created_streams_.empty()) {
     CreatedStreamSet::iterator it = created_streams_.begin();
     LogAbandonedStream(*it, status);
     CloseCreatedStream((*it)->GetWeakPtr(), status);
   }
 
   write_queue_.RemovePendingWritesForStreamsAfter(last_good_stream_id);
@@ skipping 175 matching lines @@
   stream->set_stream_id(GetNewStreamId());
   scoped_ptr<SpdyStream> owned_stream(stream);
   created_streams_.erase(stream);
   return owned_stream.Pass();
 }
 
 void SpdySession::InsertActivatedStream(scoped_ptr<SpdyStream> stream) {
   SpdyStreamId stream_id = stream->stream_id();
   DCHECK_NE(stream_id, 0u);
   std::pair<ActiveStreamMap::iterator, bool> result =
-      active_streams_.insert(std::make_pair(stream_id, stream.get()));
+      active_streams_.insert(
+          std::make_pair(stream_id, ActiveStreamInfo(stream.get())));
   if (result.second) {
     ignore_result(stream.release());
   } else {
     NOTREACHED();
   }
 }
 
 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
@@ skipping 18 matching lines @@
     spdy_session_pool_ = NULL;
     pool->Remove(make_scoped_refptr(this));
   }
 }
 
 base::WeakPtr<SpdyStream> SpdySession::GetActivePushStream(
     const std::string& path) {
   base::StatsCounter used_push_streams("spdy.claimed_push_streams");
 
   PushedStreamMap::iterator it = unclaimed_pushed_streams_.find(path);
-  if (it != unclaimed_pushed_streams_.end()) {
-    net_log_.AddEvent(NetLog::TYPE_SPDY_STREAM_ADOPTED_PUSH_STREAM);
-    base::WeakPtr<SpdyStream> stream = it->second.first->GetWeakPtr();
-    unclaimed_pushed_streams_.erase(it);
-    used_push_streams.Increment();
-    return stream;
+  if (it == unclaimed_pushed_streams_.end())
+    return base::WeakPtr<SpdyStream>();
+
+  SpdyStreamId stream_id = it->second.stream_id;
+  unclaimed_pushed_streams_.erase(it);
+
+  ActiveStreamMap::iterator it2 = active_streams_.find(stream_id);
+  if (it2 == active_streams_.end()) {
+    NOTREACHED();
+    return base::WeakPtr<SpdyStream>();
   }
-  return base::WeakPtr<SpdyStream>();
+
+  net_log_.AddEvent(NetLog::TYPE_SPDY_STREAM_ADOPTED_PUSH_STREAM);
+  used_push_streams.Increment();
+  return it2->second.stream->GetWeakPtr();
 }
 
 bool SpdySession::GetSSLInfo(SSLInfo* ssl_info,
                              bool* was_npn_negotiated,
                              NextProto* protocol_negotiated) {
   *was_npn_negotiated = connection_->socket()->WasNpnNegotiated();
   *protocol_negotiated = connection_->socket()->GetNegotiatedProtocol();
   return connection_->socket()->GetSSLInfo(ssl_info);
 }
 
@@ skipping 18 matching lines @@
       "SPDY_ERROR error_code: %d.", error_code);
   CloseSessionOnError(ERR_SPDY_PROTOCOL_ERROR, true, description);
 }
 
 void SpdySession::OnStreamError(SpdyStreamId stream_id,
                                 const std::string& description) {
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   // We still want to reset the stream even if we don't know anything
   // about it.
   RequestPriority priority =
-      (it == active_streams_.end()) ? IDLE : it->second->priority();
+      (it == active_streams_.end()) ? IDLE : it->second.stream->priority();
   ResetStream(stream_id, priority, RST_STREAM_PROTOCOL_ERROR, description);
 }
 
 void SpdySession::OnStreamFrameData(SpdyStreamId stream_id,
                                     const char* data,
                                     size_t len,
                                     bool fin) {
   DCHECK_LT(len, 1u << 24);
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_RECV_DATA,
         base::Bind(&NetLogSpdyDataCallback, stream_id, len, fin));
   }
 
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
 
   // By the time data comes in, the stream may already be inactive.
   if (it == active_streams_.end())
     return;
 
+  SpdyStream* stream = it->second.stream;
+  CHECK_EQ(stream->stream_id(), stream_id);
+
+  if (it->second.waiting_for_syn_reply) {
+    const std::string& error = "Data received before SYN_REPLY.";
+    stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
+    ResetStream(stream_id, stream->priority(),
+                RST_STREAM_PROTOCOL_ERROR, error);
+    return;
+  }
+
   scoped_ptr<SpdyBuffer> buffer;
   if (data) {
     DCHECK_GT(len, 0u);
     buffer.reset(new SpdyBuffer(data, len));
 
     if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
       DecreaseRecvWindowSize(static_cast<int32>(len));
       buffer->AddConsumeCallback(
           base::Bind(&SpdySession::OnReadBufferConsumed,
                      weak_factory_.GetWeakPtr()));
     }
   } else {
     DCHECK_EQ(len, 0u);
   }
-  it->second->OnDataReceived(buffer.Pass());
+  stream->OnDataReceived(buffer.Pass());
 }
 
 void SpdySession::OnSettings(bool clear_persisted) {
   if (clear_persisted)
     http_server_properties_->ClearSpdySettings(host_port_pair());
 
   if (net_log_.IsLoggingAllEvents()) {
     net_log_.AddEvent(
         NetLog::TYPE_SPDY_SESSION_RECV_SETTINGS,
         base::Bind(&NetLogSpdySettingsCallback, host_port_pair(),
@@ skipping 109 matching lines @@
   // is trusted explicitly via the --trusted-spdy-proxy switch.
   if (trusted_spdy_proxy_.Equals(host_port_pair())) {
     // Disallow pushing of HTTPS content.
     if (gurl.SchemeIs("https")) {
       ResetStream(stream_id, request_priority, RST_STREAM_REFUSED_STREAM,
                   base::StringPrintf(
                       "Rejected push of Cross Origin HTTPS content %d",
                       associated_stream_id));
     }
   } else {
-    GURL associated_url(associated_it->second->GetUrl());
+    GURL associated_url(associated_it->second.stream->GetUrl());
     if (associated_url.GetOrigin() != gurl.GetOrigin()) {
       ResetStream(stream_id, request_priority, RST_STREAM_REFUSED_STREAM,
                   base::StringPrintf(
                       "Rejected Cross Origin Push Stream %d",
                       associated_stream_id));
       return;
     }
   }
 
   // There should not be an existing pushed stream with the same path.
   if (unclaimed_pushed_streams_.find(url) != unclaimed_pushed_streams_.end()) {
     ResetStream(stream_id, request_priority, RST_STREAM_PROTOCOL_ERROR,
                 "Received duplicate pushed stream with url: " + url);
     return;
   }
 
   scoped_ptr<SpdyStream> stream(
       new SpdyStream(SPDY_PUSH_STREAM, this, gurl.PathForRequest(),
                      request_priority,
                      stream_initial_send_window_size_,
                      stream_initial_recv_window_size_,
                      net_log_));
   stream->set_stream_id(stream_id);
 
   DeleteExpiredPushedStreams();
-  unclaimed_pushed_streams_[url] =
-      std::pair<SpdyStream*, base::TimeTicks>(stream.get(), time_func_());
+  unclaimed_pushed_streams_[url] = PushedStreamInfo(stream_id, time_func_());
 
   stream->set_response_received();
   InsertActivatedStream(stream.Pass());
 
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end()) {
     NOTREACHED();
     return;
   }
 
   // Parse the headers.
-  if (!Respond(headers, it->second))
+  if (!Respond(headers, it->second.stream))
     return;
 
   base::StatsCounter push_requests("spdy.pushed_streams");
   push_requests.Increment();
 }
 
 void SpdySession::DeleteExpiredPushedStreams() {
   if (unclaimed_pushed_streams_.empty())
     return;
 
   // Check that adequate time has elapsed since the last sweep.
   if (time_func_() < next_unclaimed_push_stream_sweep_time_)
     return;
 
-  // Delete old streams.
+  // Gather old streams to delete.
   base::TimeTicks minimum_freshness = time_func_() -
       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
-  PushedStreamMap::iterator it;
-  for (it = unclaimed_pushed_streams_.begin();
-      it != unclaimed_pushed_streams_.end(); ) {
-    SpdyStream* stream = it->second.first;
-    base::TimeTicks creation_time = it->second.second;
-    // CloseActiveStream() will invalidate the current iterator, so
-    // move to next.
-    ++it;
-    if (minimum_freshness > creation_time) {
-      base::StatsCounter abandoned_push_streams(
-          "spdy.abandoned_push_streams");
-      base::StatsCounter abandoned_streams("spdy.abandoned_streams");
-      abandoned_push_streams.Increment();
-      abandoned_streams.Increment();
-      streams_abandoned_count_++;
-      CloseActiveStream(stream->stream_id(), ERR_INVALID_SPDY_STREAM);
-    }
+  std::vector<SpdyStreamId> streams_to_close;
+  for (PushedStreamMap::iterator it = unclaimed_pushed_streams_.begin();
+       it != unclaimed_pushed_streams_.end(); ++it) {
+    if (minimum_freshness > it->second.creation_time)
+      streams_to_close.push_back(it->second.stream_id);
   }
+
+  for (std::vector<SpdyStreamId>::const_iterator it = streams_to_close.begin();
+       it != streams_to_close.end(); ++it) {
+    base::StatsCounter abandoned_push_streams(
+        "spdy.abandoned_push_streams");
+    base::StatsCounter abandoned_streams("spdy.abandoned_streams");
+    abandoned_push_streams.Increment();
+    abandoned_streams.Increment();
+    streams_abandoned_count_++;
+    // CloseActiveStream() will remove the stream from
+    // |unclaimed_pushed_streams_|.
+    CloseActiveStream(*it, ERR_INVALID_SPDY_STREAM);
+  }
+
   next_unclaimed_push_stream_sweep_time_ = time_func_() +
       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
 }
 
 void SpdySession::OnSynReply(SpdyStreamId stream_id,
                              bool fin,
                              const SpdyHeaderBlock& headers) {
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SYN_REPLY,
         base::Bind(&NetLogSpdySynCallback,
                    &headers, fin, false,  // not unidirectional
                    stream_id, 0));
   }
 
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end()) {
     // NOTE:  it may just be that the stream was cancelled.
     return;
   }
 
-  SpdyStream* stream = it->second;
+  SpdyStream* stream = it->second.stream;
   CHECK_EQ(stream->stream_id(), stream_id);
 
-  if (stream->response_received()) {
-    stream->LogStreamError(ERR_SYN_REPLY_NOT_RECEIVED,
-                           "Received duplicate SYN_REPLY for stream.");
-    RecordProtocolErrorHistogram(PROTOCOL_ERROR_SYN_REPLY_NOT_RECEIVED);

[Ryan Hamilton, 2013/06/21 15:35:37]

Did this get lost or is it called by LogStreamError or ResetSTream?

[akalin, 2013/06/21 18:30:44]

Called by ResetStream

-    CloseActiveStream(stream->stream_id(), ERR_SPDY_PROTOCOL_ERROR);
+  if (!it->second.waiting_for_syn_reply) {
+    const std::string& error =
+        "Received duplicate SYN_REPLY for stream.";
+    stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
+    ResetStream(stream_id, stream->priority(),
+                RST_STREAM_STREAM_IN_USE, error);
     return;
   }
   stream->set_response_received();
+  it->second.waiting_for_syn_reply = false;
 
   Respond(headers, stream);
 }
 
 void SpdySession::OnHeaders(SpdyStreamId stream_id,
                             bool fin,
                             const SpdyHeaderBlock& headers) {
   if (net_log().IsLoggingAllEvents()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_RECV_HEADERS,
         base::Bind(&NetLogSpdySynCallback,
                    &headers, fin, /*unidirectional=*/false,
                    stream_id, 0));
   }
 
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end()) {
     // NOTE:  it may just be that the stream was cancelled.
     LOG(WARNING) << "Received HEADERS for invalid stream " << stream_id;
     return;
   }
 
-  CHECK_EQ(it->second->stream_id(), stream_id);
+  SpdyStream* stream = it->second.stream;
+  CHECK_EQ(stream->stream_id(), stream_id);
 
-  int rv = it->second->OnHeaders(headers);
+  int rv = stream->OnHeaders(headers);
   if (rv < 0) {
     DCHECK_NE(rv, ERR_IO_PENDING);
     CloseActiveStream(stream_id, rv);
   }
 }
 
 void SpdySession::OnRstStream(SpdyStreamId stream_id,
                               SpdyRstStreamStatus status) {
   std::string description;
   net_log().AddEvent(
       NetLog::TYPE_SPDY_SESSION_RST_STREAM,
       base::Bind(&NetLogSpdyRstCallback,
                  stream_id, status, &description));
 
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end()) {
     // NOTE:  it may just be that the stream was cancelled.
     LOG(WARNING) << "Received RST for invalid stream" << stream_id;
     return;
   }
 
-  CHECK_EQ(it->second->stream_id(), stream_id);
+  CHECK_EQ(it->second.stream->stream_id(), stream_id);
 
   if (status == 0) {
-    it->second->OnDataReceived(scoped_ptr<SpdyBuffer>());
+    it->second.stream->OnDataReceived(scoped_ptr<SpdyBuffer>());
   } else if (status == RST_STREAM_REFUSED_STREAM) {
     CloseActiveStream(stream_id, ERR_SPDY_SERVER_REFUSED_STREAM);
   } else {
     RecordProtocolErrorHistogram(
         PROTOCOL_ERROR_RST_STREAM_FOR_NON_ACTIVE_STREAM);
-    it->second->LogStreamError(
+    it->second.stream->LogStreamError(
         ERR_SPDY_PROTOCOL_ERROR,
         base::StringPrintf("SPDY stream closed with status: %d", status));
     // TODO(mbelshe): Map from Spdy-protocol errors to something sensical.
     //                For now, it doesn't matter much - it is a protocol error.
     CloseActiveStream(stream_id, ERR_SPDY_PROTOCOL_ERROR);
   }
 }
 
 void SpdySession::OnGoAway(SpdyStreamId last_accepted_stream_id,
                            SpdyGoAwayStatus status) {
@@ skipping 68 matching lines @@
     if (flow_control_state_ < FLOW_CONTROL_STREAM) {
       // TODO(akalin): Record an error and close the session.
       LOG(WARNING) << "Received WINDOW_UPDATE for stream " << stream_id
                    << " when flow control is not turned on";
       return;
     }
 
     ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
 
     if (it == active_streams_.end()) {
-      // TODO(akalin): Record an error and close the session.
+      // NOTE:  it may just be that the stream was cancelled.
       LOG(WARNING) << "Received WINDOW_UPDATE for invalid stream " << stream_id;
       return;
     }
 
+    SpdyStream* stream = it->second.stream;
+    CHECK_EQ(stream->stream_id(), stream_id);
+
     if (delta_window_size < 1u) {
-      ResetStream(stream_id, it->second->priority(),
+      ResetStream(stream_id, it->second.stream->priority(),
                   RST_STREAM_FLOW_CONTROL_ERROR,
                   base::StringPrintf(
                       "Received WINDOW_UPDATE with an invalid "
                       "delta_window_size %ud", delta_window_size));
       return;
     }
 
-    CHECK_EQ(it->second->stream_id(), stream_id);
-    it->second->IncreaseSendWindowSize(static_cast<int32>(delta_window_size));
+    CHECK_EQ(it->second.stream->stream_id(), stream_id);
+    it->second.stream->IncreaseSendWindowSize(
+        static_cast<int32>(delta_window_size));
   }
 }
 
 void SpdySession::SendStreamWindowUpdate(SpdyStreamId stream_id,
                                          uint32 delta_window_size) {
   CHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
   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());
+  CHECK_EQ(it->second.stream->stream_id(), stream_id);
+  SendWindowUpdateFrame(
+      stream_id, delta_window_size, it->second.stream->priority());
 }
 
 void SpdySession::SendInitialSettings() {
   // First notify the server about the settings they should use when
   // communicating with us.
   if (GetProtocolVersion() >= 2 && enable_sending_initial_settings_) {
     SettingsMap settings_map;
     // Create a new settings frame notifying the sever of our
     // max_concurrent_streams_ and initial window size.
     settings_map[SETTINGS_MAX_CONCURRENT_STREAMS] =
@@ skipping 78 matching lines @@
           NetLog::IntegerCallback("delta_window_size", delta_window_size));
       break;
     }
   }
 }
 
 void SpdySession::UpdateStreamsSendWindowSize(int32 delta_window_size) {
   DCHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
   for (ActiveStreamMap::iterator it = active_streams_.begin();
        it != active_streams_.end(); ++it) {
-    it->second->AdjustSendWindowSize(delta_window_size);
+    it->second.stream->AdjustSendWindowSize(delta_window_size);
   }
 
   for (CreatedStreamSet::const_iterator it = created_streams_.begin();
        it != created_streams_.end(); it++) {
     (*it)->AdjustSendWindowSize(delta_window_size);
   }
 }
 
 void SpdySession::SendPrefacePingIfNoneInFlight() {
   if (pings_in_flight_ || !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::SendWindowUpdateFrame(SpdyStreamId stream_id,
                                         uint32 delta_window_size,
                                         RequestPriority priority) {
   CHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   if (it != active_streams_.end()) {
-    CHECK_EQ(it->second->stream_id(), stream_id);
+    CHECK_EQ(it->second.stream->stream_id(), stream_id);
   } else {
     CHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
     CHECK_EQ(stream_id, kSessionFlowControlStreamId);
   }
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_SENT_WINDOW_UPDATE_FRAME,
       base::Bind(&NetLogSpdyWindowUpdateFrameCallback,
                  stream_id, delta_window_size));
 
@@ skipping 333 matching lines @@
       old_size = GetTotalSize(stream_send_unstall_queue_);
 
     SpdyStreamId stream_id = PopStreamToPossiblyResume();
     if (stream_id == 0)
       break;
     ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
     // The stream may actually still be send-stalled after this (due
     // to its own send window) but that's okay -- it'll then be
     // resumed once its send window increases.
     if (it != active_streams_.end())
-      it->second->PossiblyResumeIfSendStalled();
+      it->second.stream->PossiblyResumeIfSendStalled();
 
     // The size should decrease unless we got send-stalled again.
     if (!IsSendStalled())
       DCHECK_LT(GetTotalSize(stream_send_unstall_queue_), old_size);
   }
 }
 
 SpdyStreamId SpdySession::PopStreamToPossiblyResume() {
   for (int i = NUM_PRIORITIES - 1; i >= 0; --i) {
     std::deque<SpdyStreamId>* queue = &stream_send_unstall_queue_[i];
     if (!queue->empty()) {
       SpdyStreamId stream_id = queue->front();
       queue->pop_front();
       return stream_id;
     }
   }
   return 0;
 }
 
 }  // namespace net