[SPDY] Change SpdyStream::QueueStreamData() To SendStreamData()

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 111 matching lines @@
       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),
       pushed_(pushed),
       response_received_(false),
       session_(session),
       delegate_(NULL),
+      pending_send_flags_(DATA_FLAG_NONE),
       request_time_(base::Time::Now()),
       response_(new SpdyHeaderBlock),
       io_state_(STATE_NONE),
       response_status_(OK),
       has_upload_data_(false),
       net_log_(net_log),
       send_bytes_(0),
       recv_bytes_(0),
       domain_bound_cert_type_(CLIENT_CERT_INVALID_TYPE),
       just_completed_frame_type_(DATA),
@@ skipping 476 matching lines @@
     DCHECK(!has_upload_data_);
     DCHECK(response_received());
     send_time_ = base::TimeTicks::Now();
     return ERR_IO_PENDING;
   }
   CHECK_EQ(STATE_NONE, io_state_);
   io_state_ = STATE_GET_DOMAIN_BOUND_CERT;
   return DoLoop(OK);
 }
 
-void SpdyStream::QueueHeaders(scoped_ptr<SpdyHeaderBlock> headers) {
+void SpdyStream::SendHeaders(scoped_ptr<SpdyHeaderBlock> headers) {
   // Until the first headers by SYN_STREAM have been completely sent, we can
   // not be sure that our stream_id is correct.
   DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
   CHECK_GT(stream_id_, 0u);
 
   session_->EnqueueStreamWrite(
       GetWeakPtr(), HEADERS,
       scoped_ptr<SpdyBufferProducer>(
           new HeaderBufferProducer(GetWeakPtr(), headers.Pass())));
 }
 
-void SpdyStream::QueueStreamData(IOBuffer* data,
-                                 int length,
-                                 SpdyDataFlags flags) {
-  // Until the headers have been completely sent, we can not be sure
-  // that our stream_id is correct.
-  DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
-  CHECK_GT(stream_id_, 0u);
-
-  scoped_ptr<SpdyBuffer> data_buffer(session_->CreateDataBuffer(
-      stream_id_, data, length, flags));
-  // We'll get called again by PossiblyResumeIfSendStalled().
-  if (!data_buffer)
-    return;
-
-  if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
-    DCHECK_GE(data_buffer->GetRemainingSize(),
-              session_->GetDataFrameMinimumSize());
-    size_t payload_size =
-        data_buffer->GetRemainingSize() - session_->GetDataFrameMinimumSize();
-    DCHECK_LE(payload_size, session_->GetDataFrameMaximumPayload());
-    DecreaseSendWindowSize(static_cast<int32>(payload_size));
-    // This currently isn't strictly needed, since write frames are
-    // discarded only if the stream is about to be closed. But have it
-    // here anyway just in case this changes.
-    data_buffer->AddConsumeCallback(
-        base::Bind(&SpdyStream::OnWriteBufferConsumed,
-                   GetWeakPtr(), payload_size));
-  }
-
-  session_->EnqueueStreamWrite(
-      GetWeakPtr(), DATA,
-      scoped_ptr<SpdyBufferProducer>(
-          new SimpleBufferProducer(data_buffer.Pass())));
+void SpdyStream::SendStreamData(IOBuffer* data,
+                                int length,
+                                SpdyDataFlags flags) {
+  CHECK(!pending_send_data_);
+  pending_send_data_ = new DrainableIOBuffer(data, length);
+  pending_send_flags_ = flags;
+  QueueNextDataFrame();
 }
 
 bool SpdyStream::GetSSLInfo(SSLInfo* ssl_info,
                             bool* was_npn_negotiated,
                             NextProto* protocol_negotiated) {
   return session_->GetSSLInfo(
       ssl_info, was_npn_negotiated, protocol_negotiated);
 }
 
 bool SpdyStream::GetSSLCertRequestInfo(SSLCertRequestInfo* cert_request_info) {
   return session_->GetSSLCertRequestInfo(cert_request_info);
 }
 
 void SpdyStream::PossiblyResumeIfSendStalled() {
   DCHECK(!closed());
 
   if (send_stalled_by_flow_control_ && !session_->IsSendStalled() &&
       send_window_size_ > 0) {
     net_log_.AddEvent(
         NetLog::TYPE_SPDY_STREAM_FLOW_CONTROL_UNSTALLED,
         NetLog::IntegerCallback("stream_id", stream_id_));
     send_stalled_by_flow_control_ = false;
-    io_state_ = STATE_SEND_BODY;
-    DoLoop(OK);
+    QueueNextDataFrame();
   }
 }
 
 base::WeakPtr<SpdyStream> SpdyStream::GetWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
 
 bool SpdyStream::HasUrl() const {
   if (pushed_)
     return response_received();
@@ skipping 189 matching lines @@
   io_state_ =
       (delegate_->OnSendHeadersComplete() == MORE_DATA_TO_SEND) ?
       STATE_SEND_BODY : STATE_WAITING_FOR_RESPONSE;
 
   return OK;
 }
 
 // DoSendBody is called to send the optional body for the request.  This call
 // will also be called as each write of a chunk of the body completes.
 int SpdyStream::DoSendBody() {
-  // If we're already in the STATE_SEND_BODY state, then we've already
-  // sent a portion of the body.  In that case, we need to first consume
-  // the bytes written in the body stream.  Note that the bytes written is
-  // the number of bytes in the frame that were written, only consume the
-  // data portion, of course.
   io_state_ = STATE_SEND_BODY_COMPLETE;
   CHECK(delegate_);
   delegate_->OnSendBody();
   return ERR_IO_PENDING;
 }
 
 int SpdyStream::DoSendBodyComplete(int result) {
   if (result != OK)
     return result;
 
   if (just_completed_frame_type_ != DATA) {
     NOTREACHED();
     return ERR_UNEXPECTED;
   }
 
   if (just_completed_frame_size_ < session_->GetDataFrameMinimumSize()) {
     NOTREACHED();
     return ERR_UNEXPECTED;
   }
 
   size_t frame_payload_size =
       just_completed_frame_size_ - session_->GetDataFrameMinimumSize();
   if (frame_payload_size > session_->GetDataFrameMaximumPayload()) {
     NOTREACHED();
     return ERR_UNEXPECTED;
   }
 
+  send_bytes_ += frame_payload_size;
+
+  pending_send_data_->DidConsume(frame_payload_size);
+  if (pending_send_data_->BytesRemaining() > 0) {
+    io_state_ = STATE_SEND_BODY_COMPLETE;
+    QueueNextDataFrame();
+    return ERR_IO_PENDING;
+  }
+
+  pending_send_data_ = NULL;
+  pending_send_flags_ = DATA_FLAG_NONE;
+
   if (!delegate_) {
     NOTREACHED();
     return ERR_UNEXPECTED;
   }
 
-  send_bytes_ += frame_payload_size;
-
   io_state_ =
-      (delegate_->OnSendBodyComplete(frame_payload_size) == MORE_DATA_TO_SEND) ?
+      (delegate_->OnSendBodyComplete() == MORE_DATA_TO_SEND) ?
       STATE_SEND_BODY : STATE_WAITING_FOR_RESPONSE;
 
   return OK;
 }
 
 int SpdyStream::DoOpen() {
   io_state_ = STATE_OPEN;
 
   switch (just_completed_frame_type_) {
     case DATA: {
       if (just_completed_frame_size_ < session_->GetDataFrameMinimumSize()) {
         NOTREACHED();
         return ERR_UNEXPECTED;
       }
 
       size_t frame_payload_size =
           just_completed_frame_size_ - session_->GetDataFrameMinimumSize();
       if (frame_payload_size > session_->GetDataFrameMaximumPayload()) {
         NOTREACHED();
         return ERR_UNEXPECTED;
       }
 
       send_bytes_ += frame_payload_size;
+
+      pending_send_data_->DidConsume(frame_payload_size);
+      if (pending_send_data_->BytesRemaining() > 0) {
+        QueueNextDataFrame();
+        return ERR_IO_PENDING;
+      }
+
+      pending_send_data_ = NULL;
+      pending_send_flags_ = DATA_FLAG_NONE;

[Ryan Hamilton, 2013/05/23 04:12:00]

It's a pity that lines 948-957 are almost completely duplicates of 907-917.  Can
you factor it out in some way?  What do you think?

[akalin, 2013/05/23 05:29:38]

I did think about this. The problem is, factoring it into a helper function now
would be a bit awkward (would have to check the return value of such a function,
it would non-obviously have to set io_state_, etc.). However, I am working
towards simplifying the state machine, so one of the copies might go away
eventually. If not, then I'll factor it out then.

[Ryan Hamilton, 2013/05/23 16:26:30]

Sounds great.  I thought about a helper that returned a bool and then ... and
then I decided that didn't seem awesome so I thought, "Perhaps Fred has a better
idea" :>  As long as you've thought about this, I'm happy.

+
       if (delegate_)
-        delegate_->OnDataSent(frame_payload_size);
+        delegate_->OnDataSent();
+
       break;
     }
 
     case HEADERS:
       if (delegate_)
         delegate_->OnHeadersSent();
       break;
 
     default:
       NOTREACHED();
@@ skipping 13 matching lines @@
       recv_first_byte_time_ - send_time_);
   UMA_HISTOGRAM_TIMES("Net.SpdyStreamDownloadTime",
       recv_last_byte_time_ - recv_first_byte_time_);
   UMA_HISTOGRAM_TIMES("Net.SpdyStreamTime",
       recv_last_byte_time_ - send_time_);
 
   UMA_HISTOGRAM_COUNTS("Net.SpdySendBytes", send_bytes_);
   UMA_HISTOGRAM_COUNTS("Net.SpdyRecvBytes", recv_bytes_);
 }
 
+void SpdyStream::QueueNextDataFrame() {
+  // Until the headers have been completely sent, we can not be sure
+  // that our stream_id is correct.
+  DCHECK_GT(io_state_, STATE_SEND_HEADERS_COMPLETE);
+  CHECK_GT(stream_id_, 0u);
+  CHECK(pending_send_data_);
+  CHECK_GT(pending_send_data_->BytesRemaining(), 0);
+
+  scoped_ptr<SpdyBuffer> data_buffer(session_->CreateDataBuffer(
+      stream_id_,
+      pending_send_data_, pending_send_data_->BytesRemaining(),
+      pending_send_flags_));
+  // We'll get called again by PossiblyResumeIfSendStalled().
+  if (!data_buffer)
+    return;
+
+  if (session_->flow_control_state() >= SpdySession::FLOW_CONTROL_STREAM) {
+    DCHECK_GE(data_buffer->GetRemainingSize(),
+              session_->GetDataFrameMinimumSize());
+    size_t payload_size =
+        data_buffer->GetRemainingSize() - session_->GetDataFrameMinimumSize();
+    DCHECK_LE(payload_size, session_->GetDataFrameMaximumPayload());
+    DecreaseSendWindowSize(static_cast<int32>(payload_size));
+    // This currently isn't strictly needed, since write frames are
+    // discarded only if the stream is about to be closed. But have it
+    // here anyway just in case this changes.
+    data_buffer->AddConsumeCallback(
+        base::Bind(&SpdyStream::OnWriteBufferConsumed,
+                   GetWeakPtr(), payload_size));
+  }
+
+  session_->EnqueueStreamWrite(
+      GetWeakPtr(), DATA,
+      scoped_ptr<SpdyBufferProducer>(
+          new SimpleBufferProducer(data_buffer.Pass())));
+}
+
 }  // namespace net