Add an async ReadInitialHeaders method to QuicChromiumClientStream::Handle

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;
+  }
+
+  ~ScopedBoolSaver() { *var_ = old_val_; }
+
+ private:
+  bool* var_;
+  bool old_val_;
+};
+
+}  // namespace
+
 QuicChromiumClientStream::Handle::Handle(QuicChromiumClientStream* stream,
                                          Delegate* delegate)
-    : stream_(stream), delegate_(delegate) {
+    : stream_(stream),
+      delegate_(delegate),
+      can_invoke_callbacks_(true),
+      read_headers_buffer_(nullptr) {
   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(
-    const SpdyHeaderBlock& headers,
-    size_t frame_len) {
-  delegate_->OnInitialHeadersAvailable(headers, frame_len);
+void QuicChromiumClientStream::Handle::OnInitialHeadersAvailable() {
+  if (!read_headers_callback_ || !can_invoke_callbacks_)
+    return;  // Wait for ReadInitialHeaders to be called.
+
+  int rv;

[xunjieli, 2017/05/10 17:54:27]

Looks like DeliverInitialiHeaders() can fail to initialize |rv|. Should we
initialize |rv|?

[Ryan Hamilton, 2017/05/10 18:26:40]

I think that should only happen when Deliver returns false, in which case we
don't use rv. But better safe than sorry! Done.

+  if (!stream_->DeliverInitialHeaders(read_headers_buffer_, &rv))

[xunjieli, 2017/05/10 17:54:27]

If BidiStream or QuicHttpStream is gone but Handle is not,
|read_headers_buffer_| will be invalid.
Right now everything works because BidiStream/QuicHttpStream owns Handle, so
|read_headers_buffer_| is guaranteed to be alive when Handle is. That's fragile
because it also depends on destruction order.

We could make Handle own the SpdyHeaderBlock, though that would result in having
an extra accessor. 

[Ryan Hamilton, 2017/05/10 18:26:40]

The Handle is by definition owned by "the caller" (bidi/http stream). They must
outlive the handle. This is a common net async idiom, isn't it? If you call an
async method on some object which has not yet completed by the time you are
destroyed, you need to do one of two things:

1) delete that object (if you own it)
2) cancel the async operation

In this case, because ownership is clear, we're doing #1.
I don't understand how that would work. Would the caller not pass in a
SpdyHeaderBlock*? That's not the common net async read pattern.

[xunjieli, 2017/05/10 18:58:08]

Acknowledged. The ownership model takes a while to get used to. But I like the
lifetime guarantee that now we know we are going to enforce it :)

[Ryan Hamilton, 2017/05/10 19:33:46]

*fingers crossed*! 

+    rv = ERR_QUIC_PROTOCOL_ERROR;
+
+  ResetAndReturn(&read_headers_callback_).Run(rv);
 }
 
 void QuicChromiumClientStream::Handle::OnTrailingHeadersAvailable(
     const SpdyHeaderBlock& headers,
     size_t frame_len) {
   delegate_->OnTrailingHeadersAvailable(headers, frame_len);
 }
 
 void QuicChromiumClientStream::Handle::OnDataAvailable() {
   delegate_->OnDataAvailable();
@@ skipping 14 matching lines @@
   if (stream_)
     SaveState();
   stream_ = nullptr;
   if (delegate_) {
     auto* delegate = delegate_;
     delegate_ = nullptr;
     delegate->OnError(error);
   }
 }
 
+int QuicChromiumClientStream::Handle::ReadInitialHeaders(
+    SpdyHeaderBlock* header_block,
+    const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&can_invoke_callbacks_, false);

[xunjieli, 2017/05/10 17:54:27]

Why do we need to have this bool? I don't see any re-entrancy in this method.

[Ryan Hamilton, 2017/05/10 18:26:40]

AH! Sorry, I meant to mention this in the CL description but forgot to. So this
CL is one of 5 (I think). We need to add async support to:

1) Read Initial headers
2) Read body
3) Read trailers (I think this can always be synchronous)
4) Send headers (I think this can always be synchronous)
5) Send body (This is already async, but in the wrong place)

Once we've done all of this, we can remove OnClose/OnError. In order to remove
the OnError/OnClose up-call, the handle needs to know when it's ok to call the
callbacks, and when it's not. This is not required for *this* CL, but I'm
jumping through these hoops now in preparation for the later changes. (But I can
skip them now if you prefer). For example, there might be a situation where a
ReadBody() async call is pending when a WriteBody() call causes a write error,
which shows up as Handle::OnError being called from the stream. Handle::OnError
will notice that read_body_callback_ is set, and might be inclined to do
read_body_callback_.Run(error), but that would be wrong. The right thing to do
would be to return error from WriteBody(). So this "can_invoke_callbacks_"
member is the trigger to avoid invoking the callback.

But I can totally nuke this if for now if you prefer, since it's speculative at
this point. Your call.

[xunjieli, 2017/05/10 18:58:08]

I see. That makes sense. Let's omit this in this CL then. I would like to
re-evaluate this part when we run into the situation that you described and make
sure we have tests covering the sync-failure code paths. Mostly is that I wanted
to make sure I understand how the pieces fit together :)
Thanks! 

[Ryan Hamilton, 2017/05/10 19:33:46]

SGTM. Done.

+  if (!stream_)
+    return ERR_CONNECTION_CLOSED;
+
+  int frame_len;

[xunjieli, 2017/05/10 17:54:27]

same here. |frame_len| might not be initialized.

[Ryan Hamilton, 2017/05/10 18:26:40]

Done.

+  if (stream_->DeliverInitialHeaders(header_block, &frame_len))
+    return frame_len;
+
+  read_headers_buffer_ = header_block;
+  SetCallback(callback, &read_headers_callback_);
+  return ERR_IO_PENDING;
+}
+
 size_t 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);
 }
@@ skipping 136 matching lines @@
   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) {
+  DCHECK(!can_invoke_callbacks_);
+  *callback = new_callback;
+}
+
 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 17 matching lines @@
   if (!SpdyUtils::CopyAndValidateHeaders(header_list, &length, &header_block)) {
     DLOG(ERROR) << "Failed to parse header list: " << header_list.DebugString();
     ConsumeHeaderList();
     Reset(QUIC_BAD_APPLICATION_PAYLOAD);
     return;
   }
 
   ConsumeHeaderList();
   session_->OnInitialHeadersComplete(id(), header_block);
 
-  if (handle_) {
-    // The handle will receive the headers via a posted task.
-    NotifyHandleOfInitialHeadersAvailableLater(std::move(header_block),
-                                               frame_len);
-    return;
-  }
-
   // Buffer the headers and deliver them when the handle arrives.
   initial_headers_ = std::move(header_block);
   initial_headers_frame_len_ = frame_len;
+
+  if (handle_) {
+    // The handle will be notified of the headers via a posted task.
+    NotifyHandleOfInitialHeadersAvailableLater();
+  }
 }
 
 void QuicChromiumClientStream::OnTrailingHeadersComplete(
     bool fin,
     size_t frame_len,
     const QuicHeaderList& header_list) {
   QuicSpdyStream::OnTrailingHeadersComplete(fin, frame_len, header_list);
   NotifyHandleOfTrailingHeadersAvailableLater(received_trailers().Clone(),
                                               frame_len);
 }
@@ skipping 113 matching lines @@
 
 std::unique_ptr<QuicChromiumClientStream::Handle>
 QuicChromiumClientStream::CreateHandle(
     QuicChromiumClientStream::Delegate* delegate) {
   DCHECK(!handle_);
   auto handle = std::unique_ptr<QuicChromiumClientStream::Handle>(
       new QuicChromiumClientStream::Handle(this, delegate));
   handle_ = handle.get();
 
   // Should this perhaps be via PostTask to make reasoning simpler?
-  if (!initial_headers_.empty()) {
-    handle_->OnInitialHeadersAvailable(std::move(initial_headers_),
-                                       initial_headers_frame_len_);
-  }
+  if (!initial_headers_.empty())
+    handle_->OnInitialHeadersAvailable();
 
   return handle;
 }
 
 void QuicChromiumClientStream::ClearHandle() {
   handle_ = nullptr;
 }
 
 void QuicChromiumClientStream::OnError(int error) {
   if (handle_) {
@@ skipping 12 matching lines @@
 
   iovec iov;
   iov.iov_base = buf->data();
   iov.iov_len = buf_len;
   size_t bytes_read = Readv(&iov, 1);
   // Since HasBytesToRead is true, Readv() must of read some data.
   DCHECK_NE(0u, bytes_read);
   return bytes_read;
 }
 
-void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailableLater(
-    SpdyHeaderBlock headers,
-    size_t frame_len) {
+void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailableLater() {
   DCHECK(handle_);
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE,
       base::Bind(
           &QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailable,
-          weak_factory_.GetWeakPtr(), base::Passed(std::move(headers)),
-          frame_len));
+          weak_factory_.GetWeakPtr()));
 }
 
-void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailable(
-    SpdyHeaderBlock headers,
-    size_t frame_len) {
+void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailable() {
   if (!handle_)
     return;
 
-  DCHECK(!headers_delivered_);
-  headers_delivered_ = true;
-  net_log_.AddEvent(
-      NetLogEventType::QUIC_CHROMIUM_CLIENT_STREAM_READ_RESPONSE_HEADERS,
-      base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
-
-  handle_->OnInitialHeadersAvailable(headers, frame_len);
+  if (!headers_delivered_)
+    handle_->OnInitialHeadersAvailable();
 }
 
 void QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailableLater(
     SpdyHeaderBlock headers,
     size_t frame_len) {
   DCHECK(handle_);
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE,
       base::Bind(
           &QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable,
           weak_factory_.GetWeakPtr(), base::Passed(std::move(headers)),
           frame_len));
 }
 
 void QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable(
     SpdyHeaderBlock headers,
     size_t frame_len) {
   if (!handle_)
     return;
 
   DCHECK(headers_delivered_);
   // Only mark trailers consumed when we are about to notify delegate.
   MarkTrailersConsumed();
   // Post an async task to notify delegate of the FIN flag.
   NotifyHandleOfDataAvailableLater();
   net_log_.AddEvent(
       NetLogEventType::QUIC_CHROMIUM_CLIENT_STREAM_READ_RESPONSE_TRAILERS,
       base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
+  handle_->OnTrailingHeadersAvailable(headers, frame_len);
+}
 
-  handle_->OnTrailingHeadersAvailable(headers, frame_len);
+bool QuicChromiumClientStream::DeliverInitialHeaders(SpdyHeaderBlock* headers,
+                                                     int* frame_len) {
+  if (initial_headers_.empty())
+    return false;
+
+  headers_delivered_ = true;
+  net_log_.AddEvent(
+      NetLogEventType::QUIC_CHROMIUM_CLIENT_STREAM_READ_RESPONSE_HEADERS,
+      base::Bind(&SpdyHeaderBlockNetLogCallback, &initial_headers_));
+
+  *headers = std::move(initial_headers_);
+  *frame_len = initial_headers_frame_len_;
+  return true;
 }
 
 void QuicChromiumClientStream::NotifyHandleOfDataAvailableLater() {
   DCHECK(handle_);
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE,
       base::Bind(&QuicChromiumClientStream::NotifyHandleOfDataAvailable,
                  weak_factory_.GetWeakPtr()));
 }
 
 void QuicChromiumClientStream::NotifyHandleOfDataAvailable() {
   if (handle_)
     handle_->OnDataAvailable();
 }
 
 void QuicChromiumClientStream::DisableConnectionMigration() {
   can_migrate_ = false;
 }
 
 bool QuicChromiumClientStream::IsFirstStream() {
   return id() == kHeadersStreamId + 2;
 }
 
 }  // namespace net