[proof-of-concept] SW thread independent of the main thread

content/renderer/render_thread_impl.cc

Index: content/renderer/render_thread_impl.cc
diff --git a/content/renderer/render_thread_impl.cc b/content/renderer/render_thread_impl.cc
index 431af215394334381e53e98d50cd39439fe93d87..5625946db27d30c9fb73ac5032a089d34dc9cb46 100644
--- a/content/renderer/render_thread_impl.cc
+++ b/content/renderer/render_thread_impl.cc
@@ -148,6 +148,9 @@
 #include "third_party/WebKit/public/platform/scheduler/child/compositor_worker_scheduler.h"
 #include "third_party/WebKit/public/platform/scheduler/child/webthread_impl_for_worker_scheduler.h"
 #include "third_party/WebKit/public/platform/scheduler/renderer/renderer_scheduler.h"
+#include "third_party/WebKit/public/platform/modules/serviceworker/service_worker.mojom.h"
+#include "third_party/WebKit/public/web/modules/serviceworker/WebIsolatedWorkerContextClient.h"
+#include "third_party/WebKit/public/web/modules/serviceworker/WebIsolatedWorker.h"
 #include "third_party/WebKit/public/web/WebCache.h"
 #include "third_party/WebKit/public/web/WebDatabase.h"
 #include "third_party/WebKit/public/web/WebDocument.h"
@@ -162,6 +165,9 @@
 #include "third_party/skia/include/core/SkGraphics.h"
 #include "ui/base/layout.h"
 #include "ui/base/ui_base_switches.h"
+#include "mojo/public/cpp/system/watcher.h"
+#include "content/child/service_worker/service_worker_registration_handle_reference.h"
+#include "content/child/service_worker/web_service_worker_registration_impl.h"
 
 #if defined(OS_ANDROID)
 #include <cpu-features.h>
@@ -434,8 +440,407 @@ bool IsRunningInMash() {
   return cmdline->HasSwitch(switches::kIsRunningInMash);
 }
 
+class IsolatedWorkerContextClient
+    : public blink::WebIsolatedWorkerContextClient {
+ public:
+  IsolatedWorkerContextClient(
+      const std::string& scope,
+      shell::mojom::InterfaceProviderRequest request,
+      shell::mojom::InterfaceProviderPtr remote_interfaces,
+      mojom::WorkerScriptListPtr worker_script_list,
+      scoped_refptr<base::SingleThreadTaskRunner> io_thread_task_runner,
+      base::Closure worker_destroyed_closure)
+      : scope_(scope),
+        interface_bind_callback_(
+            base::Bind(&IsolatedWorkerContextClient::BindInterfaceProviders,
+                       base::Passed(&request),
+                       base::Passed(remote_interfaces.PassInterface()))),
+        io_thread_task_runner_(io_thread_task_runner),
+        worker_destroyed_closure_(worker_destroyed_closure),
+        worker_script_list_(std::move(worker_script_list)),
+        script_ready_(base::WaitableEvent::ResetPolicy::MANUAL,
+                      base::WaitableEvent::InitialState::NOT_SIGNALED),
+        weak_factory_(this) {
+    TRACE_EVENT0("ServiceWorker",
+                 "IsolatedWorkerContextClient::IsolatedWorkerContextClient");
+    io_thread_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&IsolatedWorkerContextClient::StartScriptRead,
+                              weak_factory_.GetWeakPtr()));
+  }
+  ~IsolatedWorkerContextClient() override {}
+
+  // Must be called in the worker thread.
+  void initializeOnWorkerThread() override {
+    TRACE_EVENT0("ServiceWorker",
+                 "IsolatedWorkerContextClient::initializeOnWorkerThread");
+    context_.reset(new WorkerContextData());
+    interface_bind_callback_.Run(context_.get());
+    interface_bind_callback_.Reset();
+  }
+  // Must be called in the worker thread.
+  void connectToRemoteService(const char* name,
+                              mojo::ScopedMessagePipeHandle handle) override {
+    TRACE_EVENT0("ServiceWorker",
+                 "IsolatedWorkerContextClient::connectToRemoteService");
+    DCHECK(context_);
+    context_->remote_interfaces.GetInterface(name, std::move(handle));
+  }
+
+  // Must be called in the worker thread.
+  std::unique_ptr<blink::WebServiceWorkerRegistration::Handle>
+  takeAssociatedRegistration() override {
+    ServiceWorkerRegistrationObjectInfo info;
+    ServiceWorkerVersionAttributes attrs;
+    info.scope = GURL(scope_);
+    info.handle_id = 1;        // test
+    info.registration_id = 1;  // test
+
+    scoped_refptr<WebServiceWorkerRegistrationImpl> registration(
+        new WebServiceWorkerRegistrationImpl(
+            ServiceWorkerRegistrationHandleReference::Create(info, nullptr)));
+
+    return WebServiceWorkerRegistrationImpl::CreateHandle(registration);
+  }
+
+  // Must be called in the worker thread.
+  std::unique_ptr<std::vector<std::unique_ptr<WebScriptData>>>
+  takeWorkerScriptList() override {
+    TRACE_EVENT0("ServiceWorker",
+                 "IsolatedWorkerContextClient::takeWorkerScriptList");
+    LOG(ERROR) << "takeWorkerScriptList";
+    {
+      TRACE_EVENT0("ServiceWorker",
+                   "IsolatedWorkerContextClient::takeWorkerScriptList wait");
+      LOG(ERROR) << "script_ready_.Wait();";
+      script_ready_.Wait();
+      LOG(ERROR) << "script_ready_.Wait(); done";
+    }
+    std::unique_ptr<std::vector<std::unique_ptr<WebScriptData>>> list(
+        new std::vector<std::unique_ptr<WebScriptData>>());
+
+    for (auto& it : data_map_) {
+      std::unique_ptr<WebScriptData> script_data(new WebScriptData(
+          it.first, std::move(it.second), std::move(metadata_map_[it.first])));
+      list->push_back(std::move(script_data));
+    }
+
+    return list;
+  }
+
+  // Must be called in the worker thread.
+  void workerDestroyed() override {
+    context_.reset();
+    io_thread_task_runner_->PostTask(FROM_HERE, worker_destroyed_closure_);
+    // |this| is deleted
+  }
+
+ private:
+  struct WorkerContextData {
+    WorkerContextData() : interface_registry() {}
+    ~WorkerContextData() { DCHECK(thread_checker.CalledOnValidThread()); }
+    base::ThreadChecker thread_checker;
+    shell::InterfaceRegistry interface_registry;
+    shell::InterfaceProvider remote_interfaces;
+  };
+  class ScriptPump {
+   public:
+    ScriptPump(const std::string& url,
+               mojo::ScopedDataPipeConsumerHandle pipe,
+               base::Callback<void(ScriptPump*)> callback)
+        : url_(url),
+          pipe_(std::move(pipe)),
+          data_vector(new DataVector),
+          callback_(callback),
+          weak_factory_(this) {}
+    ~ScriptPump() {}
+    void StartRead() { ReadNext(); }
+    std::unique_ptr<DataVector> TakeData() { return std::move(data_vector); }
+    std::string url() const { return url_; }
+
+   private:
+    void OnReadable(MojoResult unused) {
+      LOG(ERROR) << "ScriptPump::OnReadable";
+      ReadNext();
+    }
+    void ReadNext() {
+      LOG(ERROR) << "ScriptPump::ReadNext";
+      TRACE_EVENT0("ServiceWorker", "ScriptPump::ReadNext");
+
+      const void* buffer = nullptr;
+      uint32_t available = 0;
+      MojoResult result = mojo::BeginReadDataRaw(
+          pipe_.get(), &buffer, &available, MOJO_READ_DATA_FLAG_NONE);
+      // LOG(ERROR) << " mojo::BeginReadDataRaw: " << result;
+      // LOG(ERROR) << " mojo::BeginReadDataRaw: available: " << available;
+
+      // TODO: We need error handling code.
+      if (result == MOJO_RESULT_FAILED_PRECONDITION) {
+        callback_.Run(this);
+        // Deleted
+        return;
+      }
+      if (result == MOJO_RESULT_OK) {
+        if (available > 0) {
+          std::unique_ptr<std::vector<char>> data(new std::vector<char>(
+              static_cast<const char*>(buffer),
+              static_cast<const char*>(buffer) + available));
+          data_vector->push_back(std::move(data));
+        }
+        result = mojo::EndReadDataRaw(pipe_.get(), available);
+        // LOG(ERROR) << " mojo::EndReadDataRaw: " << result;
+        ReadNext();
+        return;
+      }
+      if (result == MOJO_RESULT_SHOULD_WAIT) {
+        LOG(ERROR) << "handle_watcher_.Start";
+        handle_watcher_.Start(
+            pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+            base::Bind(&ScriptPump::OnReadable, weak_factory_.GetWeakPtr()));
+      }
+    }
+
+    const std::string url_;
+    mojo::ScopedDataPipeConsumerHandle pipe_;
+    std::unique_ptr<DataVector> data_vector;
+    base::Callback<void(ScriptPump*)> callback_;
+    mojo::Watcher handle_watcher_;
+    base::WeakPtrFactory<ScriptPump> weak_factory_;
+  };
+
+  // Must be called in the worker thread.
+  static void BindInterfaceProviders(
+      shell::mojom::InterfaceProviderRequest request,
+      shell::mojom::InterfaceProviderPtrInfo remote_interfaces,
+      WorkerContextData* context) {
+    DCHECK(context->thread_checker.CalledOnValidThread());
+    context->interface_registry.Bind(std::move(request));
+    context->remote_interfaces.Bind(
+        mojo::MakeProxy(std::move(remote_interfaces)));
+  }
+  // Must be called in IO thread.
+  void StartScriptRead() {
+    LOG(ERROR) << "StartScriptRead: size: "
+               << worker_script_list_->scripts.size();
+    std::vector<ScriptPump*> pump_list;
+    for (size_t i = 0; i < worker_script_list_->scripts.size(); ++i) {
+      std::unique_ptr<ScriptPump> pump(
+          new ScriptPump(worker_script_list_->scripts[i]->url,
+                         std::move(worker_script_list_->scripts[i]->data_pipe),
+                         base::Bind(&IsolatedWorkerContextClient::PumpFinished,
+                                    weak_factory_.GetWeakPtr())));
+      ScriptPump* ptr = pump.get();
+      pump_map_[ptr] = std::move(pump);
+      pump_list.push_back(ptr);
+
+      std::unique_ptr<ScriptPump> metadata_pump(new ScriptPump(
+          worker_script_list_->scripts[i]->url,
+          std::move(worker_script_list_->scripts[i]->meta_data_pipe),
+          base::Bind(&IsolatedWorkerContextClient::MetadataPumpFinished,
+                     weak_factory_.GetWeakPtr())));
+      ScriptPump* metadata_pump_ptr = metadata_pump.get();
+      metadata_pump_map_[metadata_pump_ptr] = std::move(metadata_pump);
+      pump_list.push_back(metadata_pump_ptr);
+    }
+    for (auto& pump : pump_list) {
+      pump->StartRead();
+    }
+    LOG(ERROR) << "pump_map_.size() " << pump_map_.size()
+               << " metadata_pump_map_.size(): " << metadata_pump_map_.size();
+  }
+  void PumpFinished(ScriptPump* pump) {
+    TRACE_EVENT0("ServiceWorker", "ScriptPump::PumpFinished");
+    LOG(ERROR) << "ScriptPump::PumpFinished";
+    data_map_[pump->url()] = pump->TakeData();
+    pump_map_.erase(pump);
+    LOG(ERROR) << "pump_map_.size() " << pump_map_.size()
+               << " metadata_pump_map_.size(): " << metadata_pump_map_.size();
+    if (pump_map_.empty() && metadata_pump_map_.empty()) {
+      TRACE_EVENT0("ServiceWorker", "ScriptPump::PumpFinished ALL");
+      LOG(ERROR) << "ScriptPump::PumpFinished ALL";
+      script_ready_.Signal();
+    }
+  }
+  void MetadataPumpFinished(ScriptPump* pump) {
+    TRACE_EVENT0("ServiceWorker", "ScriptPump::MetadataPumpFinished");
+    LOG(ERROR) << "ScriptPump::MetadataPumpFinished";
+    metadata_map_[pump->url()] = pump->TakeData();
+    metadata_pump_map_.erase(pump);
+    LOG(ERROR) << "pump_map_.size() " << pump_map_.size()
+               << " metadata_pump_map_.size(): " << metadata_pump_map_.size();
+    if (pump_map_.empty() && metadata_pump_map_.empty()) {
+      TRACE_EVENT0("ServiceWorker", "ScriptPump::MetadataPumpFinished ALL");
+      LOG(ERROR) << "ScriptPump::MetadataPumpFinished ALL";
+      script_ready_.Signal();
+    }
+  }
+
+  // Initialized on the worker thread in workerContextStarted and
+  // destructed on the worker thread in willDestroyWorkerContext.
+  std::unique_ptr<WorkerContextData> context_;
+  const std::string scope_;
+  base::Callback<void(WorkerContextData*)> interface_bind_callback_;
+  scoped_refptr<base::SingleThreadTaskRunner> io_thread_task_runner_;
+  base::Closure worker_destroyed_closure_;
+  mojom::WorkerScriptListPtr worker_script_list_;
+  base::WaitableEvent script_ready_;
+  base::hash_map<ScriptPump*, std::unique_ptr<ScriptPump>> pump_map_;
+  base::hash_map<ScriptPump*, std::unique_ptr<ScriptPump>> metadata_pump_map_;
+  base::hash_map<std::string, std::unique_ptr<DataVector>> data_map_;
+  base::hash_map<std::string, std::unique_ptr<DataVector>> metadata_map_;
+  base::WeakPtrFactory<IsolatedWorkerContextClient> weak_factory_;
+};
+
 }  // namespace
 
+class IsolatedWorkerManager
+    : public base::RefCountedThreadSafe<IsolatedWorkerManager> {
+ public:
+  IsolatedWorkerManager(
+      scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner,
+      scoped_refptr<base::SingleThreadTaskRunner> io_thread_task_runner,
+      mojom::IsolatedWorkerDispatcherRecieverPtr dispatcher_reciever)
+      : main_thread_task_runner_(main_thread_task_runner),
+        io_thread_task_runner_(io_thread_task_runner),
+        weak_factory_(this) {
+    TRACE_EVENT0("ServiceWorker",
+                 "IsolatedWorkerManager::IsolatedWorkerManager");
+    LOG(ERROR) << "IsolatedWorkerManager " << this;
+    io_thread_task_runner->PostTask(
+        FROM_HERE,
+        base::Bind(&IsolatedWorkerManager::InitializeOnIO, this,
+                   base::Passed(dispatcher_reciever.PassInterface())));
+  }
+  void InitializeOnIO(
+      mojom::IsolatedWorkerDispatcherRecieverPtrInfo dispatcher_reciever_info) {
+    TRACE_EVENT0("ServiceWorker", "IsolatedWorkerManager::InitializeOnIO");
+    mojom::IsolatedWorkerDispatcherRecieverPtr dispatcher_reciever;
+    dispatcher_reciever.Bind(std::move(dispatcher_reciever_info));
+    DCHECK(io_thread_task_runner_->BelongsToCurrentThread());
+    LOG(ERROR) << "InitializeOnIO " << this;
+    // pass
+    LOG(ERROR) << "IsolatedWorkerDispatcherImpl ";
+    dispatcher_.reset(new IsolatedWorkerDispatcherImpl(this));
+    mojom::IsolatedWorkerDispatcherPtr dispatcher_ptr;
+    LOG(ERROR) << " dispatcher_->Bind ";
+    dispatcher_->Bind(GetProxy(&dispatcher_ptr));
+    LOG(ERROR) << " dispatcher_reciever->SetDispatcher ";
+    dispatcher_reciever->SetDispatcher(std::move(dispatcher_ptr));
+    LOG(ERROR) << " dispatcher_reciever->SetDispatcher done";
+  }
+
+  void StartServiceWorker(int64_t version_id,
+                          const std::string& scope,
+                          const std::string& script_url,
+                          shell::mojom::InterfaceProviderRequest request,
+                          shell::mojom::InterfaceProviderPtr remote_interfaces,
+                          mojom::WorkerScriptListPtr worker_script_list) {
+    TRACE_EVENT0("ServiceWorker", "IsolatedWorkerManager::StartServiceWorker");
+    DCHECK(io_thread_task_runner_->BelongsToCurrentThread());
+    int32_t isolated_worker_id = next_isolated_worker_id_++;
+    std::unique_ptr<IsolatedWorkerContextClient> client(
+        new IsolatedWorkerContextClient(
+            scope, std::move(request), std::move(remote_interfaces),
+            std::move(worker_script_list), io_thread_task_runner_,
+            base::Bind(&IsolatedWorkerManager::WorkerDestroyed, this,
+                       isolated_worker_id)));
+    std::unique_ptr<blink::WebIsolatedWorker> worker(
+        blink::WebIsolatedWorker::create(client.release()));
+    std::unique_ptr<WorkerWrapper> wrapper(
+        new WorkerWrapper(std::move(worker), main_thread_task_runner_));
+    wrapper->worker()->startWorker(version_id, GURL(scope), GURL(script_url));
+    workers_.AddWithID(wrapper.release(), isolated_worker_id);
+  }
+
+  base::WeakPtr<IsolatedWorkerManager> AsWeakPtr() {
+    return weak_factory_.GetWeakPtr();
+  }
+
+ private:
+  friend class base::RefCountedThreadSafe<IsolatedWorkerManager>;
+
+  // Lives in
+  class IsolatedWorkerDispatcherImpl : public mojom::IsolatedWorkerDispatcher {
+   public:
+    explicit IsolatedWorkerDispatcherImpl(
+        scoped_refptr<IsolatedWorkerManager> manager)
+        : manager_(manager), binding_(this) {
+      TRACE_EVENT0(
+          "ServiceWorker",
+          "IsolatedWorkerDispatcherImpl::IsolatedWorkerDispatcherImpl");
+      LOG(ERROR) << "IsolatedWorkerDispatcherImpl" << this;
+    }
+    ~IsolatedWorkerDispatcherImpl() override {
+      LOG(ERROR) << "~IsolatedWorkerDispatcherImpl" << this;
+    }
+    void Bind(mojom::IsolatedWorkerDispatcherRequest local_interfaces_request) {
+      DCHECK(!binding_.is_bound());
+      binding_.Bind(std::move(local_interfaces_request));
+    }
+
+    void StartServiceWorker(
+        int64_t version_id,
+        const std::string& scope,
+        const std::string& script_url,
+        ::shell::mojom::InterfaceProviderRequest request,
+        ::shell::mojom::InterfaceProviderPtr remote_interfaces,
+        mojom::WorkerScriptListPtr script_list) override {
+      LOG(ERROR) << "IsolatedWorkerDispatcherImpl[[2]] :: StartServiceWorker";
+      if (!manager_)
+        return;
+      manager_->StartServiceWorker(
+          version_id, scope, script_url, std::move(request),
+          std::move(remote_interfaces), std::move(script_list));
+    }
+
+   private:
+    scoped_refptr<IsolatedWorkerManager> manager_;
+    mojo::Binding<mojom::IsolatedWorkerDispatcher> binding_;
+  };
+  class WorkerWrapper {
+   public:
+    WorkerWrapper(
+        std::unique_ptr<blink::WebIsolatedWorker> worker,
+        scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner)
+        : worker_(std::move(worker)),
+          main_thread_task_runner_(main_thread_task_runner) {
+      main_thread_task_runner_->PostTask(
+          FROM_HERE, base::Bind(WorkerWrapper::AddRefProcessOnMain));
+    }
+    ~WorkerWrapper() {
+      main_thread_task_runner_->PostTask(
+          FROM_HERE, base::Bind(WorkerWrapper::ReleaseProcessOnMain));
+    }
+    blink::WebIsolatedWorker* worker() { return worker_.get(); }
+
+   private:
+    static void AddRefProcessOnMain() {
+      LOG(ERROR) << "AddRefProcessOnMain-------------";
+      // TODO: ChildProcess::current()->AddRefProcess();
+    }
+    static void ReleaseProcessOnMain() {
+      LOG(ERROR) << "ReleaseProcessOnMain-------------";
+      // TODO:  ChildProcess::current()->ReleaseProcess();
+    }
+    std::unique_ptr<blink::WebIsolatedWorker> worker_;
+    scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner_;
+  };
+
+  ~IsolatedWorkerManager() { LOG(ERROR) << "~IsolatedWorkerManager " << this; }
+  void WorkerDestroyed(int64_t isolated_worker_id) {
+    workers_.Remove(isolated_worker_id);
+  }
+  scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> io_thread_task_runner_;
+
+  std::unique_ptr<IsolatedWorkerDispatcherImpl> dispatcher_;
+
+  int32_t next_isolated_worker_id_ = 0;
+  IDMap<WorkerWrapper, IDMapOwnPointer> workers_;
+  base::WeakPtrFactory<IsolatedWorkerManager> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(IsolatedWorkerManager);
+};
+
 // For measuring memory usage after each task. Behind a command line flag.
 class MemoryObserver : public base::MessageLoop::TaskObserver {
  public:
@@ -851,6 +1256,18 @@ void RenderThreadImpl::Init(
 
   GetInterfaceRegistry()->AddInterface(base::Bind(CreateFrameFactory));
   GetInterfaceRegistry()->AddInterface(base::Bind(CreateEmbeddedWorkerSetup));
+  LOG(ERROR) << "GetInterfaceRegistry()->AddInterface";
+  LOG(ERROR) << "base::ThreadTaskRunnerHandle::IsSet() "
+             << base::ThreadTaskRunnerHandle::IsSet();
+
+  mojom::IsolatedWorkerDispatcherRecieverPtr dispatcher_reciever;
+  LOG(ERROR) << "GetRemoteInterfaces()->GetInterface";
+  GetRemoteInterfaces()->GetInterface(mojo::GetProxy(&dispatcher_reciever));
+
+  isolated_worker_manager_ =
+      new IsolatedWorkerManager(base::ThreadTaskRunnerHandle::Get(),
+                                ChildProcess::current()->io_task_runner(),
+                                std::move(dispatcher_reciever));
 
   GetRemoteInterfaces()->GetInterface(
       mojo::GetProxy(&storage_partition_service_));