Merge pull request #1553 from abarth/io_thread

Introduce IO thread for background tasks

sky/engine/core/loader/CanvasImageDecoder.cpp

@@ -9,6 +9,7 @@
 #include "sky/engine/core/painting/CanvasImage.h"
 #include "sky/engine/platform/SharedBuffer.h"
 #include "sky/engine/platform/image-decoders/ImageDecoder.h"
+#include "sky/engine/platform/mojo/data_pipe.h"
 
 namespace blink {
 
@@ -20,14 +21,12 @@ PassRefPtr<CanvasImageDecoder> CanvasImageDecoder::create(
 CanvasImageDecoder::CanvasImageDecoder(PassOwnPtr<ImageDecoderCallback> callback)
     : callback_(callback), weak_factory_(this) {
   CHECK(callback_);
-  buffer_ = SharedBuffer::create();
 }
 
 CanvasImageDecoder::~CanvasImageDecoder() {
 }
 
 void CanvasImageDecoder::initWithConsumer(mojo::ScopedDataPipeConsumerHandle handle) {
-  CHECK(!drainer_);
   if (!handle.is_valid()) {
     base::MessageLoop::current()->PostTask(
         FROM_HERE, base::Bind(&CanvasImageDecoder::RejectCallback,
@@ -35,26 +34,23 @@ void CanvasImageDecoder::initWithConsumer(mojo::ScopedDataPipeConsumerHandle han
     return;
   }
 
-  drainer_ = adoptPtr(new mojo::common::DataPipeDrainer(this, handle.Pass()));
+  DrainDataPipeInBackground(handle.Pass(),
+      base::Bind(&CanvasImageDecoder::Decode, weak_factory_.GetWeakPtr()));
 }
 
 void CanvasImageDecoder::initWithList(const Uint8List& list) {
-  CHECK(!drainer_);
-
-  OnDataAvailable(list.data(), list.num_elements());
+  RefPtr<SharedBuffer> buffer = SharedBuffer::create(list.num_elements());
+  buffer->append(reinterpret_cast<const char*>(list.data()),
+                 list.num_elements());
   base::MessageLoop::current()->PostTask(
-      FROM_HERE, base::Bind(&CanvasImageDecoder::OnDataComplete,
-                            weak_factory_.GetWeakPtr()));
-}
-
-void CanvasImageDecoder::OnDataAvailable(const void* data, size_t num_bytes) {
-  buffer_->append(static_cast<const char*>(data), num_bytes);
+      FROM_HERE, base::Bind(&CanvasImageDecoder::Decode,
+                            weak_factory_.GetWeakPtr(), buffer.release()));
 }
 
-void CanvasImageDecoder::OnDataComplete() {
-  TRACE_EVENT0("blink", "CanvasImageDecoder::OnDataComplete");
+void CanvasImageDecoder::Decode(PassRefPtr<SharedBuffer> buffer) {
+  TRACE_EVENT0("blink", "CanvasImageDecoder::Decode");
   OwnPtr<ImageDecoder> decoder =
-      ImageDecoder::create(*buffer_.get(), ImageSource::AlphaPremultiplied,
+      ImageDecoder::create(*buffer.get(), ImageSource::AlphaPremultiplied,
                            ImageSource::GammaAndColorProfileIgnored);
   // decoder can be null if the buffer we was empty and we couldn't even guess
   // what type of image to decode.
@@ -62,7 +58,7 @@ void CanvasImageDecoder::OnDataComplete() {
     callback_->handleEvent(nullptr);
     return;
   }
-  decoder->setData(buffer_.get(), true);
+  decoder->setData(buffer.get(), true);
   if (decoder->failed() || decoder->frameCount() == 0) {
     callback_->handleEvent(nullptr);
     return;

sky/engine/core/loader/CanvasImageDecoder.h

@@ -7,7 +7,7 @@
 #define SKY_ENGINE_CORE_LOADER_CANVASIMAGELOADER_H_
 
 #include "base/memory/weak_ptr.h"
-#include "mojo/data_pipe_utils/data_pipe_drainer.h"
+#include "mojo/public/cpp/system/core.h"
 #include "sky/engine/core/loader/ImageDecoderCallback.h"
 #include "sky/engine/platform/SharedBuffer.h"
 #include "sky/engine/tonic/dart_wrappable.h"
@@ -17,28 +17,22 @@
 
 namespace blink {
 
-class CanvasImageDecoder : public mojo::common::DataPipeDrainer::Client,
-                           public RefCounted<CanvasImageDecoder>,
+class CanvasImageDecoder : public RefCounted<CanvasImageDecoder>,
                            public DartWrappable {
   DEFINE_WRAPPERTYPEINFO();
  public:
   static PassRefPtr<CanvasImageDecoder> create(PassOwnPtr<ImageDecoderCallback> callback);
   virtual ~CanvasImageDecoder();
 
-  // mojo::common::DataPipeDrainer::Client
-  void OnDataAvailable(const void*, size_t) override;
-  void OnDataComplete() override;
-
   void initWithConsumer(mojo::ScopedDataPipeConsumerHandle handle);
   void initWithList(const Uint8List& list);
 
  private:
-  CanvasImageDecoder(PassOwnPtr<ImageDecoderCallback> callback);
+  explicit CanvasImageDecoder(PassOwnPtr<ImageDecoderCallback> callback);
 
+  void Decode(PassRefPtr<SharedBuffer> buffer);
   void RejectCallback();
 
-  OwnPtr<mojo::common::DataPipeDrainer> drainer_;
-  RefPtr<SharedBuffer> buffer_;
   OwnPtr<ImageDecoderCallback> callback_;
 
   base::WeakPtrFactory<CanvasImageDecoder> weak_factory_;

sky/engine/core/script/dart_controller.cc

@@ -54,7 +54,7 @@ void CallHandleMessage(base::WeakPtr<DartState> dart_state) {
 
 void MessageNotifyCallback(Dart_Isolate dest_isolate) {
   DCHECK(Platform::current());
-  Platform::current()->mainThreadTaskRunner()->PostTask(FROM_HERE,
+  Platform::current()->GetUITaskRunner()->PostTask(FROM_HERE,
       base::Bind(&CallHandleMessage, DartState::From(dest_isolate)->GetWeakPtr()));
 }
 

sky/engine/platform/BUILD.gn

@@ -427,6 +427,8 @@ source_set("platform") {
     "image-encoders/skia/JPEGImageEncoder.h",
     "image-encoders/skia/PNGImageEncoder.cpp",
     "image-encoders/skia/PNGImageEncoder.h",
+    "mojo/data_pipe.cc",
+    "mojo/data_pipe.h",
     "text/BidiCharacterRun.cpp",
     "text/BidiCharacterRun.h",
     "text/BidiContext.cpp",
@@ -542,6 +544,7 @@ source_set("platform") {
     ":make_platform_generated",
     "//base:base",
     "//mojo/application",
+    "//mojo/data_pipe_utils",
     "//mojo/environment:chromium",
     "//mojo/public/c/system",
     "//mojo/public/cpp/bindings",

sky/engine/platform/SharedBuffer.cpp

@@ -126,7 +126,7 @@ static void didCreateSharedBuffer(SharedBuffer* buffer)
     MutexLocker locker(statsMutex());
     liveBuffers().add(buffer);
 
-    Platform::current()->mainThreadTaskRunner()->PostTask(FROM_HERE, base::Bind(&printStats));
+    Platform::current()->GetUITaskRunner()->PostTask(FROM_HERE, base::Bind(&printStats));
 }
 
 static void willDestroySharedBuffer(SharedBuffer* buffer)

sky/engine/platform/mojo/data_pipe.cc

+// Copyright 2013 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 "sky/engine/platform/mojo/data_pipe.h"
+
+#include <memory>
+
+#include "base/bind.h"
+#include "mojo/data_pipe_utils/data_pipe_drainer.h"
+#include "sky/engine/platform/mojo/data_pipe.h"
+#include "sky/engine/public/platform/Platform.h"
+#include "sky/engine/wtf/RefPtr.h"
+
+namespace blink {
+namespace {
+
+class DrainJob : public mojo::common::DataPipeDrainer::Client {
+ public:
+  explicit DrainJob(base::Callback<void(PassRefPtr<SharedBuffer>)> callback)
+    : callback_(callback) {
+  }
+
+  void Start(mojo::ScopedDataPipeConsumerHandle handle) {
+    buffer_ = SharedBuffer::create();
+    drainer_.reset(new mojo::common::DataPipeDrainer(this, handle.Pass()));
+  }
+
+ private:
+  // mojo::common::DataPipeDrainer::Client
+  void OnDataAvailable(const void* data, size_t num_bytes) override {
+    buffer_->append(static_cast<const char*>(data), num_bytes);
+  }
+
+  void OnDataComplete() override {
+    Platform::current()->GetUITaskRunner()->PostTask(FROM_HERE,
+      base::Bind(callback_, buffer_.release()));
+    delete this;
+  }
+
+  base::Callback<void(PassRefPtr<SharedBuffer>)> callback_;
+  RefPtr<SharedBuffer> buffer_;
+  std::unique_ptr<mojo::common::DataPipeDrainer> drainer_;
+
+  DISALLOW_COPY_AND_ASSIGN(DrainJob);
+};
+
+} // namespace
+
+
+void DrainDataPipeInBackground(
+    mojo::ScopedDataPipeConsumerHandle handle,
+    base::Callback<void(PassRefPtr<SharedBuffer>)> callback) {
+  DrainJob* job = new DrainJob(callback);
+  Platform::current()->GetIOTaskRunner()->PostTask(FROM_HERE,
+    base::Bind(&DrainJob::Start, base::Unretained(job), base::Passed(&handle)));
+}
+
+}  // namespace blink

sky/engine/platform/mojo/data_pipe.h

+// Copyright 2013 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.
+
+#ifndef SKY_ENGINE_PLATFORM_MOJO_DATA_PIPE_H_
+#define SKY_ENGINE_PLATFORM_MOJO_DATA_PIPE_H_
+
+#include "base/callback.h"
+#include "mojo/public/cpp/system/core.h"
+#include "sky/engine/platform/SharedBuffer.h"
+
+namespace blink {
+
+void DrainDataPipeInBackground(
+    mojo::ScopedDataPipeConsumerHandle handle,
+    base::Callback<void(PassRefPtr<SharedBuffer>)> callback);
+
+}  // namespace blink
+
+#endif  // SKY_ENGINE_PLATFORM_MOJO_DATA_PIPE_H_

sky/engine/public/platform/Platform.h

@@ -170,8 +170,8 @@ public:
     // Returns a value such as "en-US".
     virtual WebString defaultLocale() { return WebString(); }
 
-    virtual base::SingleThreadTaskRunner* mainThreadTaskRunner() { return 0; }
-
+    virtual base::SingleThreadTaskRunner* GetUITaskRunner() { return 0; }
+    virtual base::SingleThreadTaskRunner* GetIOTaskRunner() { return 0; }
 
     // Vibration -----------------------------------------------------------
 

sky/shell/shell.cc

@@ -62,9 +62,15 @@ Shell::Shell(scoped_ptr<ServiceProviderContext> service_provider_context)
 
   gpu_thread_.reset(new base::Thread("gpu_thread"));
   gpu_thread_->StartWithOptions(options);
+  gpu_task_runner_ = gpu_thread_->message_loop()->task_runner();
 
   ui_thread_.reset(new base::Thread("ui_thread"));
   ui_thread_->StartWithOptions(options);
+  ui_task_runner_ = ui_thread_->message_loop()->task_runner();
+
+  io_thread_.reset(new base::Thread("io_thread"));
+  io_thread_->StartWithOptions(options);
+  io_task_runner_ = io_thread_->message_loop()->task_runner();
 
   ui_task_runner()->PostTask(FROM_HERE, base::Bind(&Engine::Init));
 }

sky/shell/shell.h

@@ -25,12 +25,16 @@ class Shell {
       scoped_ptr<ServiceProviderContext> service_provider_context);
   static Shell& Shared();
 
-  scoped_refptr<base::SingleThreadTaskRunner> gpu_task_runner() const {
-    return gpu_thread_->message_loop()->task_runner();
+  base::SingleThreadTaskRunner* gpu_task_runner() const {
+    return gpu_task_runner_.get();
   }
 
-  scoped_refptr<base::SingleThreadTaskRunner> ui_task_runner() const {
-    return ui_thread_->message_loop()->task_runner();
+  base::SingleThreadTaskRunner* ui_task_runner() const {
+    return ui_task_runner_.get();
+  }
+
+  base::SingleThreadTaskRunner* io_task_runner() const {
+    return io_task_runner_.get();
   }
 
   ServiceProviderContext* service_provider_context() const {
@@ -47,6 +51,12 @@ class Shell {
 
   scoped_ptr<base::Thread> gpu_thread_;
   scoped_ptr<base::Thread> ui_thread_;
+  scoped_ptr<base::Thread> io_thread_;
+
+  scoped_refptr<base::SingleThreadTaskRunner> gpu_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> ui_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_;
+
   scoped_ptr<ServiceProviderContext> service_provider_context_;
   TracingController tracing_controller_;
 

sky/shell/ui/platform_impl.cc

@@ -4,11 +4,12 @@
 
 #include "sky/shell/ui/platform_impl.h"
 
+#include "sky/shell/shell.h"
+
 namespace sky {
 namespace shell {
 
-PlatformImpl::PlatformImpl()
-    : main_thread_task_runner_(base::MessageLoop::current()->task_runner()) {
+PlatformImpl::PlatformImpl() {
 }
 
 PlatformImpl::~PlatformImpl() {
@@ -18,8 +19,12 @@ blink::WebString PlatformImpl::defaultLocale() {
   return blink::WebString::fromUTF8("en-US");
 }
 
-base::SingleThreadTaskRunner* PlatformImpl::mainThreadTaskRunner() {
-  return main_thread_task_runner_.get();
+base::SingleThreadTaskRunner* PlatformImpl::GetUITaskRunner() {
+  return Shell::Shared().ui_task_runner();
+}
+
+base::SingleThreadTaskRunner* PlatformImpl::GetIOTaskRunner() {
+  return Shell::Shared().io_task_runner();
 }
 
 }  // namespace shell

sky/shell/ui/platform_impl.h

@@ -18,11 +18,11 @@ class PlatformImpl : public blink::Platform {
 
   // blink::Platform methods:
   blink::WebString defaultLocale() override;
-  base::SingleThreadTaskRunner* mainThreadTaskRunner() override;
 
- private:
-  scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner_;
+  base::SingleThreadTaskRunner* GetUITaskRunner() override;
+  base::SingleThreadTaskRunner* GetIOTaskRunner() override;
 
+ private:
   DISALLOW_COPY_AND_ASSIGN(PlatformImpl);
 };