Make net::BidirectionalStream handle RST_STREAM_NO_ERROR

net/spdy/bidirectional_stream_spdy_impl.cc

 // Copyright 2015 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/bidirectional_stream_spdy_impl.h"
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/time/time.h"
@@ skipping 16 matching lines @@
 }  // namespace
 
 BidirectionalStreamSpdyImpl::BidirectionalStreamSpdyImpl(
     const base::WeakPtr<SpdySession>& spdy_session)
     : spdy_session_(spdy_session),
       request_info_(nullptr),
       delegate_(nullptr),
       negotiated_protocol_(kProtoUnknown),
       more_read_data_pending_(false),
       read_buffer_len_(0),
+      written_end_of_stream_(false),
+      write_pending_(false),
       stream_closed_(false),
       closed_stream_status_(ERR_FAILED),
       closed_stream_received_bytes_(0),
       closed_stream_sent_bytes_(0),
       closed_has_load_timing_info_(false),
       weak_factory_(this) {}
 
 BidirectionalStreamSpdyImpl::~BidirectionalStreamSpdyImpl() {
   // Sends a RST to the remote if the stream is destroyed before it completes.
   ResetStream();
+  DCHECK(!write_pending_);
 }
 
 void BidirectionalStreamSpdyImpl::Start(
     const BidirectionalStreamRequestInfo* request_info,
     const NetLogWithSource& net_log,
     bool /*send_request_headers_automatically*/,
     BidirectionalStreamImpl::Delegate* delegate,
     std::unique_ptr<base::Timer> timer) {
   DCHECK(!stream_);
   DCHECK(timer);
@@ skipping 43 matching lines @@
   // called upon completion.
   read_buffer_ = buf;
   read_buffer_len_ = buf_len;
   return ERR_IO_PENDING;
 }
 
 void BidirectionalStreamSpdyImpl::SendData(const scoped_refptr<IOBuffer>& data,
                                            int length,
                                            bool end_stream) {
   DCHECK(length > 0 || (length == 0 && end_stream));
+  DCHECK(!write_pending_);
 

[kapishnikov, 2016/10/31 17:03:07]

Should we add DCHECK(!written_end_of_stream_) here or will it break some of our
tests?

[xunjieli, 2016/10/31 17:18:03]

The newly added test will crash instead of having an error.

-  if (!stream_) {
-    LOG(ERROR) << "Trying to send data after stream has been destroyed.";
-    base::ThreadTaskRunnerHandle::Get()->PostTask(
-        FROM_HERE, base::Bind(&BidirectionalStreamSpdyImpl::NotifyError,
-                              weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
+  write_pending_ = true;
+  if (MaybeHandleStreamClosedInSendData()) {
+    written_end_of_stream_ = end_stream;
     return;
   }
 
   DCHECK(!stream_closed_);
+  written_end_of_stream_ = end_stream;
+
   stream_->SendData(data.get(), length,
                     end_stream ? NO_MORE_DATA_TO_SEND : MORE_DATA_TO_SEND);
 }
 
 void BidirectionalStreamSpdyImpl::SendvData(
     const std::vector<scoped_refptr<IOBuffer>>& buffers,
     const std::vector<int>& lengths,
     bool end_stream) {
   DCHECK_EQ(buffers.size(), lengths.size());
+  DCHECK(!write_pending_);
 

[kapishnikov, 2016/10/31 17:03:07]

Same DCHECK(!written_end_of_stream_) here

[xunjieli, 2016/10/31 17:18:02]

Same as above. The newly added test will crash. We either have a DCHECk crash or
an OnFailed() failure. I am fine with going either way.

-  if (!stream_) {
-    LOG(ERROR) << "Trying to send data after stream has been destroyed.";
-    base::ThreadTaskRunnerHandle::Get()->PostTask(
-        FROM_HERE, base::Bind(&BidirectionalStreamSpdyImpl::NotifyError,
-                              weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
+  write_pending_ = true;
+  if (MaybeHandleStreamClosedInSendData()) {
+    written_end_of_stream_ = end_stream;
     return;
   }
 
   DCHECK(!stream_closed_);
+  written_end_of_stream_ = end_stream;
+
   int total_len = 0;
   for (int len : lengths) {
     total_len += len;
   }
 
   pending_combined_buffer_ = new net::IOBuffer(total_len);
   int len = 0;
   // TODO(xunjieli): Get rid of extra copy. Coalesce headers and data frames.
   for (size_t i = 0; i < buffers.size(); ++i) {
     memcpy(pending_combined_buffer_->data() + len, buffers[i]->data(),
@@ skipping 112 matching lines @@
 
   if (status != OK) {
     NotifyError(status);
     return;
   }
   ResetStream();
   // Complete any remaining read, as all data has been buffered.
   // If user has not called ReadData (i.e |read_buffer_| is nullptr), this will
   // do nothing.
   timer_->Stop();
+
+  // |this| might get destroyed after calling into |delegate_| in
+  // DoBufferedRead().
+  auto weak_this = weak_factory_.GetWeakPtr();
   DoBufferedRead();
+  if (weak_this.get() && write_pending_)
+    NotifyDataSent();
 }
 
 int BidirectionalStreamSpdyImpl::SendRequestHeadersHelper() {
   SpdyHeaderBlock headers;
   HttpRequestInfo http_request_info;
   http_request_info.url = request_info_->url;
   http_request_info.method = request_info_->method;
   http_request_info.extra_headers = request_info_->extra_headers;
 
   CreateSpdyHeadersFromHttpRequest(
       http_request_info, http_request_info.extra_headers, true, &headers);
+  written_end_of_stream_ = request_info_->end_stream_on_headers;
   return stream_->SendRequestHeaders(std::move(headers),
                                      request_info_->end_stream_on_headers
                                          ? NO_MORE_DATA_TO_SEND
                                          : MORE_DATA_TO_SEND);
 }
 
 void BidirectionalStreamSpdyImpl::OnStreamInitialized(int rv) {
   DCHECK_NE(ERR_IO_PENDING, rv);
   if (rv == OK) {
     stream_ = stream_request_.ReleaseStream();
     stream_->SetDelegate(this);
     rv = SendRequestHeadersHelper();
     if (rv == OK) {
       OnRequestHeadersSent();
       return;
     } else if (rv == ERR_IO_PENDING) {
       return;
     }
   }
   NotifyError(rv);
 }
 
 void BidirectionalStreamSpdyImpl::NotifyError(int rv) {
   ResetStream();
+  write_pending_ = false;
   if (delegate_) {
     BidirectionalStreamImpl::Delegate* delegate = delegate_;
     delegate_ = nullptr;
     // Cancel any pending callback.
     weak_factory_.InvalidateWeakPtrs();
     delegate->OnFailed(rv);
     // |this| can be null when returned from delegate.
   }
 }
 
+void BidirectionalStreamSpdyImpl::NotifyDataSent() {
+  DCHECK(write_pending_);
+
+  write_pending_ = false;
+  if (delegate_)
+    delegate_->OnDataSent();
+}
+
 void BidirectionalStreamSpdyImpl::ResetStream() {
   if (!stream_)
     return;
   if (!stream_->IsClosed()) {
     // This sends a RST to the remote.
     stream_->DetachDelegate();
     DCHECK(!stream_);
   } else {
     // Stream is already closed, so it is not legal to call DetachDelegate.
     stream_.reset();
@@ skipping 39 matching lines @@
 }
 
 bool BidirectionalStreamSpdyImpl::ShouldWaitForMoreBufferedData() const {
   if (stream_closed_)
     return false;
   DCHECK_GT(read_buffer_len_, 0);
   return read_data_queue_.GetTotalSize() <
          static_cast<size_t>(read_buffer_len_);
 }
 
+bool BidirectionalStreamSpdyImpl::MaybeHandleStreamClosedInSendData() {
+  if (stream_)
+    return false;
+  // If |stream_| is closed without an error before client half closes,
+  // blackhole any pending write data. crbug.com/650438.
+  if (!written_end_of_stream_ && stream_closed_ &&

[kapishnikov, 2016/10/31 17:03:07]

To make the code easier to understand, I think we should break the 'if' into 2
different cases:

1) When the client calls SendData() with end_stream == true second time, we
should return an error with the corresponding corresponding log message.
2) When SendData() is executed on a closed stream (as before).

Some can assume that if the client called SendData() with end_stream == true,
written_end_of_stream_ is always et to true. However, it is not always the case.

[xunjieli, 2016/10/31 17:18:02]

Done. 

+      closed_stream_status_ == OK) {
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(&BidirectionalStreamSpdyImpl::NotifyDataSent,
+                              weak_factory_.GetWeakPtr()));
+    return true;
+  }
+  LOG(ERROR) << "Trying to send data after stream has been destroyed.";
+  base::ThreadTaskRunnerHandle::Get()->PostTask(
+      FROM_HERE, base::Bind(&BidirectionalStreamSpdyImpl::NotifyError,
+                            weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
+  return true;
+}
+
 }  // namespace net