Remove QuicChromiumClientStream::Delegate in favor of async methods.

net/quic/chromium/quic_chromium_client_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/quic/chromium/quic_chromium_client_stream.h"
 
 #include <utility>
 
 #include "base/bind_helpers.h"
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/log/net_log_event_type.h"
 #include "net/quic/chromium/quic_chromium_client_session.h"
 #include "net/quic/chromium/quic_http_utils.h"
 #include "net/quic/core/quic_spdy_session.h"
 #include "net/quic/core/quic_write_blocked_list.h"
 #include "net/quic/core/spdy_utils.h"
 
 namespace net {
+namespace {
+// Sets a boolean to a value, and restores it to the previous value once
+// the saver goes out of scope.
+class ScopedBoolSaver {
+ public:
+  ScopedBoolSaver(bool* var, bool new_val) : var_(var), old_val_(*var) {
+    *var_ = new_val;
+  }
 
-QuicChromiumClientStream::Handle::Handle(QuicChromiumClientStream* stream,
-                                         Delegate* delegate)
+  ~ScopedBoolSaver() { *var_ = old_val_; }
+
+ private:
+  bool* var_;
+  bool old_val_;
+};
+}  // namespace
+
+QuicChromiumClientStream::Handle::Handle(QuicChromiumClientStream* stream)
     : stream_(stream),
-      delegate_(delegate),
+      may_invoke_callbacks_(true),
       read_headers_buffer_(nullptr),
-      read_body_buffer_len_(0) {
+      read_body_buffer_len_(0),
+      net_error_(ERR_UNEXPECTED),
+      weak_factory_(this) {
   SaveState();
 }
 
 QuicChromiumClientStream::Handle::~Handle() {
   if (stream_) {
     stream_->ClearHandle();
     // TODO(rch): If stream_ is still valid, it should probably be Reset()
     // so that it does not leak.
     // stream_->Reset(QUIC_STREAM_CANCELLED);
   }
 }
 
-void QuicChromiumClientStream::Handle::ClearDelegate() {
-  delegate_ = nullptr;
-}
-
 void QuicChromiumClientStream::Handle::OnInitialHeadersAvailable() {
   if (!read_headers_callback_)
     return;  // Wait for ReadInitialHeaders to be called.
 
   int rv = ERR_QUIC_PROTOCOL_ERROR;
   if (!stream_->DeliverInitialHeaders(read_headers_buffer_, &rv))
     rv = ERR_QUIC_PROTOCOL_ERROR;
 
-  ResetAndReturn(&read_headers_callback_).Run(rv);
+  ResetAndRun(&read_headers_callback_, rv);
 }
 
 void QuicChromiumClientStream::Handle::OnTrailingHeadersAvailable() {
   if (!read_headers_callback_)
     return;  // Wait for ReadInitialHeaders to be called.
 
   int rv = ERR_QUIC_PROTOCOL_ERROR;
   if (!stream_->DeliverTrailingHeaders(read_headers_buffer_, &rv))
     rv = ERR_QUIC_PROTOCOL_ERROR;
 
-  ResetAndReturn(&read_headers_callback_).Run(rv);
+  ResetAndRun(&read_headers_callback_, rv);
 }
 
 void QuicChromiumClientStream::Handle::OnDataAvailable() {
   if (!read_body_callback_)
     return;  // Wait for ReadBody to be called.
 
   int rv = stream_->Read(read_body_buffer_, read_body_buffer_len_);
   if (rv == ERR_IO_PENDING)
     return;  // Spurrious, likely because of trailers?
 
   read_body_buffer_ = nullptr;
   read_body_buffer_len_ = 0;
-  ResetAndReturn(&read_body_callback_).Run(rv);
+  ResetAndRun(&read_body_callback_, rv);
 }
 
 void QuicChromiumClientStream::Handle::OnCanWrite() {
   if (!write_callback_)
     return;
 
-  base::ResetAndReturn(&write_callback_).Run(OK);
+  ResetAndRun(&write_callback_, OK);
 }
 
 void QuicChromiumClientStream::Handle::OnClose() {
+  if (net_error_ == ERR_UNEXPECTED) {
+    if (stream_error() == QUIC_STREAM_NO_ERROR &&
+        connection_error() == QUIC_NO_ERROR && fin_sent() && fin_received()) {
+      net_error_ = ERR_CONNECTION_CLOSED;
+    } else {
+      net_error_ = ERR_QUIC_PROTOCOL_ERROR;
+    }
+  }
+  OnError(net_error_);
+}
+
+void QuicChromiumClientStream::Handle::OnError(int error) {
+  net_error_ = error;
   if (stream_)
     SaveState();
   stream_ = nullptr;
-  if (delegate_) {
-    auto* delegate = delegate_;
-    delegate_ = nullptr;
-    delegate->OnClose();
-  }
+
+  // Post a task to invoke the callbacks to ensure that there is no reentrancy.
+  // A ScopedPacketBundler might cause an error which closes the stream under
+  // the call stack of the owner of the handle.
+  base::ThreadTaskRunnerHandle::Get()->PostTask(

[xunjieli, 2017/05/31 00:25:22]

Is this PostTask necessary? 
The callback shouldn't call into Handle again if it's being called with an
error, right? If so, we should be able to get rid of the PostTask and just
invoke the callback here.

[Ryan Hamilton, 2017/05/31 02:49:56]

The problem is not that the caller might call into the handle again, but that
the handle might call into the caller when the caller is still on the stack. I
did not have a PostTask here until I started working on a CL to bundle POST
headers together with body by use of a scoped packet bundler. In this instance,
we can do:

DoPost() {
// do stuff
{
 ScopedPacketBundler;
 handle_->WriteHeaders() => OK
 handle_->WriteData(callback) => ERR_IO_PENDING
}
// do more stuff
}

In this case, when the bundler goes out of scope, it can cause the write to
flush and the connection to close because of a write error. This write error
then triggers the callbacks to be executed. But this violates the reentrancy
guarantee that the callback will not be executed while under the callstack of
the caller. 

This is super annoying. Buck and I spent quite a while talking through this and
this was the best we came up with. Happy to do something better though!

[xunjieli, 2017/06/01 15:30:32]

Acknowledged. You are totally right. I missed the part where Handle::OnClose()
can be called re-entrantly. 

+      FROM_HERE,
+      base::Bind(&QuicChromiumClientStream::Handle::InvokeCallbacksOnError,
+                 weak_factory_.GetWeakPtr(), error));
 }
 
-void QuicChromiumClientStream::Handle::OnError(int error) {
-  if (stream_)
-    SaveState();
-  stream_ = nullptr;
-  if (delegate_) {
-    auto* delegate = delegate_;
-    delegate_ = nullptr;
-    delegate->OnError(error);
+void QuicChromiumClientStream::Handle::InvokeCallbacksOnError(int error) {
+  // Invoking a callback may cause |this| to be deleted. If this happens, no
+  // more callbacks should be invoked. Guard against this by holding a WeakPtr
+  // to |this| and ensuring it's still valid.
+  auto guard(weak_factory_.GetWeakPtr());
+  for (auto* callback :
+       {&read_headers_callback_, &read_body_callback_, &write_callback_}) {
+    if (*callback)
+      ResetAndRun(callback, error);
+    if (!guard.get())
+      return;
   }
 }
 
 int QuicChromiumClientStream::Handle::ReadInitialHeaders(
     SpdyHeaderBlock* header_block,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   int frame_len = 0;
   if (stream_->DeliverInitialHeaders(header_block, &frame_len))
     return frame_len;
 
   read_headers_buffer_ = header_block;
-  read_headers_callback_ = callback;
+  SetCallback(callback, &read_headers_callback_);
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::ReadBody(
     IOBuffer* buffer,
     int buffer_len,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
+  if (IsDoneReading())
+    return OK;
+
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   int rv = stream_->Read(buffer, buffer_len);
   if (rv != ERR_IO_PENDING)
     return rv;
 
-  read_body_callback_ = callback;
+  SetCallback(callback, &read_body_callback_);
   read_body_buffer_ = buffer;
   read_body_buffer_len_ = buffer_len;
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::ReadTrailingHeaders(
     SpdyHeaderBlock* header_block,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   int frame_len = 0;
   if (stream_->DeliverTrailingHeaders(header_block, &frame_len))
     return frame_len;
 
   read_headers_buffer_ = header_block;
-  read_headers_callback_ = callback;
+  SetCallback(callback, &read_headers_callback_);
   return ERR_IO_PENDING;
 }
 
-size_t QuicChromiumClientStream::Handle::WriteHeaders(
+int QuicChromiumClientStream::Handle::WriteHeaders(
     SpdyHeaderBlock header_block,
     bool fin,
     QuicReferenceCountedPointer<QuicAckListenerInterface>
         ack_notifier_delegate) {
   if (!stream_)
     return 0;
-  return stream_->WriteHeaders(std::move(header_block), fin,
-                               ack_notifier_delegate);
+  return HandleIOComplete(stream_->WriteHeaders(std::move(header_block), fin,

[xunjieli, 2017/05/31 00:25:22]

Doesn't stream_->WriteHeaders() return a size_t ? Should we check that before
invoking HandleIOComplete() which accepts a net error code?

[Ryan Hamilton, 2017/05/31 02:49:56]

What should we check?

+                                                ack_notifier_delegate));
 }
 
 int QuicChromiumClientStream::Handle::WriteStreamData(
     base::StringPiece data,
     bool fin,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   if (stream_->WriteStreamData(data, fin))
-    return OK;
+    return HandleIOComplete(OK);
 
-  write_callback_ = callback;
+  SetCallback(callback, &write_callback_);
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::WritevStreamData(
     const std::vector<scoped_refptr<IOBuffer>>& buffers,
     const std::vector<int>& lengths,
     bool fin,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   if (stream_->WritevStreamData(buffers, lengths, fin))
-    return OK;
+    return HandleIOComplete(OK);
 
-  write_callback_ = callback;
+  SetCallback(callback, &write_callback_);
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::Read(IOBuffer* buf, int buf_len) {
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
   return stream_->Read(buf, buf_len);
 }
 
 void QuicChromiumClientStream::Handle::OnFinRead() {
+  read_headers_callback_.Reset();

[xunjieli, 2017/05/31 00:25:22]

Should we also reset the read trailers callback and read body callback here?

[Ryan Hamilton, 2017/05/31 02:49:56]

There is no read trailers callback, only read_headers_callback_ which is set
when ReadInitialHeaders() is called and when ReadTrailingHeaders() is called.
(You can't read trailers until you've read initial headers, of course)

[xunjieli, 2017/06/01 15:30:32]

Got it. Thanks for the details!

   if (stream_)
     stream_->OnFinRead();
 }
 
 void QuicChromiumClientStream::Handle::DisableConnectionMigration() {
   if (stream_)
     stream_->DisableConnectionMigration();
 }
 
 void QuicChromiumClientStream::Handle::SetPriority(SpdyPriority priority) {
@@ skipping 79 matching lines @@
     return priority_;
   return stream_->priority();
 }
 
 bool QuicChromiumClientStream::Handle::can_migrate() {
   if (!stream_)
     return false;
   return stream_->can_migrate();
 }
 
-QuicChromiumClientStream::Delegate*
-QuicChromiumClientStream::Handle::GetDelegate() {
-  return delegate_;
-}
-
 void QuicChromiumClientStream::Handle::SaveState() {
   DCHECK(stream_);
   fin_sent_ = stream_->fin_sent();
   fin_received_ = stream_->fin_received();
   num_bytes_consumed_ = stream_->sequencer()->NumBytesConsumed();
   id_ = stream_->id();
   connection_error_ = stream_->connection_error();
   stream_error_ = stream_->stream_error();
   is_done_reading_ = stream_->IsDoneReading();
   is_first_stream_ = stream_->IsFirstStream();
   stream_bytes_read_ = stream_->stream_bytes_read();
   stream_bytes_written_ = stream_->stream_bytes_written();
   priority_ = stream_->priority();
 }
 
+void QuicChromiumClientStream::Handle::SetCallback(
+    CompletionCallback new_callback,
+    CompletionCallback* callback) {
+  CHECK(!may_invoke_callbacks_);

[xunjieli, 2017/05/31 00:25:22]

Should we add a TODO to convert this CHECK to DCHECK once the API is stable?

[Ryan Hamilton, 2017/05/31 02:49:56]

Done.

+  *callback = new_callback;
+}
+
+void QuicChromiumClientStream::Handle::ResetAndRun(CompletionCallback* callback,
+                                                   int rv) {
+  CHECK(may_invoke_callbacks_);
+  ResetAndReturn(callback).Run(rv);
+}
+
+int QuicChromiumClientStream::Handle::HandleIOComplete(int rv) {
+  if (rv < 0 || stream_)

[xunjieli, 2017/05/31 00:25:22]

Why do we need the |stream_| check? Could you add a comment?
It's not immediately obvious because |stream_| is not being deferenced in this
function.

[Ryan Hamilton, 2017/05/31 02:49:56]

If |stream_| is still valid the stream has not been closed. if the stream has
not been closed, then we don't need to return the close error code. Does that
make sense?

+    return rv;
+
+  if (stream_error_ == QUIC_STREAM_NO_ERROR &&
+      connection_error_ == QUIC_NO_ERROR && fin_sent_ && fin_received_) {
+    return rv;
+  }
+
+  return net_error_;
+}
+
 QuicChromiumClientStream::QuicChromiumClientStream(
     QuicStreamId id,
     QuicClientSessionBase* session,
     const NetLogWithSource& net_log)
     : QuicSpdyStream(id, session),
       net_log_(net_log),
       handle_(nullptr),
       headers_delivered_(false),
       initial_headers_sent_(false),
       session_(session),
@@ skipping 139 matching lines @@
   // Writes the data, or buffers it.
   for (size_t i = 0; i < buffers.size(); ++i) {
     bool is_fin = fin && (i == buffers.size() - 1);
     QuicStringPiece string_data(buffers[i]->data(), lengths[i]);
     WriteOrBufferData(string_data, is_fin, nullptr);
   }
   return !HasBufferedData();  // Was all data written?
 }
 
 std::unique_ptr<QuicChromiumClientStream::Handle>
-QuicChromiumClientStream::CreateHandle(
-    QuicChromiumClientStream::Delegate* delegate) {
+QuicChromiumClientStream::CreateHandle() {
   DCHECK(!handle_);
   auto handle = std::unique_ptr<QuicChromiumClientStream::Handle>(
-      new QuicChromiumClientStream::Handle(this, delegate));
+      new QuicChromiumClientStream::Handle(this));
   handle_ = handle.get();
 
   // Should this perhaps be via PostTask to make reasoning simpler?
   if (!initial_headers_.empty())
     handle_->OnInitialHeadersAvailable();
 
   return handle;
 }
 
 void QuicChromiumClientStream::ClearHandle() {
@@ skipping 48 matching lines @@
       base::Bind(
           &QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable,
           weak_factory_.GetWeakPtr()));
 }
 
 void QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable() {
   if (!handle_)
     return;
 
   DCHECK(headers_delivered_);
-  // Post an async task to notify delegate of the FIN flag.
+  // Post an async task to notify handle of the FIN flag.
   NotifyHandleOfDataAvailableLater();
   handle_->OnTrailingHeadersAvailable();
 }
 
 bool QuicChromiumClientStream::DeliverInitialHeaders(SpdyHeaderBlock* headers,
                                                      int* frame_len) {
   if (initial_headers_.empty())
     return false;
 
   headers_delivered_ = true;
@@ skipping 37 matching lines @@
 
 void QuicChromiumClientStream::DisableConnectionMigration() {
   can_migrate_ = false;
 }
 
 bool QuicChromiumClientStream::IsFirstStream() {
   return id() == kHeadersStreamId + 2;
 }
 
 }  // namespace net