Revert 273680 "Defer SpdySession destruction to support closing ..."

trunk/src/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 509 matching lines @@
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair()));
   next_unclaimed_push_stream_sweep_time_ = time_func_() +
       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 }
 
 SpdySession::~SpdySession() {
   CHECK(!in_io_loop_);
-  DcheckDraining();
+  DCHECK(!pool_);
+  DcheckClosed();
 
   // TODO(akalin): Check connection->is_initialized() instead. This
   // requires re-working CreateFakeSpdySession(), though.
   DCHECK(connection_->socket());
   // With SPDY we can't recycle sockets.
   connection_->socket()->Disconnect();
 
   RecordHistograms();
 
   net_log_.EndEvent(NetLog::TYPE_SPDY_SESSION);
@@ skipping 53 matching lines @@
 #if defined(SPDY_PROXY_AUTH_ORIGIN)
   UMA_HISTOGRAM_BOOLEAN("Net.SpdySessions_DataReductionProxy",
                         host_port_pair().Equals(HostPortPair::FromURL(
                             GURL(SPDY_PROXY_AUTH_ORIGIN))));
 #endif
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_INITIALIZED,
       connection_->socket()->NetLog().source().ToEventParametersCallback());
 
-  DCHECK_EQ(availability_state_, STATE_AVAILABLE);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   connection_->AddHigherLayeredPool(this);
   if (enable_sending_initial_data_)
     SendInitialData();
   pool_ = pool;
 
   // Bootstrap the read loop.
   base::MessageLoop::current()->PostTask(
       FROM_HERE,
       base::Bind(&SpdySession::PumpReadLoop,
                  weak_factory_.GetWeakPtr(), READ_STATE_DO_READ, OK));
 }
 
 bool SpdySession::VerifyDomainAuthentication(const std::string& domain) {
   if (!verify_domain_authentication_)
     return true;
 
-  if (availability_state_ == STATE_DRAINING)
+  if (availability_state_ == STATE_CLOSED)
     return false;
 
   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.
 
   bool unused = false;
   return
       !ssl_info.client_cert_sent &&
       (!ssl_info.channel_id_sent ||
        (ServerBoundCertService::GetDomainForHost(domain) ==
         ServerBoundCertService::GetDomainForHost(host_port_pair().host()))) &&
       ssl_info.cert->VerifyNameMatch(domain, &unused);
 }
 
 int SpdySession::GetPushStream(
     const GURL& url,
     base::WeakPtr<SpdyStream>* stream,
     const BoundNetLog& stream_net_log) {
   CHECK(!in_io_loop_);
 
   stream->reset();
 
-  if (availability_state_ == STATE_DRAINING)
+  // TODO(akalin): Add unit test exercising this code path.
+  if (availability_state_ == STATE_CLOSED)
     return ERR_CONNECTION_CLOSED;
 
   Error err = TryAccessStream(url);
   if (err != OK)
     return err;
 
   *stream = GetActivePushStream(url);
   if (*stream) {
     DCHECK_LT(streams_pushed_and_claimed_count_, streams_pushed_count_);
     streams_pushed_and_claimed_count_++;
   }
   return OK;
 }
 
 // {,Try}CreateStream() and TryAccessStream() can be called with
 // |in_io_loop_| set if a stream is being created in response to
 // another being closed due to received data.
 
 Error SpdySession::TryAccessStream(const GURL& url) {
-  CHECK_NE(availability_state_, STATE_DRAINING);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
 
   if (is_secure_ && certificate_error_code_ != OK &&
       (url.SchemeIs("https") || url.SchemeIs("wss"))) {
     RecordProtocolErrorHistogram(
         PROTOCOL_ERROR_REQUEST_FOR_SECURE_CONTENT_OVER_INSECURE_SESSION);
-    DoDrainSession(
+    CloseSessionResult result = DoCloseSession(
         static_cast<Error>(certificate_error_code_),
         "Tried to get SPDY stream for secure content over an unauthenticated "
         "session.");
+    DCHECK_EQ(result, SESSION_CLOSED_AND_REMOVED);
     return ERR_SPDY_PROTOCOL_ERROR;
   }
   return OK;
 }
 
 int SpdySession::TryCreateStream(
     const base::WeakPtr<SpdyStreamRequest>& request,
     base::WeakPtr<SpdyStream>* stream) {
   DCHECK(request);
 
   if (availability_state_ == STATE_GOING_AWAY)
     return ERR_FAILED;
 
-  if (availability_state_ == STATE_DRAINING)
+  // TODO(akalin): Add unit test exercising this code path.
+  if (availability_state_ == STATE_CLOSED)
     return ERR_CONNECTION_CLOSED;
 
   Error err = TryAccessStream(request->url());
   if (err != OK)
     return err;
 
   if (!max_concurrent_streams_ ||
       (active_streams_.size() + created_streams_.size() <
        max_concurrent_streams_)) {
     return CreateStream(*request, stream);
   }
 
   stalled_streams_++;
   net_log().AddEvent(NetLog::TYPE_SPDY_SESSION_STALLED_MAX_STREAMS);
   RequestPriority priority = request->priority();
   CHECK_GE(priority, MINIMUM_PRIORITY);
   CHECK_LE(priority, MAXIMUM_PRIORITY);
   pending_create_stream_queues_[priority].push_back(request);
   return ERR_IO_PENDING;
 }
 
 int SpdySession::CreateStream(const SpdyStreamRequest& request,
                               base::WeakPtr<SpdyStream>* stream) {
   DCHECK_GE(request.priority(), MINIMUM_PRIORITY);
   DCHECK_LE(request.priority(), MAXIMUM_PRIORITY);
 
   if (availability_state_ == STATE_GOING_AWAY)
     return ERR_FAILED;
 
-  if (availability_state_ == STATE_DRAINING)
+  // TODO(akalin): Add unit test exercising this code path.
+  if (availability_state_ == STATE_CLOSED)
     return ERR_CONNECTION_CLOSED;
 
   Error err = TryAccessStream(request.url());
   if (err != OK) {
     // This should have been caught in TryCreateStream().
     NOTREACHED();
     return err;
   }
 
   DCHECK(connection_->socket());
   DCHECK(connection_->socket()->IsConnected());
   if (connection_->socket()) {
     UMA_HISTOGRAM_BOOLEAN("Net.SpdySession.CreateStreamWithSocketConnected",
                           connection_->socket()->IsConnected());
     if (!connection_->socket()->IsConnected()) {
-      DoDrainSession(
+      CloseSessionResult result = DoCloseSession(
           ERR_CONNECTION_CLOSED,
           "Tried to create SPDY stream for a closed socket connection.");
+      DCHECK_EQ(result, SESSION_CLOSED_AND_REMOVED);
       return ERR_CONNECTION_CLOSED;
     }
   }
 
   scoped_ptr<SpdyStream> new_stream(
       new SpdyStream(request.type(), GetWeakPtr(), request.url(),
                      request.priority(),
                      stream_initial_send_window_size_,
                      stream_initial_recv_window_size_,
                      request.net_log()));
@@ skipping 120 matching lines @@
 
   return true;
 }
 
 base::WeakPtr<SpdySession> SpdySession::GetWeakPtr() {
   return weak_factory_.GetWeakPtr();
 }
 
 bool SpdySession::CloseOneIdleConnection() {
   CHECK(!in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK(pool_);
-  if (active_streams_.empty()) {
-    DoDrainSession(ERR_CONNECTION_CLOSED, "Closing idle connection.");
+  if (!active_streams_.empty())
+    return false;
+  CloseSessionResult result =
+      DoCloseSession(ERR_CONNECTION_CLOSED, "Closing one idle connection.");
+  if (result != SESSION_CLOSED_AND_REMOVED) {
+    NOTREACHED();
+    return false;
   }
-  // Return false as the socket wasn't immediately closed.
-  return false;
+  return true;
 }
 
 void SpdySession::EnqueueStreamWrite(
     const base::WeakPtr<SpdyStream>& stream,
     SpdyFrameType frame_type,
     scoped_ptr<SpdyBufferProducer> producer) {
   DCHECK(frame_type == HEADERS ||
          frame_type == DATA ||
          frame_type == CREDENTIAL ||
          frame_type == SYN_STREAM);
@@ skipping 32 matching lines @@
                    stream_id));
   }
 
   return syn_frame.Pass();
 }
 
 scoped_ptr<SpdyBuffer> SpdySession::CreateDataBuffer(SpdyStreamId stream_id,
                                                      IOBuffer* data,
                                                      int len,
                                                      SpdyDataFlags flags) {
-  if (availability_state_ == STATE_DRAINING) {
+  if (availability_state_ == STATE_CLOSED) {
+    NOTREACHED();
     return scoped_ptr<SpdyBuffer>();
   }
 
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
   CHECK(it != active_streams_.end());
   SpdyStream* stream = it->second.stream;
   CHECK_EQ(stream->stream_id(), stream_id);
 
   if (len < 0) {
     NOTREACHED();
@@ skipping 168 matching lines @@
   // TODO(akalin): When SpdyStream was ref-counted (and
   // |unclaimed_pushed_streams_| held scoped_refptr<SpdyStream>), this
   // was only done when status was not OK. This meant that pushed
   // streams can still be claimed after they're closed. This is
   // probably something that we still want to support, although server
   // push is hardly used. Write tests for this and fix this. (See
   // http://crbug.com/261712 .)
   if (owned_stream->type() == SPDY_PUSH_STREAM)
     unclaimed_pushed_streams_.erase(owned_stream->url());
 
+  base::WeakPtr<SpdySession> weak_this = GetWeakPtr();
+
   DeleteStream(owned_stream.Pass(), status);
-  MaybeFinishGoingAway();
+
+  if (!weak_this)
+    return;
+
+  if (availability_state_ == STATE_CLOSED)
+    return;
 
   // If there are no active streams and the socket pool is stalled, close the
   // session to free up a socket slot.
   if (active_streams_.empty() && connection_->IsPoolStalled()) {
-    DoDrainSession(ERR_CONNECTION_CLOSED, "Closing idle connection.");
+    CloseSessionResult result =
+        DoCloseSession(ERR_CONNECTION_CLOSED, "Closing idle connection.");
+    CHECK_NE(result, SESSION_ALREADY_CLOSED);
   }
 }
 
 void SpdySession::CloseCreatedStreamIterator(CreatedStreamSet::iterator it,
                                              int status) {
   scoped_ptr<SpdyStream> owned_stream(*it);
   created_streams_.erase(it);
   DeleteStream(owned_stream.Pass(), status);
 }
 
@@ skipping 24 matching lines @@
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyFrame> rst_frame(
       buffered_spdy_framer_->CreateRstStream(stream_id, status));
 
   EnqueueSessionWrite(priority, RST_STREAM, rst_frame.Pass());
   RecordProtocolErrorHistogram(MapRstStreamStatusToProtocolError(status));
 }
 
 void SpdySession::PumpReadLoop(ReadState expected_read_state, int result) {
   CHECK(!in_io_loop_);
-  if (availability_state_ == STATE_DRAINING) {
+  CHECK_NE(availability_state_, STATE_CLOSED);
+  CHECK_EQ(read_state_, expected_read_state);
+
+  result = DoReadLoop(expected_read_state, result);
+
+  if (availability_state_ == STATE_CLOSED) {
+    CHECK_EQ(result, error_on_close_);
+    CHECK_LT(error_on_close_, ERR_IO_PENDING);
+    RemoveFromPool();
     return;
   }
-  ignore_result(DoReadLoop(expected_read_state, result));
+
+  CHECK(result == OK || result == ERR_IO_PENDING);
 }
 
 int SpdySession::DoReadLoop(ReadState expected_read_state, int result) {
   CHECK(!in_io_loop_);
+  CHECK_NE(availability_state_, STATE_CLOSED);
   CHECK_EQ(read_state_, expected_read_state);
 
   in_io_loop_ = true;
 
   int bytes_read_without_yielding = 0;
 
-  // Loop until the session is draining, the read becomes blocked, or
+  // Loop until the session is closed, the read becomes blocked, or
   // the read limit is exceeded.
   while (true) {
     switch (read_state_) {
       case READ_STATE_DO_READ:
         CHECK_EQ(result, OK);
         result = DoRead();
         break;
       case READ_STATE_DO_READ_COMPLETE:
         if (result > 0)
           bytes_read_without_yielding += result;
         result = DoReadComplete(result);
         break;
       default:
         NOTREACHED() << "read_state_: " << read_state_;
         break;
     }
 
-    if (availability_state_ == STATE_DRAINING)
+    if (availability_state_ == STATE_CLOSED) {
+      CHECK_EQ(result, error_on_close_);
+      CHECK_LT(result, ERR_IO_PENDING);
       break;
+    }
 
     if (result == ERR_IO_PENDING)
       break;
 
     if (bytes_read_without_yielding > kMaxReadBytesWithoutYielding) {
       read_state_ = READ_STATE_DO_READ;
       base::MessageLoop::current()->PostTask(
           FROM_HERE,
           base::Bind(&SpdySession::PumpReadLoop,
                      weak_factory_.GetWeakPtr(), READ_STATE_DO_READ, OK));
       result = ERR_IO_PENDING;
       break;
     }
   }
 
   CHECK(in_io_loop_);
   in_io_loop_ = false;
 
   return result;
 }
 
 int SpdySession::DoRead() {
   CHECK(in_io_loop_);
+  CHECK_NE(availability_state_, STATE_CLOSED);
 
   CHECK(connection_);
   CHECK(connection_->socket());
   read_state_ = READ_STATE_DO_READ_COMPLETE;
   return connection_->socket()->Read(
       read_buffer_.get(),
       kReadBufferSize,
       base::Bind(&SpdySession::PumpReadLoop,
                  weak_factory_.GetWeakPtr(), READ_STATE_DO_READ_COMPLETE));
 }
 
 int SpdySession::DoReadComplete(int result) {
   CHECK(in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
 
   // Parse a frame.  For now this code requires that the frame fit into our
   // buffer (kReadBufferSize).
   // TODO(mbelshe): support arbitrarily large frames!
 
   if (result == 0) {
     UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySession.BytesRead.EOF",
                                 total_bytes_received_, 1, 100000000, 50);
-    DoDrainSession(ERR_CONNECTION_CLOSED, "Connection closed");
-
+    CloseSessionResult close_session_result =
+        DoCloseSession(ERR_CONNECTION_CLOSED, "Connection closed");
+    DCHECK_EQ(close_session_result, SESSION_CLOSED_BUT_NOT_REMOVED);
+    DCHECK_EQ(availability_state_, STATE_CLOSED);
+    DCHECK_EQ(error_on_close_, ERR_CONNECTION_CLOSED);
     return ERR_CONNECTION_CLOSED;
   }
 
   if (result < 0) {
-    DoDrainSession(static_cast<Error>(result), "result is < 0.");
+    CloseSessionResult close_session_result =
+        DoCloseSession(static_cast<Error>(result), "result is < 0.");
+    DCHECK_EQ(close_session_result, SESSION_CLOSED_BUT_NOT_REMOVED);
+    DCHECK_EQ(availability_state_, STATE_CLOSED);
+    DCHECK_EQ(error_on_close_, result);
     return result;
   }
   CHECK_LE(result, kReadBufferSize);
   total_bytes_received_ += result;
 
   last_activity_time_ = time_func_();
 
   DCHECK(buffered_spdy_framer_.get());
   char* data = read_buffer_->data();
   while (result > 0) {
     uint32 bytes_processed = buffered_spdy_framer_->ProcessInput(data, result);
     result -= bytes_processed;
     data += bytes_processed;
 
-    if (availability_state_ == STATE_DRAINING) {
-      return ERR_CONNECTION_CLOSED;
+    if (availability_state_ == STATE_CLOSED) {
+      DCHECK_LT(error_on_close_, ERR_IO_PENDING);
+      return error_on_close_;
     }
 
     DCHECK_EQ(buffered_spdy_framer_->error_code(), SpdyFramer::SPDY_NO_ERROR);
   }
 
   read_state_ = READ_STATE_DO_READ;
   return OK;
 }
 
 void SpdySession::PumpWriteLoop(WriteState expected_write_state, int result) {
   CHECK(!in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK_EQ(write_state_, expected_write_state);
 
-  DoWriteLoop(expected_write_state, result);
+  result = DoWriteLoop(expected_write_state, result);
 
-  if (availability_state_ == STATE_DRAINING && !in_flight_write_ &&
-      write_queue_.IsEmpty()) {
-    pool_->RemoveUnavailableSession(GetWeakPtr());  // Destroys |this|.
+  if (availability_state_ == STATE_CLOSED) {
+    DCHECK_EQ(result, error_on_close_);
+    DCHECK_LT(error_on_close_, ERR_IO_PENDING);
+    RemoveFromPool();
     return;
   }
+
+  DCHECK(result == OK || result == ERR_IO_PENDING);
 }
 
 int SpdySession::DoWriteLoop(WriteState expected_write_state, int result) {
   CHECK(!in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK_NE(write_state_, WRITE_STATE_IDLE);
   DCHECK_EQ(write_state_, expected_write_state);
 
   in_io_loop_ = true;
 
   // Loop until the session is closed or the write becomes blocked.
   while (true) {
     switch (write_state_) {
       case WRITE_STATE_DO_WRITE:
         DCHECK_EQ(result, OK);
         result = DoWrite();
         break;
       case WRITE_STATE_DO_WRITE_COMPLETE:
         result = DoWriteComplete(result);
         break;
       case WRITE_STATE_IDLE:
       default:
         NOTREACHED() << "write_state_: " << write_state_;
         break;
     }
 
+    if (availability_state_ == STATE_CLOSED) {
+      DCHECK_EQ(result, error_on_close_);
+      DCHECK_LT(result, ERR_IO_PENDING);
+      break;
+    }
+
     if (write_state_ == WRITE_STATE_IDLE) {
       DCHECK_EQ(result, ERR_IO_PENDING);
       break;
     }
 
     if (result == ERR_IO_PENDING)
       break;
   }
 
   CHECK(in_io_loop_);
   in_io_loop_ = false;
 
   return result;
 }
 
 int SpdySession::DoWrite() {
   CHECK(in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
 
   DCHECK(buffered_spdy_framer_);
   if (in_flight_write_) {
     DCHECK_GT(in_flight_write_->GetRemainingSize(), 0u);
   } else {
     // Grab the next frame to send.
     SpdyFrameType frame_type = DATA;
     scoped_ptr<SpdyBufferProducer> producer;
     base::WeakPtr<SpdyStream> stream;
     if (!write_queue_.Dequeue(&frame_type, &producer, &stream)) {
@@ skipping 43 matching lines @@
       in_flight_write_->GetIOBufferForRemainingData();
   return connection_->socket()->Write(
       write_io_buffer.get(),
       in_flight_write_->GetRemainingSize(),
       base::Bind(&SpdySession::PumpWriteLoop,
                  weak_factory_.GetWeakPtr(), WRITE_STATE_DO_WRITE_COMPLETE));
 }
 
 int SpdySession::DoWriteComplete(int result) {
   CHECK(in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK_NE(result, ERR_IO_PENDING);
   DCHECK_GT(in_flight_write_->GetRemainingSize(), 0u);
 
   last_activity_time_ = time_func_();
 
   if (result < 0) {
     DCHECK_NE(result, ERR_IO_PENDING);
     in_flight_write_.reset();
     in_flight_write_frame_type_ = DATA;
     in_flight_write_frame_size_ = 0;
     in_flight_write_stream_.reset();
-    write_state_ = WRITE_STATE_DO_WRITE;
-    DoDrainSession(static_cast<Error>(result), "Write error");
-    return OK;
+    CloseSessionResult close_session_result =
+        DoCloseSession(static_cast<Error>(result), "Write error");
+    DCHECK_EQ(close_session_result, SESSION_CLOSED_BUT_NOT_REMOVED);
+    DCHECK_EQ(availability_state_, STATE_CLOSED);
+    DCHECK_EQ(error_on_close_, result);
+    return result;
   }
 
   // It should not be possible to have written more bytes than our
   // in_flight_write_.
   DCHECK_LE(static_cast<size_t>(result),
             in_flight_write_->GetRemainingSize());
 
   if (result > 0) {
     in_flight_write_->Consume(static_cast<size_t>(result));
 
@@ skipping 23 matching lines @@
 void SpdySession::DcheckGoingAway() const {
 #if DCHECK_IS_ON
   DCHECK_GE(availability_state_, STATE_GOING_AWAY);
   for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
     DCHECK(pending_create_stream_queues_[i].empty());
   }
   DCHECK(created_streams_.empty());
 #endif
 }
 
-void SpdySession::DcheckDraining() const {
+void SpdySession::DcheckClosed() const {
   DcheckGoingAway();
-  DCHECK_EQ(availability_state_, STATE_DRAINING);
+  DCHECK_EQ(availability_state_, STATE_CLOSED);
+  DCHECK_LT(error_on_close_, ERR_IO_PENDING);
   DCHECK(active_streams_.empty());
   DCHECK(unclaimed_pushed_streams_.empty());
+  DCHECK(write_queue_.IsEmpty());
 }
 
 void SpdySession::StartGoingAway(SpdyStreamId last_good_stream_id,
                                  Error status) {
   DCHECK_GE(availability_state_, STATE_GOING_AWAY);
 
   // The loops below are carefully written to avoid reentrancy problems.
 
   while (true) {
     size_t old_size = GetTotalSize(pending_create_stream_queues_);
@@ skipping 29 matching lines @@
     // away.
     DCHECK_GT(old_size, created_streams_.size());
   }
 
   write_queue_.RemovePendingWritesForStreamsAfter(last_good_stream_id);
 
   DcheckGoingAway();
 }
 
 void SpdySession::MaybeFinishGoingAway() {
-  if (active_streams_.empty() && availability_state_ == STATE_GOING_AWAY) {
-    DoDrainSession(OK, "Finished going away");
+  DcheckGoingAway();
+  if (active_streams_.empty() && availability_state_ != STATE_CLOSED) {
+    CloseSessionResult result =
+        DoCloseSession(ERR_CONNECTION_CLOSED, "Finished going away");
+    CHECK_NE(result, SESSION_ALREADY_CLOSED);
   }
 }
 
-void SpdySession::DoDrainSession(Error err, const std::string& description) {
-  if (availability_state_ == STATE_DRAINING) {
-    return;
-  }
-  MakeUnavailable();
+SpdySession::CloseSessionResult SpdySession::DoCloseSession(
+    Error err,
+    const std::string& description) {
+  CHECK_LT(err, ERR_IO_PENDING);
 
-  // TODO(jgraettinger): If draining with an |err|, enqueue a GOAWAY frame here.
-
-  availability_state_ = STATE_DRAINING;
-  error_on_close_ = err;
+  if (availability_state_ == STATE_CLOSED)
+    return SESSION_ALREADY_CLOSED;
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_CLOSE,
       base::Bind(&NetLogSpdySessionCloseCallback, err, &description));
 
   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.SpdySession.ClosedOnError", -err);
   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySession.BytesRead.OtherErrors",
                               total_bytes_received_, 1, 100000000, 50);
 
+  CHECK(pool_);
+  if (availability_state_ != STATE_GOING_AWAY)
+    pool_->MakeSessionUnavailable(GetWeakPtr());
+
+  availability_state_ = STATE_CLOSED;
+  error_on_close_ = err;
+
   StartGoingAway(0, err);
-  DcheckDraining();
-  MaybePostWriteLoop();
+  write_queue_.Clear();
+
+  DcheckClosed();
+
+  if (in_io_loop_)
+    return SESSION_CLOSED_BUT_NOT_REMOVED;
+
+  RemoveFromPool();
+  return SESSION_CLOSED_AND_REMOVED;
+}
+
+void SpdySession::RemoveFromPool() {
+  DcheckClosed();
+  CHECK(pool_);
+
+  SpdySessionPool* pool = pool_;
+  pool_ = NULL;
+  pool->RemoveUnavailableSession(GetWeakPtr());
 }
 
 void SpdySession::LogAbandonedStream(SpdyStream* stream, Error status) {
   DCHECK(stream);
   std::string description = base::StringPrintf(
       "ABANDONED (stream_id=%d): ", stream->stream_id()) +
       stream->url().spec();
   stream->LogStreamError(status, description);
   // We don't increment the streams abandoned counter here. If the
   // stream isn't active (i.e., it hasn't written anything to the wire
@@ skipping 18 matching lines @@
 
 SpdyStreamId SpdySession::GetNewStreamId() {
   CHECK_LE(stream_hi_water_mark_, kLastStreamId);
   SpdyStreamId id = stream_hi_water_mark_;
   stream_hi_water_mark_ += 2;
   return id;
 }
 
 void SpdySession::CloseSessionOnError(Error err,
                                       const std::string& description) {
-  DoDrainSession(err, description);
+  // We may be called from anywhere, so we can't expect a particular
+  // return value.
+  ignore_result(DoCloseSession(err, description));
 }
 
 void SpdySession::MakeUnavailable() {
-  CHECK_NE(availability_state_, STATE_DRAINING);
-  if (availability_state_ == STATE_AVAILABLE) {
+  if (availability_state_ < STATE_GOING_AWAY) {
     availability_state_ = STATE_GOING_AWAY;
+    DCHECK(pool_);
     pool_->MakeSessionUnavailable(GetWeakPtr());
   }
 }
 
 base::Value* SpdySession::GetInfoAsValue() const {
   base::DictionaryValue* dict = new base::DictionaryValue();
 
   dict->SetInteger("source_id", net_log_.source().id);
 
   dict->SetString("host_port_pair", host_port_pair().ToString());
@@ skipping 88 matching lines @@
       scoped_ptr<SpdyBufferProducer>(
           new SimpleBufferProducer(
               scoped_ptr<SpdyBuffer>(new SpdyBuffer(frame.Pass())))),
       base::WeakPtr<SpdyStream>());
 }
 
 void SpdySession::EnqueueWrite(RequestPriority priority,
                                SpdyFrameType frame_type,
                                scoped_ptr<SpdyBufferProducer> producer,
                                const base::WeakPtr<SpdyStream>& stream) {
-  if (availability_state_ == STATE_DRAINING)
+  if (availability_state_ == STATE_CLOSED)
     return;
 
+  bool was_idle = write_queue_.IsEmpty();
   write_queue_.Enqueue(priority, frame_type, producer.Pass(), stream);
-  MaybePostWriteLoop();
-}
-
-void SpdySession::MaybePostWriteLoop() {
   if (write_state_ == WRITE_STATE_IDLE) {
-    CHECK(!in_flight_write_);
+    DCHECK(was_idle);
+    DCHECK(!in_flight_write_);
     write_state_ = WRITE_STATE_DO_WRITE;
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&SpdySession::PumpWriteLoop,
                    weak_factory_.GetWeakPtr(), WRITE_STATE_DO_WRITE, OK));
   }
 }
 
 void SpdySession::InsertCreatedStream(scoped_ptr<SpdyStream> stream) {
   CHECK_EQ(stream->stream_id(), 0u);
@@ skipping 23 matching lines @@
 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
     // need to let the write complete, so we clear
     // |in_flight_write_stream_| and let the write finish on its own
     // without notifying |in_flight_write_stream_|.
     in_flight_write_stream_.reset();
   }
 
   write_queue_.RemovePendingWritesForStream(stream->GetWeakPtr());
+
+  // |stream->OnClose()| may end up closing |this|, so detect that.
+  base::WeakPtr<SpdySession> weak_this = GetWeakPtr();
+
   stream->OnClose(status);
 
-  if (availability_state_ == STATE_AVAILABLE) {
-    ProcessPendingStreamRequests();
+  if (!weak_this)
+    return;
+
+  switch (availability_state_) {
+    case STATE_AVAILABLE:
+      ProcessPendingStreamRequests();
+      break;
+    case STATE_GOING_AWAY:
+      DcheckGoingAway();
+      MaybeFinishGoingAway();
+      break;
+    case STATE_CLOSED:
+      // Do nothing.
+      break;
   }
 }
 
 base::WeakPtr<SpdyStream> SpdySession::GetActivePushStream(const GURL& url) {
   base::StatsCounter used_push_streams("spdy.claimed_push_streams");
 
   PushedStreamMap::iterator unclaimed_it = unclaimed_pushed_streams_.find(url);
   if (unclaimed_it == unclaimed_pushed_streams_.end())
     return base::WeakPtr<SpdyStream>();
 
@@ skipping 23 matching lines @@
     SSLCertRequestInfo* cert_request_info) {
   if (!is_secure_)
     return false;
   GetSSLClientSocket()->GetSSLCertRequestInfo(cert_request_info);
   return true;
 }
 
 void SpdySession::OnError(SpdyFramer::SpdyError error_code) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   RecordProtocolErrorHistogram(MapFramerErrorToProtocolError(error_code));
   std::string description = base::StringPrintf(
       "SPDY_ERROR error_code: %d.", error_code);
-  DoDrainSession(ERR_SPDY_PROTOCOL_ERROR, description);
+  CloseSessionResult result =
+      DoCloseSession(ERR_SPDY_PROTOCOL_ERROR, description);
+  DCHECK_EQ(result, SESSION_CLOSED_BUT_NOT_REMOVED);
 }
 
 void SpdySession::OnStreamError(SpdyStreamId stream_id,
                                 const std::string& description) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end()) {
     // We still want to send a frame to reset the stream even if we
     // don't know anything about it.
     EnqueueResetStreamFrame(
         stream_id, IDLE, RST_STREAM_PROTOCOL_ERROR, description);
     return;
   }
 
   ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, description);
 }
 
 void SpdySession::OnDataFrameHeader(SpdyStreamId stream_id,
                                     size_t length,
                                     bool fin) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   ActiveStreamMap::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);
 
   DCHECK(buffered_spdy_framer_);
   size_t header_len = buffered_spdy_framer_->GetDataFrameMinimumSize();
   stream->IncrementRawReceivedBytes(header_len);
 }
 
 void SpdySession::OnStreamFrameData(SpdyStreamId stream_id,
                                     const char* data,
                                     size_t len,
                                     bool fin) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   if (data == NULL && len != 0) {
     // This is notification of consumed data padding.
     // TODO(jgraettinger): Properly flow padding into WINDOW_UPDATE frames.
     // See crbug.com/353012.
     return;
   }
 
   DCHECK_LT(len, 1u << 24);
   if (net_log().IsLogging()) {
     net_log().AddEvent(
@@ skipping 39 matching lines @@
     ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, error);
     return;
   }
 
   stream->OnDataReceived(buffer.Pass());
 }
 
 void SpdySession::OnSettings(bool clear_persisted) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   if (clear_persisted)
     http_server_properties_->ClearSpdySettings(host_port_pair());
 
   if (net_log_.IsLogging()) {
     net_log_.AddEvent(
         NetLog::TYPE_SPDY_SESSION_RECV_SETTINGS,
         base::Bind(&NetLogSpdySettingsCallback, host_port_pair(),
                    clear_persisted));
   }
 
   if (GetProtocolVersion() >= SPDY4) {
     // Send an acknowledgment of the setting.
     SpdySettingsIR settings_ir;
     settings_ir.set_is_ack(true);
     EnqueueSessionWrite(
         HIGHEST,
         SETTINGS,
         scoped_ptr<SpdyFrame>(
             buffered_spdy_framer_->SerializeFrame(settings_ir)));
   }
 }
 
 void SpdySession::OnSetting(SpdySettingsIds id,
                             uint8 flags,
                             uint32 value) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   HandleSetting(id, value);
   http_server_properties_->SetSpdySetting(
       host_port_pair(),
       id,
       static_cast<SpdySettingsFlags>(flags),
       value);
   received_settings_ = true;
 
   // Log the setting.
   net_log_.AddEvent(
@@ skipping 47 matching lines @@
 }
 
 void SpdySession::OnSynStream(SpdyStreamId stream_id,
                               SpdyStreamId associated_stream_id,
                               SpdyPriority priority,
                               bool fin,
                               bool unidirectional,
                               const SpdyHeaderBlock& headers) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   base::Time response_time = base::Time::Now();
   base::TimeTicks recv_first_byte_time = time_func_();
 
   if (net_log_.IsLogging()) {
     net_log_.AddEvent(
         NetLog::TYPE_SPDY_SESSION_PUSHED_SYN_STREAM,
         base::Bind(&NetLogSpdySynStreamReceivedCallback,
                    &headers, fin, unidirectional, priority,
                    stream_id, associated_stream_id));
   }
@@ skipping 165 matching lines @@
 
   next_unclaimed_push_stream_sweep_time_ = time_func_() +
       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
 }
 
 void SpdySession::OnSynReply(SpdyStreamId stream_id,
                              bool fin,
                              const SpdyHeaderBlock& headers) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   base::Time response_time = base::Time::Now();
   base::TimeTicks recv_first_byte_time = time_func_();
 
   if (net_log().IsLogging()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_SYN_REPLY,
         base::Bind(&NetLogSpdySynReplyOrHeadersReceivedCallback,
                    &headers, fin, stream_id));
   }
 
@@ skipping 27 matching lines @@
 
   ignore_result(OnInitialResponseHeadersReceived(
       headers, response_time, recv_first_byte_time, stream));
 }
 
 void SpdySession::OnHeaders(SpdyStreamId stream_id,
                             bool fin,
                             const SpdyHeaderBlock& headers) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   if (net_log().IsLogging()) {
     net_log().AddEvent(
         NetLog::TYPE_SPDY_SESSION_RECV_HEADERS,
         base::Bind(&NetLogSpdySynReplyOrHeadersReceivedCallback,
                    &headers, fin, stream_id));
   }
 
   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
   if (it == active_streams_.end()) {
     // NOTE:  it may just be that the stream was cancelled.
@@ skipping 27 matching lines @@
       DCHECK_NE(rv, ERR_IO_PENDING);
       DCHECK(active_streams_.find(stream_id) == active_streams_.end());
     }
   }
 }
 
 void SpdySession::OnRstStream(SpdyStreamId stream_id,
                               SpdyRstStreamStatus status) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   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;
@@ skipping 15 matching lines @@
     // TODO(mbelshe): Map from Spdy-protocol errors to something sensical.
     //                For now, it doesn't matter much - it is a protocol error.
     CloseActiveStreamIterator(it, ERR_SPDY_PROTOCOL_ERROR);
   }
 }
 
 void SpdySession::OnGoAway(SpdyStreamId last_accepted_stream_id,
                            SpdyGoAwayStatus status) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   net_log_.AddEvent(NetLog::TYPE_SPDY_SESSION_GOAWAY,
       base::Bind(&NetLogSpdyGoAwayCallback,
                  last_accepted_stream_id,
                  active_streams_.size(),
                  unclaimed_pushed_streams_.size(),
                  status));
   MakeUnavailable();
   StartGoingAway(last_accepted_stream_id, ERR_ABORTED);
   // This is to handle the case when we already don't have any active
   // streams (i.e., StartGoingAway() did nothing). Otherwise, we have
   // active streams and so the last one being closed will finish the
   // going away process (see DeleteStream()).
   MaybeFinishGoingAway();
 }
 
 void SpdySession::OnPing(SpdyPingId unique_id, bool is_ack) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_PING,
       base::Bind(&NetLogSpdyPingCallback, unique_id, is_ack, "received"));
 
   // Send response to a PING from server.
   if ((protocol_ >= kProtoSPDY4 && !is_ack) ||
       (protocol_ < kProtoSPDY4 && unique_id % 2 == 0)) {
     WritePingFrame(unique_id, true);
     return;
   }
 
   --pings_in_flight_;
   if (pings_in_flight_ < 0) {
     RecordProtocolErrorHistogram(PROTOCOL_ERROR_UNEXPECTED_PING);
-    DoDrainSession(ERR_SPDY_PROTOCOL_ERROR, "pings_in_flight_ is < 0.");
+    CloseSessionResult result =
+        DoCloseSession(ERR_SPDY_PROTOCOL_ERROR, "pings_in_flight_ is < 0.");
+    DCHECK_EQ(result, SESSION_CLOSED_BUT_NOT_REMOVED);
     pings_in_flight_ = 0;
     return;
   }
 
   if (pings_in_flight_ > 0)
     return;
 
   // We will record RTT in histogram when there are no more client sent
   // pings_in_flight_.
   RecordPingRTTHistogram(time_func_() - last_ping_sent_time_);
 }
 
 void SpdySession::OnWindowUpdate(SpdyStreamId stream_id,
                                  uint32 delta_window_size) {
   CHECK(in_io_loop_);
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   DCHECK_LE(delta_window_size, static_cast<uint32>(kint32max));
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_RECEIVED_WINDOW_UPDATE_FRAME,
       base::Bind(&NetLogSpdyWindowUpdateFrameCallback,
                  stream_id, delta_window_size));
 
   if (stream_id == kSessionFlowControlStreamId) {
     // WINDOW_UPDATE for the session.
     if (flow_control_state_ < FLOW_CONTROL_STREAM_AND_SESSION) {
       LOG(WARNING) << "Received WINDOW_UPDATE for session when "
                    << "session flow control is not turned on";
       // TODO(akalin): Record an error and close the session.
       return;
     }
 
     if (delta_window_size < 1u) {
       RecordProtocolErrorHistogram(PROTOCOL_ERROR_INVALID_WINDOW_UPDATE_SIZE);
-      DoDrainSession(
+      CloseSessionResult result = DoCloseSession(
           ERR_SPDY_PROTOCOL_ERROR,
           "Received WINDOW_UPDATE with an invalid delta_window_size " +
-              base::UintToString(delta_window_size));
+          base::UintToString(delta_window_size));
+      DCHECK_EQ(result, SESSION_CLOSED_BUT_NOT_REMOVED);
       return;
     }
 
     IncreaseSendWindowSize(static_cast<int32>(delta_window_size));
   } else {
     // WINDOW_UPDATE for a stream.
     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";
@@ skipping 37 matching lines @@
   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->stream_id(), stream_id);
   SendWindowUpdateFrame(
       stream_id, delta_window_size, it->second.stream->priority());
 }
 
 void SpdySession::SendInitialData() {
   DCHECK(enable_sending_initial_data_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
 
   if (send_connection_header_prefix_) {
     DCHECK_EQ(protocol_, kProtoSPDY4);
     scoped_ptr<SpdyFrame> connection_header_prefix_frame(
         new SpdyFrame(const_cast<char*>(kHttp2ConnectionHeaderPrefix),
                       kHttp2ConnectionHeaderPrefixSize,
                       false /* take_ownership */));
     // Count the prefix as part of the subsequent SETTINGS frame.
     EnqueueSessionWrite(HIGHEST, SETTINGS,
                         connection_header_prefix_frame.Pass());
@@ skipping 45 matching lines @@
     const SpdySettingsIds new_id = it->first;
     const uint32 new_val = it->second.second;
     HandleSetting(new_id, new_val);
   }
 
   SendSettings(server_settings_map);
 }
 
 
 void SpdySession::SendSettings(const SettingsMap& settings) {
+  DCHECK_NE(availability_state_, STATE_CLOSED);
+
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_SEND_SETTINGS,
       base::Bind(&NetLogSpdySendSettingsCallback, &settings));
 
   // Create the SETTINGS frame and send it.
   DCHECK(buffered_spdy_framer_.get());
   scoped_ptr<SpdyFrame> settings_frame(
       buffered_spdy_framer_->CreateSettings(settings));
   sent_settings_ = true;
   EnqueueSessionWrite(HIGHEST, SETTINGS, settings_frame.Pass());
@@ skipping 108 matching lines @@
 
   check_ping_status_pending_ = true;
   base::MessageLoop::current()->PostDelayedTask(
       FROM_HERE,
       base::Bind(&SpdySession::CheckPingStatus, weak_factory_.GetWeakPtr(),
                  time_func_()), hung_interval_);
 }
 
 void SpdySession::CheckPingStatus(base::TimeTicks last_check_time) {
   CHECK(!in_io_loop_);
+  DCHECK_NE(availability_state_, STATE_CLOSED);
 
   // Check if we got a response back for all PINGs we had sent.
   if (pings_in_flight_ == 0) {
     check_ping_status_pending_ = false;
     return;
   }
 
   DCHECK(check_ping_status_pending_);
 
   base::TimeTicks now = time_func_();
   base::TimeDelta delay = hung_interval_ - (now - last_activity_time_);
 
   if (delay.InMilliseconds() < 0 || last_activity_time_ < last_check_time) {
     // Track all failed PING messages in a separate bucket.
     RecordPingRTTHistogram(base::TimeDelta::Max());
-    DoDrainSession(ERR_SPDY_PING_FAILED, "Failed ping.");
+    CloseSessionResult result =
+        DoCloseSession(ERR_SPDY_PING_FAILED, "Failed ping.");
+    DCHECK_EQ(result, SESSION_CLOSED_AND_REMOVED);
     return;
   }
 
   // Check the status of connection after a delay.
   base::MessageLoop::current()->PostDelayedTask(
       FROM_HERE,
       base::Bind(&SpdySession::CheckPingStatus, weak_factory_.GetWeakPtr(),
                  now),
       delay);
 }
@@ skipping 112 matching lines @@
   return ssl_socket;
 }
 
 void SpdySession::OnWriteBufferConsumed(
     size_t frame_payload_size,
     size_t consume_size,
     SpdyBuffer::ConsumeSource consume_source) {
   // We can be called with |in_io_loop_| set if a write SpdyBuffer is
   // deleted (e.g., a stream is closed due to incoming data).
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
 
   if (consume_source == SpdyBuffer::DISCARD) {
     // If we're discarding a frame or part of it, increase the send
     // window by the number of discarded bytes. (Although if we're
     // discarding part of a frame, it's probably because of a write
     // error and we'll be tearing down the session soon.)
     size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
     DCHECK_GT(remaining_payload_bytes, 0u);
     IncreaseSendWindowSize(static_cast<int32>(remaining_payload_bytes));
   }
   // For consumed bytes, the send window is increased when we receive
   // a WINDOW_UPDATE frame.
 }
 
 void SpdySession::IncreaseSendWindowSize(int32 delta_window_size) {
   // We can be called with |in_io_loop_| set if a SpdyBuffer is
   // deleted (e.g., a stream is closed due to incoming data).
 
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
   DCHECK_GE(delta_window_size, 1);
 
   // Check for overflow.
   int32 max_delta_window_size = kint32max - session_send_window_size_;
   if (delta_window_size > max_delta_window_size) {
     RecordProtocolErrorHistogram(PROTOCOL_ERROR_INVALID_WINDOW_UPDATE_SIZE);
-    DoDrainSession(
+    CloseSessionResult result = DoCloseSession(
         ERR_SPDY_PROTOCOL_ERROR,
         "Received WINDOW_UPDATE [delta: " +
-            base::IntToString(delta_window_size) +
-            "] for session overflows session_send_window_size_ [current: " +
-            base::IntToString(session_send_window_size_) + "]");
+        base::IntToString(delta_window_size) +
+        "] for session overflows session_send_window_size_ [current: " +
+        base::IntToString(session_send_window_size_) + "]");
+    DCHECK_NE(result, SESSION_ALREADY_CLOSED);
     return;
   }
 
   session_send_window_size_ += delta_window_size;
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_SEND_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  delta_window_size, session_send_window_size_));
 
   DCHECK(!IsSendStalled());
   ResumeSendStalledStreams();
 }
 
 void SpdySession::DecreaseSendWindowSize(int32 delta_window_size) {
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
 
   // We only call this method when sending a frame. Therefore,
   // |delta_window_size| should be within the valid frame size range.
   DCHECK_GE(delta_window_size, 1);
   DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
 
   // |send_window_size_| should have been at least |delta_window_size| for
   // this call to happen.
   DCHECK_GE(session_send_window_size_, delta_window_size);
 
   session_send_window_size_ -= delta_window_size;
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_SEND_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  -delta_window_size, session_send_window_size_));
 }
 
 void SpdySession::OnReadBufferConsumed(
     size_t consume_size,
     SpdyBuffer::ConsumeSource consume_source) {
   // We can be called with |in_io_loop_| set if a read SpdyBuffer is
   // deleted (e.g., discarded by a SpdyReadQueue).
 
+  if (availability_state_ == STATE_CLOSED)
+    return;
+
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
   DCHECK_GE(consume_size, 1u);
   DCHECK_LE(consume_size, static_cast<size_t>(kint32max));
 
   IncreaseRecvWindowSize(static_cast<int32>(consume_size));
 }
 
 void SpdySession::IncreaseRecvWindowSize(int32 delta_window_size) {
+  DCHECK_NE(availability_state_, STATE_CLOSED);
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
   DCHECK_GE(session_unacked_recv_window_bytes_, 0);
   DCHECK_GE(session_recv_window_size_, session_unacked_recv_window_bytes_);
   DCHECK_GE(delta_window_size, 1);
   // Check for overflow.
   DCHECK_LE(delta_window_size, kint32max - session_recv_window_size_);
 
   session_recv_window_size_ += delta_window_size;
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
@@ skipping 12 matching lines @@
 void SpdySession::DecreaseRecvWindowSize(int32 delta_window_size) {
   CHECK(in_io_loop_);
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
   DCHECK_GE(delta_window_size, 1);
 
   // Since we never decrease the initial receive window size,
   // |delta_window_size| should never cause |recv_window_size_| to go
   // negative. If we do, the receive window isn't being respected.
   if (delta_window_size > session_recv_window_size_) {
     RecordProtocolErrorHistogram(PROTOCOL_ERROR_RECEIVE_WINDOW_VIOLATION);
-    DoDrainSession(
+    CloseSessionResult result = DoCloseSession(
         ERR_SPDY_PROTOCOL_ERROR,
         "delta_window_size is " + base::IntToString(delta_window_size) +
             " in DecreaseRecvWindowSize, which is larger than the receive " +
             "window size of " + base::IntToString(session_recv_window_size_));
+    DCHECK_EQ(result, SESSION_CLOSED_BUT_NOT_REMOVED);
     return;
   }
 
   session_recv_window_size_ -= delta_window_size;
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_RECV_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  -delta_window_size, session_recv_window_size_));
 }
 
 void SpdySession::QueueSendStalledStream(const SpdyStream& stream) {
   DCHECK(stream.send_stalled_by_flow_control());
   RequestPriority priority = stream.priority();
   CHECK_GE(priority, MINIMUM_PRIORITY);
   CHECK_LE(priority, MAXIMUM_PRIORITY);
   stream_send_unstall_queue_[priority].push_back(stream.stream_id());
 }
 
 void SpdySession::ResumeSendStalledStreams() {
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
 
   // We don't have to worry about new streams being queued, since
   // doing so would cause IsSendStalled() to return true. But we do
   // have to worry about streams being closed, as well as ourselves
   // being closed.
 
-  while (!IsSendStalled()) {
+  while (availability_state_ != STATE_CLOSED && !IsSendStalled()) {
     size_t old_size = 0;
 #if DCHECK_IS_ON
     old_size = GetTotalSize(stream_send_unstall_queue_);
 #endif
 
     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
@@ skipping 14 matching lines @@
     if (!queue->empty()) {
       SpdyStreamId stream_id = queue->front();
       queue->pop_front();
       return stream_id;
     }
   }
   return 0;
 }
 
 }  // namespace net