[SPDY] Refactor SpdyStream's handling of response headers

net/spdy/spdy_stream.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_stream.h"
 
 #include <limits>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
@@ skipping 26 matching lines @@
     int32 delta,
     int32 window_size,
     NetLog::LogLevel /* log_level */) {
   base::DictionaryValue* dict = new base::DictionaryValue();
   dict->SetInteger("stream_id", stream_id);
   dict->SetInteger("delta", delta);
   dict->SetInteger("window_size", window_size);
   return dict;
 }
 
-bool ContainsUpperAscii(const std::string& str) {
+bool ContainsUppercaseAscii(const std::string& str) {
   for (std::string::const_iterator i(str.begin()); i != str.end(); ++i) {
     if (*i >= 'A' && *i <= 'Z') {
       return true;
     }
   }
   return false;
 }
 
 }  // namespace
 
@@ skipping 24 matching lines @@
 SpdyStream::SpdyStream(SpdyStreamType type,
                        SpdySession* session,
                        const std::string& path,
                        RequestPriority priority,
                        int32 initial_send_window_size,
                        int32 initial_recv_window_size,
                        const BoundNetLog& net_log)
     : type_(type),
       weak_ptr_factory_(this),
       in_do_loop_(false),
-      continue_buffering_data_(true),
+      continue_buffering_data_(type_ == SPDY_PUSH_STREAM),
       stream_id_(0),
       path_(path),
       priority_(priority),
       slot_(0),
       send_stalled_by_flow_control_(false),
       send_window_size_(initial_send_window_size),
       recv_window_size_(initial_recv_window_size),
       unacked_recv_window_bytes_(0),
-      response_received_(false),
       session_(session),
       delegate_(NULL),
       send_status_(
           (type_ == SPDY_PUSH_STREAM) ?
           NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND),
       request_time_(base::Time::Now()),
-      response_(new SpdyHeaderBlock),
       io_state_((type_ == SPDY_PUSH_STREAM) ? STATE_OPEN : STATE_NONE),
       response_status_(OK),
       net_log_(net_log),
       send_bytes_(0),
       recv_bytes_(0),
       domain_bound_cert_type_(CLIENT_CERT_INVALID_TYPE),
       just_completed_frame_type_(DATA),
       just_completed_frame_size_(0) {
   CHECK(type_ == SPDY_BIDIRECTIONAL_STREAM ||
         type_ == SPDY_REQUEST_RESPONSE_STREAM ||
         type_ == SPDY_PUSH_STREAM);
 }
 
 SpdyStream::~SpdyStream() {
   CHECK(!in_do_loop_);
   UpdateHistograms();
 }
 
 void SpdyStream::SetDelegate(Delegate* delegate) {
+  Delegate* old_delegate = delegate_;

[Ryan Hamilton, 2013/06/19 18:58:36]

Is it "valid" to have more than one delegate in the lifetime of a spdy stream? 
I sure would not have expected that.  I'd be tempted to CHECK(!delegate_);

[akalin, 2013/06/21 21:13:25]

Yeap, in the redirect case in SpdyProxyClientSocket. Added bug and TODOs to fix
that.

   CHECK(delegate);

[Ryan Hamilton, 2013/06/19 18:58:36]

nit: can you make this the first line

[akalin, 2013/06/21 21:13:25]

Done.

   delegate_ = delegate;
 
-  if (type_ == SPDY_PUSH_STREAM) {
-    CHECK(response_received());
+  if (type_ == SPDY_PUSH_STREAM && !old_delegate) {
+    DCHECK(continue_buffering_data_);
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&SpdyStream::PushedStreamReplayData, GetWeakPtr()));
-  } else {
-    continue_buffering_data_ = false;
   }
 }
 
+SpdyStream::Delegate* SpdyStream::GetDelegate() {
+  return delegate_;
+}
+
 void SpdyStream::PushedStreamReplayData() {
+  DCHECK_EQ(type_, SPDY_PUSH_STREAM);
   DCHECK_NE(stream_id_, 0u);
-
-  if (!delegate_)
-    return;
+  DCHECK(continue_buffering_data_);
 
   continue_buffering_data_ = false;
 
-  // TODO(akalin): This call may delete this object. Figure out what
-  // to do in that case.
-  int rv = delegate_->OnResponseHeadersReceived(*response_, response_time_, OK);
+  // The delegate methods called below may delete |this|, so use
+  // |weak_this| to detect that.
+  base::WeakPtr<SpdyStream> weak_this = GetWeakPtr();
+
+  CHECK(delegate_);
+  CHECK(response_headers_);
+  int rv = delegate_->OnResponseHeadersUpdated(*response_headers_);
+  DCHECK(rv == OK || rv == ERR_INCOMPLETE_SPDY_HEADERS);
   if (rv == ERR_INCOMPLETE_SPDY_HEADERS) {
-    // We don't have complete headers.  Assume we're waiting for another
-    // HEADERS frame.  Since we don't have headers, we had better not have
-    // any pending data frames.
-    if (pending_buffers_.size() != 0U) {
+    // Since ERR_INCOMPLETE_SPDY_HEADERS was returned, we know that
+    // we're not closed/deleted. Since we don't have complete headers,
+    // assume we're waiting for another HEADERS frame, and we had
+    // better not have any pending data frames.
+    CHECK(weak_this);
+    if (!pending_buffers_.empty()) {
       LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
                      "HEADERS incomplete headers, but pending data frames.");
       session_->CloseActiveStream(stream_id_, ERR_SPDY_PROTOCOL_ERROR);
     }
     return;
   }
 
-  std::vector<SpdyBuffer*> buffers;
-  pending_buffers_.release(&buffers);
-  for (size_t i = 0; i < buffers.size(); ++i) {
-    // It is always possible that a callback to the delegate results in
-    // the delegate no longer being available.
-    if (!delegate_)
-      break;
-    if (buffers[i]) {
-      delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>(buffers[i]));
-    } else {
-      delegate_->OnDataReceived(scoped_ptr<SpdyBuffer>());
+  if (!weak_this)
+    return;
+
+  while (!pending_buffers_.empty()) {
+    // Take ownership of the first element of |pending_buffers_|.
+    scoped_ptr<SpdyBuffer> buffer(pending_buffers_.front());
+    pending_buffers_.weak_erase(pending_buffers_.begin());
+
+    bool eof = (buffer == NULL);
+
+    CHECK(delegate_);
+    delegate_->OnDataReceived(buffer.Pass());
+
+    if (!weak_this)
+      return;

[Ryan Hamilton, 2013/06/19 18:58:36]

do we need to make this check every time we invoke the delegate?  (are there
other places in the the file which might need this as well?)

[akalin, 2013/06/21 21:13:25]

Audited the other delegate methods and added comments, and also removed some
overly-conservative checks. (The in_do_loop_ check helps a lot).

+
+    if (eof) {
+      DCHECK(pending_buffers_.empty());
       session_->CloseActiveStream(stream_id_, OK);
-      // Note: |this| may be deleted after calling CloseActiveStream.
-      DCHECK_EQ(buffers.size() - 1, i);
+      DCHECK(!weak_this);
     }
   }
 }
 
 scoped_ptr<SpdyFrame> SpdyStream::ProduceSynStreamFrame() {
   CHECK_EQ(io_state_, STATE_SEND_REQUEST_HEADERS_COMPLETE);
-  CHECK(request_);
+  CHECK(request_headers_);
   CHECK_GT(stream_id_, 0u);
 
   SpdyControlFlags flags =
       (send_status_ == NO_MORE_DATA_TO_SEND) ?
       CONTROL_FLAG_FIN : CONTROL_FLAG_NONE;
   scoped_ptr<SpdyFrame> frame(session_->CreateSynStream(
-      stream_id_, priority_, slot_, flags, *request_));
+      stream_id_, priority_, slot_, flags, *request_headers_));
   send_time_ = base::TimeTicks::Now();
   return frame.Pass();
 }
 
 void SpdyStream::DetachDelegate() {
   CHECK(!in_do_loop_);
   DCHECK(!closed());
   delegate_ = NULL;
   Cancel();
 }
 
+bool SpdyStream::ReceivedInitialResponseHeaders() const {
+  return response_headers_ != NULL;
+}
+
 void SpdyStream::AdjustSendWindowSize(int32 delta_window_size) {
   DCHECK_GE(session_->flow_control_state(), SpdySession::FLOW_CONTROL_STREAM);
 
   if (closed())
     return;
 
   // Check for wraparound.
   if (send_window_size_ > 0) {
     DCHECK_LE(delta_window_size, kint32max - send_window_size_);
   }
@@ skipping 152 matching lines @@
 }
 
 base::Time SpdyStream::GetRequestTime() const {
   return request_time_;
 }
 
 void SpdyStream::SetRequestTime(base::Time t) {
   request_time_ = t;
 }
 
-int SpdyStream::OnResponseHeadersReceived(const SpdyHeaderBlock& response) {
-  int rv = OK;
-
-  metrics_.StartStream();
-
-  // TODO(akalin): This should be handled as a protocol error.
-  DCHECK(response_->empty());
-  *response_ = response;  // TODO(ukai): avoid copy.
-
-  recv_first_byte_time_ = base::TimeTicks::Now();
-  response_time_ = base::Time::Now();

[Ryan Hamilton, 2013/06/19 18:58:36]

I see that you pushed this logic up from the stream into the session, which
required widening the interface slightly and doing a tad more bookkeeping there.
 How come?

[akalin, 2013/06/21 21:13:25]

There was a TODO somewhere to record the response time as close as possible to
actually receiving the data. I figure it doesn't hurt to implement that.

+int SpdyStream::OnInitialResponseHeadersReceived(
+    const SpdyHeaderBlock& initial_response_headers,
+    base::Time response_time,
+    base::TimeTicks recv_first_byte_time) {
+  // SpdySession guarantees that this is called at most once.
+  CHECK(!response_headers_);
 
   // Check to make sure that we don't receive the response headers
   // before we're ready for it.
   switch (type_) {
     case SPDY_BIDIRECTIONAL_STREAM:
       // For a bidirectional stream, we're ready for the response
       // headers once we've finished sending the request headers.
       if (io_state_ < STATE_OPEN)
         return ERR_SPDY_PROTOCOL_ERROR;
       break;
 
     case SPDY_REQUEST_RESPONSE_STREAM:
       // For a request/response stream, we're ready for the response
       // headers once we've finished sending the request headers and
       // the request body (if we have one).
       if ((io_state_ < STATE_OPEN) || (send_status_ == MORE_DATA_TO_SEND) ||
           pending_send_data_.get())
         return ERR_SPDY_PROTOCOL_ERROR;
       break;
 
     case SPDY_PUSH_STREAM:
       // For a push stream, we're ready immediately.
       DCHECK_EQ(send_status_, NO_MORE_DATA_TO_SEND);
       DCHECK_EQ(io_state_, STATE_OPEN);
       break;
   }
 
+  metrics_.StartStream();

[Ryan Hamilton, 2013/06/19 18:58:36]

I notice that this moved to here.  This means that it will not be started in the
event of any of the early returns above.  Is that change significant?

[akalin, 2013/06/21 21:13:25]

I think so. That just means that erroneous streams won't be counted in bandwidth
metrics. I think that's okay, though, since they're closed almost immediately
after being opened.

+
   DCHECK_EQ(io_state_, STATE_OPEN);
 
-  // TODO(akalin): Merge the code below with the code in OnHeaders().
-
-  // Append all the headers into the response header block.
-  for (SpdyHeaderBlock::const_iterator it = response.begin();
-       it != response.end(); ++it) {
-    // Disallow uppercase headers.
-    if (ContainsUpperAscii(it->first)) {
-      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                            "Upper case characters in header: " + it->first);
-      return ERR_SPDY_PROTOCOL_ERROR;
-    }
-  }
-
-  if ((*response_).find("transfer-encoding") != (*response_).end()) {
-    session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                         "Received transfer-encoding header");
-    return ERR_SPDY_PROTOCOL_ERROR;
-  }
-
-  if (delegate_) {
-    // May delete this object.
-    rv = delegate_->OnResponseHeadersReceived(*response_, response_time_, rv);
-  }
-  // If delegate_ is not yet attached, we'll call
-  // OnResponseHeadersReceived after the delegate gets attached to the
-  // stream.
-
-  return rv;
+  response_headers_.reset(new SpdyHeaderBlock());
+  response_time_ = response_time;
+  recv_first_byte_time_ = recv_first_byte_time;
+  return MergeWithResponseHeaders(initial_response_headers);
 }
 
-int SpdyStream::OnHeaders(const SpdyHeaderBlock& headers) {
-  DCHECK(!response_->empty());
-
-  // Append all the headers into the response header block.
-  for (SpdyHeaderBlock::const_iterator it = headers.begin();
-      it != headers.end(); ++it) {
-    // Disallow duplicate headers.  This is just to be conservative.
-    if ((*response_).find(it->first) != (*response_).end()) {
-      LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "HEADERS duplicate header");
-      response_status_ = ERR_SPDY_PROTOCOL_ERROR;
-      return ERR_SPDY_PROTOCOL_ERROR;
-    }
-
-    // Disallow uppercase headers.
-    if (ContainsUpperAscii(it->first)) {
-      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                            "Upper case characters in header: " + it->first);
-      return ERR_SPDY_PROTOCOL_ERROR;
-    }
-
-    (*response_)[it->first] = it->second;
-  }
-
-  if ((*response_).find("transfer-encoding") != (*response_).end()) {
-    session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
-                         "Received transfer-encoding header");
-    return ERR_SPDY_PROTOCOL_ERROR;
-  }
-
-  int rv = OK;
-  if (delegate_) {
-    // May delete this object.
-    rv = delegate_->OnResponseHeadersReceived(*response_, response_time_, rv);
-    // ERR_INCOMPLETE_SPDY_HEADERS means that we are waiting for more
-    // headers before the response header block is complete.
-    if (rv == ERR_INCOMPLETE_SPDY_HEADERS)
-      rv = OK;
-  }
-  return rv;
+int SpdyStream::OnAdditionalResponseHeadersReceived(
+    const SpdyHeaderBlock& additional_response_headers) {
+  return MergeWithResponseHeaders(additional_response_headers);
 }
 
 void SpdyStream::OnDataReceived(scoped_ptr<SpdyBuffer> buffer) {
   DCHECK(session_->IsStreamActive(stream_id_));
   // If we don't have a response, then the SYN_REPLY did not come through.
   // We cannot pass data up to the caller unless the reply headers have been
   // received.
-  if (!response_received()) {
+  if (!response_headers_) {
     LogStreamError(ERR_SYN_REPLY_NOT_RECEIVED, "Didn't receive a response.");
     session_->CloseActiveStream(stream_id_, ERR_SYN_REPLY_NOT_RECEIVED);
     return;
   }
 
   if (!delegate_ || continue_buffering_data_) {
+    DCHECK_EQ(type_, SPDY_PUSH_STREAM);
     // It should be valid for this to happen in the server push case.
     // We'll return received data when delegate gets attached to the stream.
     if (buffer) {
       pending_buffers_.push_back(buffer.release());
     } else {
       pending_buffers_.push_back(NULL);
       metrics_.StopStream();
       // Note: we leave the stream open in the session until the stream
       //       is claimed.
     }
@@ skipping 76 matching lines @@
 
 void SpdyStream::Close() {
   CHECK(!in_do_loop_);
   if (stream_id_ != 0) {
     session_->CloseActiveStream(stream_id_, OK);
   } else {
     session_->CloseCreatedStream(GetWeakPtr(), OK);
   }
 }
 
-int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> headers,
+int SpdyStream::SendRequestHeaders(scoped_ptr<SpdyHeaderBlock> request_headers,
                                    SpdySendStatus send_status) {
   CHECK_NE(type_, SPDY_PUSH_STREAM);
   CHECK_EQ(send_status_, MORE_DATA_TO_SEND);
-  CHECK(!request_);
+  CHECK(!request_headers_);
   CHECK(!pending_send_data_.get());
   CHECK_EQ(io_state_, STATE_NONE);
-  request_ = headers.Pass();
+  request_headers_ = request_headers.Pass();
   send_status_ = send_status;
   io_state_ = STATE_GET_DOMAIN_BOUND_CERT;
   return DoLoop(OK);
 }
 
 void SpdyStream::SendData(IOBuffer* data,
                           int length,
                           SpdySendStatus send_status) {
   CHECK_NE(type_, SPDY_PUSH_STREAM);
   CHECK_EQ(send_status_, MORE_DATA_TO_SEND);
@@ skipping 27 matching lines @@
     QueueNextDataFrame();
   }
 }
 
 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
 
 bool SpdyStream::HasUrl() const {
   if (type_ == SPDY_PUSH_STREAM)
-    return response_received();
-  return request_ != NULL;
+    return response_headers_ != NULL;
+  return request_headers_ != NULL;
 }
 
 GURL SpdyStream::GetUrl() const {
   DCHECK(HasUrl());
 
   const SpdyHeaderBlock& headers =
-      (type_ == SPDY_PUSH_STREAM) ? *response_ : *request_;
+      (type_ == SPDY_PUSH_STREAM) ? *response_headers_ : *request_headers_;
   return GetUrlFromHeaderBlock(headers, GetProtocolVersion(),
                                type_ == SPDY_PUSH_STREAM);
 }
 
 void SpdyStream::OnGetDomainBoundCertComplete(int result) {
   DCHECK_EQ(io_state_, STATE_GET_DOMAIN_BOUND_CERT_COMPLETE);
   DoLoop(result);
 }
 
 int SpdyStream::DoLoop(int result) {
@@ skipping 51 matching lines @@
   } while (result != ERR_IO_PENDING && io_state_ != STATE_NONE &&
            io_state_ != STATE_OPEN);
 
   CHECK(in_do_loop_);
   in_do_loop_ = false;
 
   return result;
 }
 
 int SpdyStream::DoGetDomainBoundCert() {
-  CHECK(request_);
+  CHECK(request_headers_);
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   GURL url = GetUrl();
   if (!session_->NeedsCredentials() || !url.SchemeIs("https")) {
     // Proceed directly to sending the request headers
     io_state_ = STATE_SEND_REQUEST_HEADERS;
     return OK;
   }
 
   slot_ = session_->credential_state()->FindCredentialSlot(GetUrl());
   if (slot_ != SpdyCredentialState::kNoEntry) {
@@ skipping 19 matching lines @@
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   if (result != OK)
     return result;
 
   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT;
   slot_ =  session_->credential_state()->SetHasCredential(GetUrl());
   return OK;
 }
 
 int SpdyStream::DoSendDomainBoundCert() {
-  CHECK(request_);
+  CHECK(request_headers_);
   DCHECK_NE(type_, SPDY_PUSH_STREAM);
   io_state_ = STATE_SEND_DOMAIN_BOUND_CERT_COMPLETE;
 
   std::string origin = GetUrl().GetOrigin().spec();
   DCHECK(origin[origin.length() - 1] == '/');
   origin.erase(origin.length() - 1);  // Trim trailing slash.
   scoped_ptr<SpdyFrame> frame;
   int rv = session_->CreateCredentialFrame(
       origin, domain_bound_cert_type_, domain_bound_private_key_,
       domain_bound_cert_, priority_, &frame);
@@ skipping 195 matching lines @@
         base::Bind(&SpdyStream::OnWriteBufferConsumed,
                    GetWeakPtr(), payload_size));
   }
 
   session_->EnqueueStreamWrite(
       GetWeakPtr(), DATA,
       scoped_ptr<SpdyBufferProducer>(
           new SimpleBufferProducer(data_buffer.Pass())));
 }
 
+int SpdyStream::MergeWithResponseHeaders(
+    const SpdyHeaderBlock& new_response_headers) {
+  DCHECK(response_headers_);
+
+  if (new_response_headers.find("transfer-encoding") !=
+      new_response_headers.end()) {
+    session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
+                         "Received transfer-encoding header");
+    return ERR_SPDY_PROTOCOL_ERROR;
+  }
+
+  for (SpdyHeaderBlock::const_iterator it = new_response_headers.begin();
+      it != new_response_headers.end(); ++it) {
+    // Disallow uppercase headers.
+    if (ContainsUppercaseAscii(it->first)) {
+      session_->ResetStream(stream_id_, priority_, RST_STREAM_PROTOCOL_ERROR,
+                            "Upper case characters in header: " + it->first);
+      return ERR_SPDY_PROTOCOL_ERROR;
+    }
+
+    SpdyHeaderBlock::iterator it2 = response_headers_->lower_bound(it->first);
+    // Disallow duplicate headers.  This is just to be conservative.
+    if (it2 != response_headers_->end() && it2->first == it->first) {
+      LogStreamError(ERR_SPDY_PROTOCOL_ERROR, "HEADERS duplicate header");
+      response_status_ = ERR_SPDY_PROTOCOL_ERROR;
+      return ERR_SPDY_PROTOCOL_ERROR;
+    }
+
+    response_headers_->insert(it2, *it);
+  }
+
+  // If delegate_ is not yet attached, we'll call
+  // OnResponseHeadersUpdated() after the delegate gets attached to
+  // the stream.
+  if (delegate_) {
+    // The call to OnResponseHeadersUpdated() may delete this object
+    // if OK is returned, so store |type_| in a local variable.
+    SpdyStreamType type = type_;
+    int rv = delegate_->OnResponseHeadersUpdated(*response_headers_);
+    DCHECK(rv == OK || rv == ERR_INCOMPLETE_SPDY_HEADERS);

[Ryan Hamilton, 2013/06/19 18:58:36]

boy, this sure screams "this method should return a boolean"

[akalin, 2013/06/21 21:13:25]

Done.

+    // Incomplete headers are OK only for push streams.
+    if (type == SPDY_PUSH_STREAM && rv == ERR_INCOMPLETE_SPDY_HEADERS)
+      rv = OK;
+    if (rv != OK)
+      return rv;
+  }
+
+  return OK;
+}
+
 }  // namespace net