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

content/browser/service_worker/service_worker_version.cc

Index: content/browser/service_worker/service_worker_version.cc
diff --git a/content/browser/service_worker/service_worker_version.cc b/content/browser/service_worker/service_worker_version.cc
index fb22db42dea541943905818470e2f64d03ae67be..b721069810c984de1860e0b4f166636a0f3887b5 100644
--- a/content/browser/service_worker/service_worker_version.cc
+++ b/content/browser/service_worker/service_worker_version.cc
@@ -47,6 +47,19 @@
 #include "mojo/common/common_type_converters.h"
 #include "net/http/http_response_headers.h"
 #include "net/http/http_response_info.h"
+#include "third_party/WebKit/public/platform/modules/serviceworker/service_worker.mojom.h"
+#include "mojo/public/cpp/bindings/strong_binding.h"
+#include "content/browser/renderer_host/render_process_host_impl.h"
+#include "services/shell/public/cpp/interface_provider.h"
+#include "services/shell/public/cpp/interface_registry.h"
+#include "base/bind_helpers.h"
+#include "content/common/service_worker/embedded_worker_setup.mojom.h"
+#include "content/browser/appcache/appcache_response.h"
+#include "content/browser/service_worker/service_worker_storage.h"
+#include "content/browser/service_worker/service_worker_disk_cache.h"
+#include "mojo/public/cpp/system/watcher.h"
+
+#include "net/base/io_buffer.h"
 
 namespace content {
 
@@ -604,6 +617,10 @@ void ServiceWorkerVersion::RunAfterStartWorker(
 void ServiceWorkerVersion::DispatchEvent(const std::vector<int>& request_ids,
                                          const IPC::Message& message) {
   DCHECK_EQ(EmbeddedWorkerStatus::RUNNING, running_status());
+  if (IsIsolatedWorker()) {
+    LOG(FATAL) << "ServiceWorkerVersion::DispatchEvent is not supported in "
+                  "IsolatedWorker";
+  }
 
   const ServiceWorkerStatusCode status = embedded_worker_->SendMessage(message);
 
@@ -1388,6 +1405,348 @@ void ServiceWorkerVersion::DidEnsureLiveRegistrationForStartWorker(
   DCHECK(timeout_timer_.IsRunning());
 }
 
+namespace {
+
+void OnIsolatedWorkerDispatcher(
+    int isolated_worker_id,
+    const GURL scope,
+    const GURL script_url,
+    shell::mojom::InterfaceProviderRequest remote_interfaces,
+    shell::mojom::InterfaceProviderPtrInfo exposed_interfaces,
+    mojom::WorkerScriptListPtr script_list,
+    mojo::InterfacePtr<mojom::IsolatedWorkerDispatcher>* dispatcher) {
+  TRACE_EVENT0("ServiceWorker", "OnIsolatedWorkerDispatcher");
+
+  (*dispatcher)
+      ->StartServiceWorker(isolated_worker_id, scope.spec(), script_url.spec(),
+                           std::move(remote_interfaces),
+                           mojo::MakeProxy(std::move(exposed_interfaces)),
+                           std::move(script_list));
+}
+
+void OnFindAvailableProcess(
+    int isolated_worker_id,
+    const GURL scope,
+    const GURL script_url,
+    shell::mojom::InterfaceProviderRequest remote_interfaces,
+    shell::mojom::InterfaceProviderPtrInfo exposed_interfaces,
+    mojom::WorkerScriptListPtr script_list,
+    base::Closure failure_callback,
+    int process_id) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  LOG(ERROR) << "OnFindAvailableProcess " << process_id;
+  RenderProcessHost* rph = RenderProcessHost::FromID(process_id);
+  // |rph| may be NULL in unit tests.
+  // rph = nullptr;
+  if (!rph) {
+    BrowserThread::PostTask(BrowserThread::IO, FROM_HERE, failure_callback);
+    return;
+  }
+  (reinterpret_cast<RenderProcessHostImpl*>(rph))
+      ->GetIsolatedWorkerDispatcher(base::Bind(
+          &OnIsolatedWorkerDispatcher, isolated_worker_id, scope, script_url,
+          base::Passed(&remote_interfaces), base::Passed(&exposed_interfaces),
+          base::Passed(std::move(script_list))));
+}
+
+}  // namespace
+
+class IsolatedWorkerInstance {
+ public:
+  IsolatedWorkerInstance(
+      base::WeakPtr<ServiceWorkerContextCore> context,
+      base::WeakPtr<ServiceWorkerVersion> version,
+      const std::vector<ServiceWorkerDatabase::ResourceRecord>& resources,
+      base::Closure failure_callback)
+      : context_(context), version_(version), weak_factory_(this) {
+    mojom::WorkerScriptListPtr script_list = mojom::WorkerScriptList::New();
+
+    for (const auto& record : resources) {
+      if (version_->script_url() == record.url)
+        PushScriptPump(record.resource_id, record.url, &script_list->scripts);
+    }
+    for (const auto& record : resources) {
+      if (version_->script_url() != record.url)
+        PushScriptPump(record.resource_id, record.url, &script_list->scripts);
+    }
+
+    isolated_worker_interface_registry_.reset(new shell::InterfaceRegistry());
+    isolated_worker_remote_interfaces_.reset(new shell::InterfaceProvider);
+    isolated_worker_interface_registry_->AddInterface(
+        base::Bind(&IsolatedWorkerInstance::CreateHostService,
+                   weak_factory_.GetWeakPtr()));
+
+    // dmy code
+    shell::mojom::InterfaceProviderPtr exposed_interfaces;
+    isolated_worker_interface_registry_->Bind(GetProxy(&exposed_interfaces));
+    shell::mojom::InterfaceProviderPtr remote_interfaces;
+    shell::mojom::InterfaceProviderRequest request =
+        GetProxy(&remote_interfaces);
+    isolated_worker_remote_interfaces_->Bind(std::move(remote_interfaces));
+
+    context->process_manager()->FindAvailableProcessAndCallbackOnUI(
+        version_->scope(),
+        base::Bind(&OnFindAvailableProcess, version_->version_id(),
+                   version_->scope(), version_->script_url(),
+                   base::Passed(&request),
+                   base::Passed(exposed_interfaces.PassInterface()),
+                   base::Passed(std::move(script_list)), failure_callback));
+  }
+  ~IsolatedWorkerInstance() {}
+
+ private:
+  class WriterIOBuffer final : public net::IOBufferWithSize {
+   public:
+    WriterIOBuffer(void* data, size_t size)
+        : net::IOBufferWithSize(static_cast<char*>(data), size) {}
+
+   private:
+    ~WriterIOBuffer() override { data_ = nullptr; }
+    DISALLOW_COPY_AND_ASSIGN(WriterIOBuffer);
+  };
+  class ScriptPump {
+   public:
+    ScriptPump(base::WeakPtr<IsolatedWorkerInstance> instance,
+               std::unique_ptr<ServiceWorkerResponseReader> reader,
+               const GURL& url,
+               mojo::ScopedDataPipeProducerHandle stream,
+               mojo::ScopedDataPipeProducerHandle metadata_stream)
+        : instance_(instance),
+          reader_(std::move(reader)),
+          url_(url),
+          stream_(std::move(stream)),
+          metadata_sender_(new MetadataSender(std::move(metadata_stream))),
+          weak_factory_(this) {}
+    void StartRead() {
+      http_info_io_buffer_ = new HttpResponseInfoIOBuffer;
+      reader_->ReadInfo(http_info_io_buffer_.get(),
+                        base::Bind(&ScriptPump::OnReadInfoComplete,
+                                   weak_factory_.GetWeakPtr()));
+    }
+    ~ScriptPump() {}
+
+   private:
+    class MetadataSender {
+     public:
+      MetadataSender(mojo::ScopedDataPipeProducerHandle metadata_stream)
+          : metadata_stream_(std::move(metadata_stream)), weak_factory_(this) {}
+      ~MetadataSender() {}
+      void StartSend(scoped_refptr<net::IOBufferWithSize> buf,
+                     base::Closure callback) {
+        TRACE_EVENT0("ServiceWorker", "MetadataSender::StartSend");
+        buffer_ = buf;
+        callback_ = callback;
+        remaining_ = buf->size();
+        SendNext();
+      }
+      void OnWritable(MojoResult unused) { SendNext(); }
+      void SendNext() {
+        TRACE_EVENT0("ServiceWorker", "MetadataSender::SendNext");
+        uint32_t num_bytes = remaining_;
+        MojoResult mojo_result =
+            mojo::WriteDataRaw(metadata_stream_.get(),
+                               buffer_->data() + (buffer_->size() - remaining_),
+                               &num_bytes,  // In/out.
+                               MOJO_WRITE_DATA_FLAG_NONE);
+        if (mojo_result == MOJO_RESULT_SHOULD_WAIT) {
+          handle_watcher_.Start(metadata_stream_.get(),
+                                MOJO_HANDLE_SIGNAL_WRITABLE,
+                                base::Bind(&MetadataSender::OnWritable,
+                                           weak_factory_.GetWeakPtr()));
+          return;
+        }
+        remaining_ -= num_bytes;
+        if (remaining_ == 0) {
+          callback_.Run();
+          return;
+        }
+        SendNext();
+      }
+
+     private:
+      mojo::ScopedDataPipeProducerHandle metadata_stream_;
+      base::Closure callback_;
+      scoped_refptr<net::IOBufferWithSize> buffer_;
+      uint32_t remaining_;
+      mojo::Watcher handle_watcher_;
+      base::WeakPtrFactory<MetadataSender> weak_factory_;
+    };
+
+    void OnReadInfoComplete(int result) {
+      TRACE_EVENT0("ServiceWorker", "ScriptPump::OnReadInfoComplete");
+      // TODO: send metadata
+      if (!http_info_io_buffer_->http_info->metadata) {
+        metadata_sender_ = nullptr;
+      } else {
+        metadata_sender_->StartSend(http_info_io_buffer_->http_info->metadata,
+                                    base::Bind(&ScriptPump::OnMetadataSent,
+                                               weak_factory_.GetWeakPtr()));
+      }
+      ReadNext();
+    }
+    void OnMetadataSent() {
+      TRACE_EVENT0("ServiceWorker", "MetadataSender::OnMetadataSent");
+      metadata_sender_ = nullptr;
+      if (sent_all_data_) {
+        instance_->PumpFinished(this);
+        return;
+      }
+    }
+    void OnWritable(MojoResult unused) { ReadNext(); }
+    void ReadNext() {
+      TRACE_EVENT0("ServiceWorker", "ScriptPump::ReadNext");
+      LOG(ERROR) << "ScriptPump::ReadNext";
+      void* buf = nullptr;
+      uint32_t available = 0;
+      MojoResult mojo_result = mojo::BeginWriteDataRaw(
+          stream_.get(), &buf, &available, MOJO_WRITE_DATA_FLAG_NONE);
+      // TODO: We need error handling code.
+      if (mojo_result == MOJO_RESULT_SHOULD_WAIT) {
+        handle_watcher_.Start(
+            stream_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+            base::Bind(&ScriptPump::OnWritable, weak_factory_.GetWeakPtr()));
+        return;
+      }
+      buffer_ = new WriterIOBuffer(buf, available);
+
+      reader_->ReadData(
+          buffer_.get(), available,
+          base::Bind(&ScriptPump::OnReadComplete, weak_factory_.GetWeakPtr()));
+    }
+    void OnReadComplete(int result) {
+      TRACE_EVENT0("ServiceWorker", "ScriptPump::OnReadComplete");
+      LOG(ERROR) << "ScriptPump::OnReadComplete";
+      if (result < 0) {
+        LOG(ERROR) << " read error !!!!";
+        return;
+      }
+      MojoResult mojo_result = mojo::EndWriteDataRaw(stream_.get(), result);
+      if (mojo_result != MOJO_RESULT_OK) {
+        LOG(ERROR) << " EndWriteDataRaw error !!!!";
+        return;
+      }
+      buffer_ = nullptr;
+      if (result == 0) {
+        sent_all_data_ = true;
+        if (!metadata_sender_) {
+          instance_->PumpFinished(this);
+        }
+        return;
+      }
+      ReadNext();
+    }
+
+    base::WeakPtr<IsolatedWorkerInstance> instance_;
+    std::unique_ptr<ServiceWorkerResponseReader> reader_;
+    const GURL url_;
+    scoped_refptr<HttpResponseInfoIOBuffer> http_info_io_buffer_;
+    mojo::ScopedDataPipeProducerHandle stream_;
+    std::unique_ptr<MetadataSender> metadata_sender_;
+    scoped_refptr<net::IOBufferWithSize> buffer_;
+    mojo::Watcher handle_watcher_;
+    bool sent_all_data_ = false;
+    base::WeakPtrFactory<ScriptPump> weak_factory_;
+  };
+  class IsolatedWorkerHostServiceImpl
+      : public blink::mojom::IsolatedWorkerHostService {
+   public:
+    explicit IsolatedWorkerHostServiceImpl(
+        mojo::InterfaceRequest<blink::mojom::IsolatedWorkerHostService> request,
+        base::WeakPtr<ServiceWorkerVersion> version)
+        : binding_(this, std::move(request)), version_(version) {
+      LOG(ERROR) << "IsolatedWorkerHostServiceImpl" << this;
+    }
+    ~IsolatedWorkerHostServiceImpl() override {
+      LOG(ERROR) << "IsolatedWorkerHostServiceImpl" << this;
+    }
+
+    void OnThreadStarted(
+        int64_t version_id,
+        blink::mojom::IsolatedWorkerClientPtr client) override {
+      TRACE_EVENT0("ServiceWorker",
+                   "IsolatedWorkerHostServiceImpl::WorkerStarted");
+      LOG(ERROR)
+          << "IsolatedWorkerHostServiceImpl::WorkerStarted <><><><><><>< "
+          << version_id;
+      version_->OnIsolatedWorkerThreadStarted(std::move(client));
+    }
+    void OnScriptEvaluated() override {
+      TRACE_EVENT0("ServiceWorker",
+                   "IsolatedWorkerHostServiceImpl::OnScriptEvaluated");
+      LOG(ERROR)
+          << "IsolatedWorkerHostServiceImpl::OnScriptEvaluated() <><><><><><><";
+      version_->OnIsolatedWorkerScriptEvaluated();
+    }
+    void OnWorkerStarted() override {
+      TRACE_EVENT0("ServiceWorker",
+                   "IsolatedWorkerHostServiceImpl::OnWorkerStarted");
+      LOG(ERROR)
+          << "IsolatedWorkerHostServiceImpl::OnWorkerStarted() <><><><><><><";
+      version_->OnIsolatedWorkerStarted();
+    }
+
+   private:
+    mojo::StrongBinding<blink::mojom::IsolatedWorkerHostService> binding_;
+    base::WeakPtr<ServiceWorkerVersion> version_;
+  };
+  void PumpFinished(ScriptPump* pump) {
+    TRACE_EVENT0("ServiceWorker", "IsolatedWorkerInstance::PumpFinished");
+    LOG(ERROR) << "IsolatedWorkerInstance::PumpFinished";
+    pump_map_.erase(pump);
+  }
+  void PushScriptPump(int64_t resource_id,
+                      const GURL& url,
+                      std::vector<mojom::WorkerScriptPtr>* scripts) {
+    TRACE_EVENT0("ServiceWorker", "IsolatedWorkerInstance::PushScriptPump");
+    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 = 1024 * 1024;
+    mojo::DataPipe data_pipe(options);
+    CHECK(data_pipe.producer_handle.is_valid());
+
+    MojoCreateDataPipeOptions options_for_metadata;
+    options_for_metadata.struct_size = sizeof(MojoCreateDataPipeOptions);
+    options_for_metadata.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
+    options_for_metadata.element_num_bytes = 1;
+    options_for_metadata.capacity_num_bytes = 1024 * 1024;
+    mojo::DataPipe data_pipe_for_metadata(options_for_metadata);
+    CHECK(data_pipe_for_metadata.producer_handle.is_valid());
+
+    std::unique_ptr<ScriptPump> pump(
+        new ScriptPump(weak_factory_.GetWeakPtr(),
+                       context_->storage()->CreateResponseReader(resource_id),
+                       url, std::move(data_pipe.producer_handle),
+                       std::move(data_pipe_for_metadata.producer_handle)));
+    pump->StartRead();
+    ScriptPump* ptr = pump.get();
+    pump_map_[ptr] = std::move(pump);
+
+    mojom::WorkerScriptPtr script = mojom::WorkerScript::New();
+    script->url = url.spec();
+    script->data_pipe = std::move(data_pipe.consumer_handle);
+    script->meta_data_pipe = std::move(data_pipe_for_metadata.consumer_handle);
+
+    scripts->push_back(std::move(script));
+  }
+
+  void CreateHostService(
+      mojo::InterfaceRequest<blink::mojom::IsolatedWorkerHostService> request) {
+    TRACE_EVENT0("ServiceWorker", "IsolatedWorkerInstance::CreateHostService");
+    new IsolatedWorkerHostServiceImpl(std::move(request), version_);
+  }
+
+  base::WeakPtr<ServiceWorkerContextCore> context_;
+  base::WeakPtr<ServiceWorkerVersion> version_;
+  base::hash_map<ScriptPump*, std::unique_ptr<ScriptPump>> pump_map_;
+  // Isolated Worker test
+  std::unique_ptr<shell::InterfaceRegistry> isolated_worker_interface_registry_;
+  std::unique_ptr<shell::InterfaceProvider> isolated_worker_remote_interfaces_;
+  base::WeakPtrFactory<IsolatedWorkerInstance> weak_factory_;
+  DISALLOW_COPY_AND_ASSIGN(IsolatedWorkerInstance);
+};
+
 void ServiceWorkerVersion::StartWorkerInternal() {
   DCHECK_EQ(EmbeddedWorkerStatus::STOPPED, running_status());
 
@@ -1409,12 +1768,74 @@ void ServiceWorkerVersion::StartWorkerInternal() {
   params->is_installed = IsInstalled(status_);
   params->pause_after_download = pause_after_download_;
 
+  if (status_ == ServiceWorkerVersion::ACTIVATED) {
+    TRACE_EVENT_ASYNC_BEGIN1("ServiceWorker", "IsolatedWorkerInstance::Start",
+                             this, "Script", script_url().spec());
+
+    std::vector<ServiceWorkerDatabase::ResourceRecord> resources;
+    script_cache_map_.GetResources(&resources);
+    DCHECK(!resources.empty());
+    isolated_worker_.reset(new IsolatedWorkerInstance(
+        context_, weak_factory_.GetWeakPtr(), resources,
+        base::Bind(&ServiceWorkerVersion::OnIsolatedWorkerStartFailed,
+                   weak_factory_.GetWeakPtr())));
+    return;
+  }
+
   embedded_worker_->Start(
       std::move(params),
       base::Bind(&ServiceWorkerVersion::OnStartSentAndScriptEvaluated,
                  weak_factory_.GetWeakPtr()));
 }
 
+bool ServiceWorkerVersion::IsIsolatedWorker() const {
+  return !!isolated_worker_client_.get();
+}
+
+void ServiceWorkerVersion::OnIsolatedWorkerStartFailed() {
+  TRACE_EVENT_ASYNC_END1("ServiceWorker", "IsolatedWorkerInstance::Start", this,
+                         "success", false);
+  LOG(ERROR) << " OnIsolatedWorkerStartFailed";
+  isolated_worker_.reset();
+
+  std::unique_ptr<EmbeddedWorkerMsg_StartWorker_Params> params(
+      new EmbeddedWorkerMsg_StartWorker_Params());
+  params->service_worker_version_id = version_id_;
+  params->scope = scope_;
+  params->script_url = script_url_;
+  params->is_installed = IsInstalled(status_);
+  params->pause_after_download = pause_after_download_;
+  embedded_worker_->Start(
+      std::move(params),
+      base::Bind(&ServiceWorkerVersion::OnStartSentAndScriptEvaluated,
+                 weak_factory_.GetWeakPtr()));
+}
+
+void ServiceWorkerVersion::OnIsolatedWorkerThreadStarted(
+    mojo::InterfacePtr<blink::mojom::IsolatedWorkerClient> client) {
+  TRACE_EVENT_ASYNC_STEP_PAST0("ServiceWorker", "IsolatedWorkerInstance::Start",
+                               this, "OnIsolatedWorkerThreadStarted");
+  LOG(ERROR) << " OnIsolatedWorkerThreadStarted";
+  isolated_worker_client_ = std::move(client);
+}
+void ServiceWorkerVersion::OnIsolatedWorkerScriptEvaluated() {
+  TRACE_EVENT_ASYNC_STEP_PAST0("ServiceWorker", "IsolatedWorkerInstance::Start",
+                               this, "OnIsolatedWorkerScriptEvaluated");
+  LOG(ERROR) << " OnIsolatedWorkerScriptEvaluated";
+}
+void ServiceWorkerVersion::OnIsolatedWorkerStarted() {
+  TRACE_EVENT_ASYNC_END1("ServiceWorker", "IsolatedWorkerInstance::Start", this,
+                         "success", true);
+  LOG(ERROR) << " OnIsolatedWorkerStarted";
+  OnStarted();
+}
+
+EmbeddedWorkerStatus ServiceWorkerVersion::running_status() const {
+  if (IsIsolatedWorker() && isolated_worker_client_)
+    return EmbeddedWorkerStatus::RUNNING;
+  return embedded_worker_->status();
+}
+
 void ServiceWorkerVersion::StartTimeoutTimer() {
   DCHECK(!timeout_timer_.IsRunning());
 
@@ -1588,6 +2009,10 @@ void ServiceWorkerVersion::RecordStartWorkerResult(
     int trace_id,
     bool is_browser_startup_complete,
     ServiceWorkerStatusCode status) {
+  // todo
+  if (IsIsolatedWorker())
+    return;
+
   if (trace_id != kInvalidTraceId) {
     TRACE_EVENT_ASYNC_END1("ServiceWorker", "ServiceWorkerVersion::StartWorker",
                            trace_id, "Status",