[android_webview] Don't block the IO thread when reading from an InputStream.

android_webview/native/android_stream_reader_url_request_job.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 "android_webview/native/android_stream_reader_url_request_job.h"
 
+#include "android_webview/native/input_stream.h"
+#include "android_webview/native/input_stream_reader.h"

[joth, 2012/11/07 16:55:38]

hmm interesting... if input_stream_reader could be moved to aw/browser then
android_stream_reader_url_request_job could too. i.e. only input_stream need
JNI.

[mkosiba (inactive), 2012/11/15 19:18:50]

yup, it's just me being lazy. I'll do the move.

 #include "base/android/jni_android.h"
 #include "base/android/jni_string.h"
 #include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/lazy_instance.h"
 #include "base/message_loop.h"
+#include "base/threading/thread.h"
+#include "content/public/browser/browser_thread.h"
 #include "net/base/io_buffer.h"
 #include "net/base/mime_util.h"
 #include "net/base/net_errors.h"
 #include "net/base/net_util.h"
 #include "net/http/http_util.h"
 #include "net/url_request/url_request.h"
-#include "net/url_request/url_request_error_job.h"
-#include "net/url_request/url_request_file_job.h"
 #include "net/url_request/url_request_job_manager.h"
-// Disable "Warnings treated as errors" for input_stream_jni as it's a Java
-// system class and we have to generate C++ hooks for all methods in the class
-// even if they're unused.
-#pragma GCC diagnostic ignored "-Wunused-function"
-#include "jni/InputStream_jni.h"
 
+using android_webview::InputStream;
+using android_webview::InputStreamReader;
 using base::android::AttachCurrentThread;
-using base::android::ClearException;
-using base::android::ConvertUTF8ToJavaString;
-using base::android::ScopedJavaGlobalRef;
-using base::android::ScopedJavaLocalRef;
-using JNI_InputStream::Java_InputStream_available;
-using JNI_InputStream::Java_InputStream_skip;
-using JNI_InputStream::Java_InputStream_readI_AB_I_I;
-
+using content::BrowserThread;
 

[joth, 2012/11/07 16:55:38]

would be helpful to have a block comment explaining how the  thread bouncing is
handled in this file and its close collaborators (but read my other comments
first).

[mkosiba (inactive), 2012/11/15 19:18:50]

simplified this significantly - you still think one is needed?

 namespace {
 
-// Maximum number of bytes to be read in a single read.
-const int kBufferSize = 4096;
+class WorkerThread : public base::Thread {
+ public:
+  WorkerThread() : base::Thread("AndroidStreamReaderWorkerThread") {

[joth, 2012/11/07 16:55:38]

nit: Worker seems spurious: AndroidStreamReaderThread is pretty clear it's the
thread that we read from streams on.

[mkosiba (inactive), 2012/11/15 19:18:50]

Done.

+  }
+};
+
+static base::LazyInstance<WorkerThread> g_worker_thread =
+    LAZY_INSTANCE_INITIALIZER;

[joth, 2012/11/07 16:55:38]

I'm not thrilled with another thread hanging out as a global here... I guess on
the plus side we'll only create it if the app actually does override the
shouldInterceptRequest method?

I presume you can't use the BrowserThread::PostBlockingPoolTaskAndReply for
these? It will allow thread reuse, and may also aid readability (and help keep
the InputStreamReader a bit less hard-wired into a given threading modal)

[mkosiba (inactive), 2012/11/15 19:18:50]

I didn't know about the PostBlockingPoolTask method. There is a sequenced
version available too but I'm starting to think we should try a parallel
approach first and scale down to single-thread if we hit issues.

 
 } // namespace
 
-bool RegisterAndroidStreamReaderUrlRequestJob(JNIEnv* env) {
-  return JNI_InputStream::RegisterNativesImpl(env);
-}
-
 AndroidStreamReaderURLRequestJob::AndroidStreamReaderURLRequestJob(
     net::URLRequest* request,
     net::NetworkDelegate* network_delegate,
     scoped_ptr<Delegate> delegate)
     : URLRequestJob(request, network_delegate),
       delegate_(delegate.Pass()),
+      input_stream_reader_(NULL),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
   DCHECK(delegate_.get());
 }
 
 AndroidStreamReaderURLRequestJob::~AndroidStreamReaderURLRequestJob() {
+  if (input_stream_reader_ != NULL) {
+    PostToWorkerThread(
+        FROM_HERE,
+        base::Bind(&base::DeletePointer<InputStreamReader>,
+                   input_stream_reader_));
+  }
 }
 
 void AndroidStreamReaderURLRequestJob::Start() {
   // Start reading asynchronously so that all error reporting and data
   // callbacks happen as they would for network requests.
+  SetStatus(net::URLRequestStatus(net::URLRequestStatus::IO_PENDING, 0));
   MessageLoop::current()->PostTask(
       FROM_HERE,
-      base::Bind(&AndroidStreamReaderURLRequestJob::StartAsync,
-                 weak_factory_.GetWeakPtr()));
+      base::Bind(
+          &AndroidStreamReaderURLRequestJob::StartAsync,
+          weak_factory_.GetWeakPtr()));
+}
+
+void AndroidStreamReaderURLRequestJob::Kill() {
+  weak_factory_.InvalidateWeakPtrs();
+  URLRequestJob::Kill();
+}
+
+InputStreamReader* AndroidStreamReaderURLRequestJob::CreateStreamReader(
+    InputStream* stream) {
+  return new InputStreamReader(stream);
 }
 
 void AndroidStreamReaderURLRequestJob::StartAsync() {
   JNIEnv* env = AttachCurrentThread();
   DCHECK(env);
 
-  stream_.Reset(env, delegate_->OpenInputStream(env, request()).obj());
-  if (!stream_.obj()) {
-    NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                     net::ERR_FAILED));
+  // TODO: This could be done in the InputStreamReader but would force more

[joth, 2012/11/07 16:55:38]

is this really a TODO then? reads like a conscious trade off decision (which is
fine)

[mkosiba (inactive), 2012/11/15 19:18:50]

ok, removed TODO.

+  // complex synchronization in the delegate.
+  stream_ = delegate_->OpenInputStream(env, request());
+  if (!stream_) {
+    NotifyDone(
+        net::URLRequestStatus(net::URLRequestStatus::FAILED, net::ERR_FAILED));
     return;
   }
 
-  if (VerifyRequestedRange(env) && SkipToRequestedRange(env))
-    NotifyHeadersComplete();
+  DCHECK(!input_stream_reader_);
+  input_stream_reader_ = CreateStreamReader(stream_.get());
+  CHECK(input_stream_reader_);

[joth, 2012/11/07 16:55:38]

nit dcheck

[mkosiba (inactive), 2012/11/15 19:18:50]

I wanted a CHECK since we'll end up crashing anyway, but if we crash while
executing the Closure the callstack will be useless.

+
+  PostToWorkerThread(
+      FROM_HERE,
+      base::Bind(&InputStreamReader::Seek,
+           base::Unretained(input_stream_reader_),
+           byte_range_,
+           base::Bind(
+               &AndroidStreamReaderURLRequestJob::OnReaderSeekCompleted,
+               weak_factory_.GetWeakPtr())));
 }
 
-bool AndroidStreamReaderURLRequestJob::VerifyRequestedRange(JNIEnv* env) {
-  int32_t size = Java_InputStream_available(env, stream_.obj());
-  if (ClearException(env)) {
+void AndroidStreamReaderURLRequestJob::OnReaderSeekCompleted(
+    int result) {
+  // Clear the IO_PENDING status
+  SetStatus(net::URLRequestStatus());
+  if (result >= 0) {
+    set_expected_content_size(result);
+    NotifyHeadersComplete();
+  } else {
+    // The InputStreamReader deletes itself on error.
+    input_stream_reader_ = NULL;
+
     NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                     net::ERR_FAILED));
-    return false;
+                                     result));
+  }
+}
+
+void AndroidStreamReaderURLRequestJob::OnReaderReadCompleted(

[benm (inactive), 2012/11/15 17:39:01]

Please add a comment to explain the differences between NotifyDone and
NotifyReadComplete, especially why we don't call NotifyDone in the case of a >0
result code. I understand it's probably due to something deep in the net stack
so it doesn't need to be much here but right now it looks a bit like something
was forgotten.

Maybe move the ReadComplete to the top of the function?

+    int result) {
+  if (result > 0) {
+    SetStatus(net::URLRequestStatus());  // Clear the IO_PENDING status
+  } else if (result == 0) {
+    NotifyDone(net::URLRequestStatus());
+  } else {
+    NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
+                                     result));
   }
 
-  if (size <= 0)
-    return true;
-
-  // Check that the requested range was valid.
-  if (!byte_range_.ComputeBounds(size)) {
-    NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-               net::ERR_REQUEST_RANGE_NOT_SATISFIABLE));
-    return false;
-  }
-
-  size = byte_range_.last_byte_position() -
-         byte_range_.first_byte_position() + 1;
-  DCHECK_GE(size, 0);
-  set_expected_content_size(size);
-
-  return true;
+  NotifyReadComplete(result);
 }
 
-bool AndroidStreamReaderURLRequestJob::SkipToRequestedRange(JNIEnv* env) {
-  // Skip to the start of the requested data. This has to be done in a loop
-  // because the underlying InputStream is not guaranteed to skip the requested
-  // number of bytes.
-  if (byte_range_.IsValid() && byte_range_.first_byte_position() != 0) {
-    int64_t skipped, bytes_to_skip = byte_range_.first_byte_position();
-    do {
-      skipped = Java_InputStream_skip(env, stream_.obj(), bytes_to_skip);
-      if (ClearException(env)) {
-        NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                         net::ERR_FAILED));
-        return false;
-      }
-      if (skipped <= 0) {
-        NotifyDone(
-            net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                  net::ERR_REQUEST_RANGE_NOT_SATISFIABLE));
-        return false;
-      }
-    } while ((bytes_to_skip -= skipped) > 0);
+void AndroidStreamReaderURLRequestJob::PostToWorkerThread(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task) {
+  if(!g_worker_thread.Get().message_loop()) {
+    bool thread_created = g_worker_thread.Get().StartWithOptions(
+        base::Thread::Options(MessageLoop::TYPE_IO, 0));
+    CHECK(thread_created);

[benm (inactive), 2012/11/15 17:39:01]

DCHECK?

[mkosiba (inactive), 2012/11/15 19:18:50]

removed the code.

   }
-  return true;
+  g_worker_thread.Get().message_loop()->PostTask(from_here, task);
 }
 
 bool AndroidStreamReaderURLRequestJob::ReadRawData(net::IOBuffer* dest,
                                                    int dest_size,
                                                    int *bytes_read) {
-  DCHECK_NE(dest_size, 0);
-  DCHECK(bytes_read);
-  DCHECK(stream_.obj());
+  DCHECK(input_stream_reader_);
 
-  JNIEnv* env = AttachCurrentThread();
-  DCHECK(env);
+  PostToWorkerThread(
+      FROM_HERE,
+      base::Bind(&InputStreamReader::ReadRawData,
+           base::Unretained(input_stream_reader_),
+           base::Unretained(dest),
+           dest_size,
+           base::Bind(
+               &AndroidStreamReaderURLRequestJob::OnReaderReadCompleted,
+               weak_factory_.GetWeakPtr())));
 
-  if (!buffer_.obj()) {
-    // Allocate transfer buffer.
-    buffer_.Reset(env, env->NewByteArray(kBufferSize));
-    if (ClearException(env)) {
-      NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                       net::ERR_FAILED));
-      return false;
-    }
-  }
-
-  jbyteArray buffer = buffer_.obj();
-  *bytes_read = 0;
-  if (!dest_size)
-    return true;
-
-  // Read data in multiples of the buffer size.
-  while (dest_size > 0) {
-    int read_size = std::min(dest_size, kBufferSize);
-    // TODO(skyostil): Make this non-blocking
-    int32_t byte_count =
-        Java_InputStream_readI_AB_I_I(env, stream_.obj(), buffer, 0, read_size);
-    if (byte_count <= 0) {
-      // net::URLRequestJob will call NotifyDone for us after the end of the
-      // file is reached.
-      break;
-    }
-
-    if (ClearException(env)) {
-      NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                       net::ERR_FAILED));
-      return false;
-    }
-
-#ifndef NDEBUG
-    int32_t buffer_length = env->GetArrayLength(buffer);
-    DCHECK_GE(read_size, byte_count);
-    DCHECK_GE(buffer_length, byte_count);
-#endif // NDEBUG
-
-    // Copy the data over to the provided C++ side buffer.
-    DCHECK_GE(dest_size, byte_count);
-    env->GetByteArrayRegion(buffer, 0, byte_count,
-        reinterpret_cast<jbyte*>(dest->data() + *bytes_read));
-
-    if (ClearException(env)) {
-      NotifyDone(net::URLRequestStatus(net::URLRequestStatus::FAILED,
-                                       net::ERR_FAILED));
-      return false;
-    }
-
-    *bytes_read += byte_count;
-    dest_size -= byte_count;
-  }
-  return true;
+  SetStatus(net::URLRequestStatus(net::URLRequestStatus::IO_PENDING, 0));
+  return false;
 }
 
 bool AndroidStreamReaderURLRequestJob::GetMimeType(
     std::string* mime_type) const {
   JNIEnv* env = AttachCurrentThread();
   DCHECK(env);
 
+  if (!stream_)
+    return false;
+
   return delegate_->GetMimeType(env,
                                 request(),
-                                stream_.obj(),
+                                *stream_,
                                 mime_type);
 }
 
 bool AndroidStreamReaderURLRequestJob::GetCharset(
     std::string* charset) {
   JNIEnv* env = AttachCurrentThread();
   DCHECK(env);
 
+  if (!stream_)
+    return false;
+
   return delegate_->GetCharset(env,
                                request(),
-                               stream_.obj(),
+                               *stream_,
                                charset);
 }
 
 void AndroidStreamReaderURLRequestJob::SetExtraRequestHeaders(
     const net::HttpRequestHeaders& headers) {
   std::string range_header;
   if (headers.GetHeader(net::HttpRequestHeaders::kRange, &range_header)) {
     // We only care about "Range" header here.
     std::vector<net::HttpByteRange> ranges;
     if (net::HttpUtil::ParseRangeHeader(range_header, &ranges)) {
       if (ranges.size() == 1) {
         byte_range_ = ranges[0];
       } else {
         // We don't support multiple range requests in one single URL request,
         // because we need to do multipart encoding here.
         NotifyDone(net::URLRequestStatus(
             net::URLRequestStatus::FAILED,
             net::ERR_REQUEST_RANGE_NOT_SATISFIABLE));
       }
     }
   }
 }