Use SharedMemoryDataConsumerHandle for stream data.

content/child/web_url_loader_impl.cc

 // Copyright 2014 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 "content/child/web_url_loader_impl.h"
 
 #include <algorithm>
-#include <deque>
 #include <string>
 
 #include "base/bind.h"
 #include "base/command_line.h"
 #include "base/files/file_path.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_util.h"
 #include "base/time/time.h"
 #include "components/mime_util/mime_util.h"
 #include "content/child/child_thread_impl.h"
 #include "content/child/ftp_directory_listing_response_delegate.h"
 #include "content/child/multipart_response_delegate.h"
 #include "content/child/request_extra_data.h"
 #include "content/child/request_info.h"
 #include "content/child/resource_dispatcher.h"
+#include "content/child/shared_memory_data_consumer_handle.h"
 #include "content/child/sync_load_response.h"
-#include "content/child/web_data_consumer_handle_impl.h"
 #include "content/child/web_url_request_util.h"
 #include "content/child/weburlresponse_extradata_impl.h"
 #include "content/common/resource_messages.h"
 #include "content/common/resource_request_body.h"
 #include "content/common/service_worker/service_worker_types.h"
 #include "content/public/child/fixed_received_data.h"
 #include "content/public/child/request_peer.h"
 #include "content/public/common/content_switches.h"
 #include "net/base/data_url.h"
 #include "net/base/filename_util.h"
 #include "net/base/net_errors.h"
 #include "net/http/http_response_headers.h"
 #include "net/http/http_util.h"
 #include "net/url_request/redirect_info.h"
 #include "net/url_request/url_request_data_job.h"
 #include "third_party/WebKit/public/platform/WebHTTPLoadInfo.h"
 #include "third_party/WebKit/public/platform/WebURL.h"
 #include "third_party/WebKit/public/platform/WebURLError.h"
 #include "third_party/WebKit/public/platform/WebURLLoadTiming.h"
 #include "third_party/WebKit/public/platform/WebURLLoaderClient.h"
 #include "third_party/WebKit/public/platform/WebURLRequest.h"
 #include "third_party/WebKit/public/platform/WebURLResponse.h"
 #include "third_party/WebKit/public/web/WebSecurityPolicy.h"
-#include "third_party/mojo/src/mojo/public/cpp/system/data_pipe.h"
 
 using base::Time;
 using base::TimeTicks;
 using blink::WebData;
 using blink::WebHTTPBody;
 using blink::WebHTTPHeaderVisitor;
 using blink::WebHTTPLoadInfo;
 using blink::WebReferrerPolicy;
 using blink::WebSecurityPolicy;
 using blink::WebString;
 using blink::WebURL;
 using blink::WebURLError;
 using blink::WebURLLoadTiming;
 using blink::WebURLLoader;
 using blink::WebURLLoaderClient;
 using blink::WebURLRequest;
 using blink::WebURLResponse;
 
 namespace content {
 
 // Utilities ------------------------------------------------------------------
 
 namespace {
 
-const size_t kBodyStreamPipeCapacity = 4 * 1024;
-
 using HeadersVector = ResourceDevToolsInfo::HeadersVector;
 
 // Converts timing data from |load_timing| to the format used by WebKit.
 void PopulateURLLoadTiming(const net::LoadTimingInfo& load_timing,
                            WebURLLoadTiming* url_timing) {
   DCHECK(!load_timing.request_start.is_null());
 
   const TimeTicks kNullTicks;
   url_timing->initialize();
   url_timing->setRequestTime(
@@ skipping 236 matching lines @@
                           const base::TimeTicks& completion_time,
                           int64 total_transfer_size) override;
 
  private:
   friend class base::RefCounted<Context>;
   ~Context() override;
 
   // We can optimize the handling of data URLs in most cases.
   bool CanHandleDataURLRequestLocally() const;
   void HandleDataURL();
-  MojoResult WriteDataOnBodyStream(const char* data, size_t size);
-  void OnHandleGotWritable(MojoResult);
 
   WebURLLoaderImpl* loader_;
   WebURLRequest request_;
   WebURLLoaderClient* client_;
   ResourceDispatcher* resource_dispatcher_;
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   WebReferrerPolicy referrer_policy_;
   scoped_ptr<FtpDirectoryListingResponseDelegate> ftp_listing_delegate_;
   scoped_ptr<MultipartResponseDelegate> multipart_delegate_;
   scoped_ptr<StreamOverrideParameters> stream_override_;
-  mojo::ScopedDataPipeProducerHandle body_stream_writer_;
-  mojo::common::HandleWatcher body_stream_writer_watcher_;
-  // TODO(yhirano): Delete this buffer after implementing the back-pressure
-  // mechanism.
-  std::deque<char> body_stream_buffer_;
-  bool got_all_stream_body_data_;
+  scoped_ptr<SharedMemoryDataConsumerHandle::Writer> body_stream_writer_;
   enum DeferState {NOT_DEFERRING, SHOULD_DEFER, DEFERRED_DATA};
   DeferState defers_loading_;
   int request_id_;
 };
 
 WebURLLoaderImpl::Context::Context(
     WebURLLoaderImpl* loader,
     ResourceDispatcher* resource_dispatcher,
     scoped_refptr<base::SingleThreadTaskRunner> task_runner)
     : loader_(loader),
       client_(NULL),
       resource_dispatcher_(resource_dispatcher),
       task_runner_(task_runner),
       referrer_policy_(blink::WebReferrerPolicyDefault),
-      got_all_stream_body_data_(false),
       defers_loading_(NOT_DEFERRING),
       request_id_(-1) {
 }
 
 void WebURLLoaderImpl::Context::Cancel() {
   if (resource_dispatcher_ && // NULL in unittest.
       request_id_ != -1) {
     resource_dispatcher_->Cancel(request_id_);
     request_id_ = -1;
   }
@@ skipping 236 matching lines @@
       response.setMIMEType("text/html");
     }
   }
 
   // Prevent |this| from being destroyed if the client destroys the loader,
   // ether in didReceiveResponse, or when the multipart/ftp delegate calls into
   // it.
   scoped_refptr<Context> protect(this);
 
   if (request_.useStreamOnResponse()) {
-    MojoCreateDataPipeOptions options;
-    options.struct_size = sizeof(MojoCreateDataPipeOptions);
-    options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
-    options.element_num_bytes = 1;
-    options.capacity_num_bytes = kBodyStreamPipeCapacity;
+    SharedMemoryDataConsumerHandle::BackpressureMode mode =
+        SharedMemoryDataConsumerHandle::kDoNotApplyBackpressure;
+    if (info.headers &&
+        info.headers->HasHeaderValue("Cache-Control", "no-store")) {
+      mode = SharedMemoryDataConsumerHandle::kApplyBackpressure;
+    }
 
-    mojo::ScopedDataPipeConsumerHandle consumer;
-    MojoResult result = mojo::CreateDataPipe(&options,
-                                             &body_stream_writer_,
-                                             &consumer);
-    if (result != MOJO_RESULT_OK) {
-      // TODO(yhirano): Handle the error.
-      return;
-    }
-    client_->didReceiveResponse(
-        loader_, response, new WebDataConsumerHandleImpl(consumer.Pass()));
+    auto reader = make_scoped_ptr(
+        new SharedMemoryDataConsumerHandle(mode, &body_stream_writer_));
+
+    // The client takes |reader|'s ownership.
+    client_->didReceiveResponse(loader_, response, reader.release());

[tyoshino (SeeGerritForStatus), 2015/06/10 13:45:55]

how about skipping the rest of this function until as we don't support
ftp_listing_delegate_, etc. for now?

[yhirano, 2015/06/11 02:17:00]

Done.

   } else {
     client_->didReceiveResponse(loader_, response);
   }
 
   // We may have been cancelled after didReceiveResponse, which would leave us
   // without a client and therefore without much need to do further handling.
   if (!client_)
     return;
 
   DCHECK(!ftp_listing_delegate_.get());
@@ skipping 29 matching lines @@
     client_->didDownloadData(loader_, len, encoded_data_length);
 }
 
 void WebURLLoaderImpl::Context::OnReceivedData(scoped_ptr<ReceivedData> data) {
   const char* payload = data->payload();
   int data_length = data->length();
   int encoded_data_length = data->encoded_length();
   if (!client_)
     return;
 
-  if (request_.useStreamOnResponse()) {
-    // We don't support ftp_listening_delegate_ and multipart_delegate_ for now.
-    // TODO(yhirano): Support ftp listening and multipart.
-    MojoResult rv = WriteDataOnBodyStream(payload, data_length);
-    if (rv != MOJO_RESULT_OK && client_) {
-      client_->didFail(
-          loader_, CreateWebURLError(request_.url(), false, net::ERR_FAILED));
-    }
-  } else if (ftp_listing_delegate_) {
+  if (ftp_listing_delegate_) {
     // The FTP listing delegate will make the appropriate calls to
     // client_->didReceiveData and client_->didReceiveResponse.  Since the
     // delegate may want to do work after sending data to the delegate, keep
     // |this| and the delegate alive until it's finished handling the data.
     scoped_refptr<Context> protect(this);
     ftp_listing_delegate_->OnReceivedData(payload, data_length);
   } else if (multipart_delegate_) {
     // The multipart delegate will make the appropriate calls to
     // client_->didReceiveData and client_->didReceiveResponse.  Since the
     // delegate may want to do work after sending data to the delegate, keep
     // |this| and the delegate alive until it's finished handling the data.
     scoped_refptr<Context> protect(this);
     multipart_delegate_->OnReceivedData(payload, data_length,
                                         encoded_data_length);
   } else {
+    scoped_refptr<Context> protect(this);
+    // We dispatch the data even when |useStreamOnResponse()| is set, in order
+    // to make Devtools work.
     client_->didReceiveData(loader_, payload, data_length, encoded_data_length);
+
+    if (request_.useStreamOnResponse()) {
+      // We don't support ftp_listening_delegate_ and multipart_delegate_ for
+      // now.
+      // TODO(yhirano): Support ftp listening and multipart.
+      body_stream_writer_->AddData(data.Pass());
+    }
   }
 }
 
 void WebURLLoaderImpl::Context::OnReceivedCachedMetadata(
     const char* data, int len) {
   if (client_)
     client_->didReceiveCachedMetadata(loader_, data, len);
 }
 
 void WebURLLoaderImpl::Context::OnCompletedRequest(
@@ skipping 12 matching lines @@
   if (ftp_listing_delegate_) {
     ftp_listing_delegate_->OnCompletedRequest();
     ftp_listing_delegate_.reset(NULL);
   } else if (multipart_delegate_) {
     multipart_delegate_->OnCompletedRequest();
     multipart_delegate_.reset(NULL);
   }
 
   if (client_) {
     if (error_code != net::OK) {
       client_->didFail(

[hiroshige, 2015/06/11 00:58:00]

We have to pass failures to |body_stream_writer_|/Stream if
request_.useStreamOnResponse().
(This seems the case in the 'Error in text() should be propagated to the body
stream.' test in stream-reader.js (if my ongoing CLs
https://codereview.chromium.org/1171913003/ (PS14) are applied); didFail() is
called but no failure is observed via WebDataConsumerHandle::Reader?)

[yhirano, 2015/06/11 02:17:00]

I have another WIP CL addressing this issue:
https://codereview.chromium.org/1181573003/
Let me land this CL to unblock other CLs.

           loader_,
           CreateWebURLError(request_.url(), stale_copy_in_cache, error_code));
     } else {
-      if (request_.useStreamOnResponse()) {
-        got_all_stream_body_data_ = true;
-        if (body_stream_buffer_.empty()) {
-          // Close the handle to notify the end of data.
-          body_stream_writer_.reset();
-          client_->didFinishLoading(
-              loader_, (completion_time - TimeTicks()).InSecondsF(),
-              total_transfer_size);
-        }
-      } else {
-        client_->didFinishLoading(
-            loader_, (completion_time - TimeTicks()).InSecondsF(),
-            total_transfer_size);
-      }
+      if (request_.useStreamOnResponse())
+        body_stream_writer_.reset();
+
+      client_->didFinishLoading(loader_,

[hiroshige, 2015/06/11 00:01:58]

Here |client_| can be null and |if(client_)| is needed?

body_stream_writer_.reset();
-> Writer is closed
-> didGetReadable() is called
-> Done is returned for beginRead()
-> Client call corresponding ThreadbleLoader::cancel()
-> WebURLLoaderImpl::cancel()
-> WebURLLoaderImpl::Context::Cancel()
-> client_ is cleared.

Stack trace:
content::WebURLLoaderImpl::Context::Cancel()
content::WebURLLoaderImpl::cancel()
blink::ResourceLoader::cancel()
blink::ResourceLoader::cancel()
blink::ResourceLoader::cancelIfNotFinishing()
blink::Resource::cancelTimerFired()
blink::Resource::allClientsRemoved()
blink::Resource::removeClient()
blink::ResourceOwner<>::setResource()
blink::ResourceOwner<>::clearResource()
blink::DocumentThreadableLoader::cancelWithError()
blink::DocumentThreadableLoader::cancel()
blink::FetchManager::Loader::cancel()
blink::FetchManager::Loader::Canceller::cancel()
blink::(anonymous namespace)::WebDataConsumerHandleAdapter::close()
blink::(anonymous namespace)::WebDataConsumerHandleAdapter::didGetReadable()
content::SharedMemoryDataConsumerHandle::Context::NotifyInternal()
content::SharedMemoryDataConsumerHandle::Context::Notify()
content::SharedMemoryDataConsumerHandle::Writer::Close()
content::SharedMemoryDataConsumerHandle::Writer::~Writer()
base::DefaultDeleter<>::operator()()
base::internal::scoped_ptr_impl<>::reset()
scoped_ptr<>::reset()
content::WebURLLoaderImpl::Context::OnCompletedRequest()

[yhirano, 2015/06/11 02:17:00]

Done.

+                                (completion_time - TimeTicks()).InSecondsF(),
+                                total_transfer_size);
     }
   }
 }
 
 WebURLLoaderImpl::Context::~Context() {
   if (request_id_ >= 0) {
     resource_dispatcher_->RemovePendingRequest(request_id_);
   }
 }
 
@@ skipping 54 matching lines @@
     OnReceivedResponse(info);
     if (!data.empty())
       OnReceivedData(
           make_scoped_ptr(new FixedReceivedData(data.data(), data.size(), 0)));
   }
 
   OnCompletedRequest(error_code, false, false, info.security_info,
                      base::TimeTicks::Now(), 0);
 }
 
-MojoResult WebURLLoaderImpl::Context::WriteDataOnBodyStream(const char* data,
-                                                            size_t size) {
-  if (body_stream_buffer_.empty() && size == 0) {
-    // Nothing to do.
-    return MOJO_RESULT_OK;
-  }
-
-  if (!body_stream_writer_.is_valid()) {
-    // The handle is already cleared.
-    return MOJO_RESULT_OK;
-  }
-
-  char* buffer = nullptr;
-  uint32_t num_bytes_writable = 0;
-  MojoResult rv = mojo::BeginWriteDataRaw(body_stream_writer_.get(),
-                                          reinterpret_cast<void**>(&buffer),
-                                          &num_bytes_writable,
-                                          MOJO_WRITE_DATA_FLAG_NONE);
-  if (rv == MOJO_RESULT_SHOULD_WAIT) {
-    body_stream_buffer_.insert(body_stream_buffer_.end(), data, data + size);
-    body_stream_writer_watcher_.Start(
-        body_stream_writer_.get(),
-        MOJO_HANDLE_SIGNAL_WRITABLE,
-        MOJO_DEADLINE_INDEFINITE,
-        base::Bind(&WebURLLoaderImpl::Context::OnHandleGotWritable,
-                   base::Unretained(this)));
-    return MOJO_RESULT_OK;
-  }
-
-  if (rv != MOJO_RESULT_OK)
-    return rv;
-
-  uint32_t num_bytes_to_write = 0;
-  if (num_bytes_writable < body_stream_buffer_.size()) {
-    auto begin = body_stream_buffer_.begin();
-    auto end = body_stream_buffer_.begin() + num_bytes_writable;
-
-    std::copy(begin, end, buffer);
-    num_bytes_to_write = num_bytes_writable;
-    body_stream_buffer_.erase(begin, end);
-    body_stream_buffer_.insert(body_stream_buffer_.end(), data, data + size);
-  } else {
-    std::copy(body_stream_buffer_.begin(), body_stream_buffer_.end(), buffer);
-    num_bytes_writable -= body_stream_buffer_.size();
-    num_bytes_to_write += body_stream_buffer_.size();
-    buffer += body_stream_buffer_.size();
-    body_stream_buffer_.clear();
-
-    size_t num_newbytes_to_write =
-        std::min(size, static_cast<size_t>(num_bytes_writable));
-    std::copy(data, data + num_newbytes_to_write, buffer);
-    num_bytes_to_write += num_newbytes_to_write;
-    body_stream_buffer_.insert(body_stream_buffer_.end(),
-                               data + num_newbytes_to_write,
-                               data + size);
-  }
-
-  rv = mojo::EndWriteDataRaw(body_stream_writer_.get(), num_bytes_to_write);
-  if (rv == MOJO_RESULT_OK && !body_stream_buffer_.empty()) {
-    body_stream_writer_watcher_.Start(
-        body_stream_writer_.get(),
-        MOJO_HANDLE_SIGNAL_WRITABLE,
-        MOJO_DEADLINE_INDEFINITE,
-        base::Bind(&WebURLLoaderImpl::Context::OnHandleGotWritable,
-                   base::Unretained(this)));
-  }
-  return rv;
-}
-
-void WebURLLoaderImpl::Context::OnHandleGotWritable(MojoResult result) {
-  if (result != MOJO_RESULT_OK) {
-    if (client_) {
-      client_->didFail(
-          loader_, CreateWebURLError(request_.url(), false, net::ERR_FAILED));
-      // |this| can be deleted here.
-    }
-    return;
-  }
-
-  if (body_stream_buffer_.empty())
-    return;
-
-  MojoResult rv = WriteDataOnBodyStream(nullptr, 0);
-  if (rv == MOJO_RESULT_OK) {
-    if (got_all_stream_body_data_ && body_stream_buffer_.empty()) {
-      // Close the handle to notify the end of data.
-      body_stream_writer_.reset();
-      if (client_) {
-        // TODO(yhirano): Pass appropriate arguments.
-        client_->didFinishLoading(loader_, 0, 0);
-        // |this| can be deleted here.
-      }
-    }
-  } else {
-    if (client_) {
-      client_->didFail(
-          loader_, CreateWebURLError(request_.url(), false, net::ERR_FAILED));
-      // |this| can be deleted here.
-    }
-  }
-}
-
 // WebURLLoaderImpl -----------------------------------------------------------
 
 WebURLLoaderImpl::WebURLLoaderImpl(
     ResourceDispatcher* resource_dispatcher,
     scoped_refptr<base::SingleThreadTaskRunner> task_runner)
     : context_(new Context(this, resource_dispatcher, task_runner)) {
 }
 
 WebURLLoaderImpl::~WebURLLoaderImpl() {
   cancel();
@@ skipping 167 matching lines @@
                                          int intra_priority_value) {
   context_->DidChangePriority(new_priority, intra_priority_value);
 }
 
 bool WebURLLoaderImpl::attachThreadedDataReceiver(
     blink::WebThreadedDataReceiver* threaded_data_receiver) {
   return context_->AttachThreadedDataReceiver(threaded_data_receiver);
 }
 
 }  // namespace content