blob: bf4e433a239edb1037b77f782ec3e0babeb22bc1 [file] [log] [blame]
// Copyright 2018 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 "third_party/blink/renderer/core/workers/experimental/thread_pool.h"
#include "services/service_manager/public/cpp/interface_provider.h"
#include "third_party/blink/public/platform/dedicated_worker_factory.mojom-blink.h"
#include "third_party/blink/renderer/core/dom/document.h"
#include "third_party/blink/renderer/core/frame/local_frame.h"
#include "third_party/blink/renderer/core/origin_trials/origin_trial_context.h"
#include "third_party/blink/renderer/core/workers/global_scope_creation_params.h"
#include "third_party/blink/renderer/core/workers/threaded_messaging_proxy_base.h"
#include "third_party/blink/renderer/core/workers/threaded_object_proxy_base.h"
namespace blink {
class ThreadPoolObjectProxy final : public ThreadedObjectProxyBase {
public:
ThreadPoolObjectProxy(ThreadPoolMessagingProxy* messaging_proxy,
ParentExecutionContextTaskRunners* task_runners)
: ThreadedObjectProxyBase(task_runners),
messaging_proxy_(messaging_proxy) {}
~ThreadPoolObjectProxy() override = default;
CrossThreadWeakPersistent<ThreadedMessagingProxyBase> MessagingProxyWeakPtr()
override {
return messaging_proxy_;
}
private:
CrossThreadWeakPersistent<ThreadPoolMessagingProxy> messaging_proxy_;
};
class ThreadPoolMessagingProxy final : public ThreadedMessagingProxyBase {
public:
explicit ThreadPoolMessagingProxy(ExecutionContext* context)
: ThreadedMessagingProxyBase(context) {
object_proxy_ = std::make_unique<ThreadPoolObjectProxy>(
this, GetParentExecutionContextTaskRunners());
}
~ThreadPoolMessagingProxy() override = default;
void StartWorker(std::unique_ptr<GlobalScopeCreationParams> creation_params) {
InitializeWorkerThread(std::move(creation_params),
WorkerBackingThreadStartupData::CreateDefault());
}
std::unique_ptr<WorkerThread> CreateWorkerThread() override {
return std::make_unique<ThreadPoolThread>(
GetExecutionContext(), *object_proxy_.get(), ThreadPoolThread::kWorker);
}
ThreadPoolThread* GetWorkerThread() const {
return static_cast<ThreadPoolThread*>(
ThreadedMessagingProxyBase::GetWorkerThread());
}
private:
std::unique_ptr<ThreadPoolObjectProxy> object_proxy_;
};
service_manager::mojom::blink::InterfaceProviderPtrInfo
ConnectToWorkerInterfaceProviderForThreadPool(
ExecutionContext* execution_context,
scoped_refptr<const SecurityOrigin> script_origin) {
// TODO(japhet): Implement a proper factory.
mojom::blink::DedicatedWorkerFactoryPtr worker_factory;
execution_context->GetInterfaceProvider()->GetInterface(&worker_factory);
service_manager::mojom::blink::InterfaceProviderPtrInfo
interface_provider_ptr;
worker_factory->CreateDedicatedWorker(
script_origin, mojo::MakeRequest(&interface_provider_ptr));
return interface_provider_ptr;
}
const char ThreadPool::kSupplementName[] = "ThreadPool";
ThreadPool* ThreadPool::From(Document& document) {
ThreadPool* thread_pool = Supplement<Document>::From<ThreadPool>(document);
if (!thread_pool) {
thread_pool = new ThreadPool(document);
Supplement<Document>::ProvideTo(document, thread_pool);
}
return thread_pool;
}
static const size_t kMaxThreadCount = 4;
ThreadPool::ThreadPool(Document& document)
: Supplement<Document>(document), ContextLifecycleObserver(&document) {}
ThreadPool::~ThreadPool() {
DCHECK(IsMainThread());
for (auto proxy : thread_proxies_) {
proxy->ParentObjectDestroyed();
}
}
ThreadPoolThread* ThreadPool::CreateNewThread() {
DCHECK(IsMainThread());
DCHECK_LT(thread_proxies_.size(), kMaxThreadCount);
base::UnguessableToken devtools_worker_token =
GetFrame() ? GetFrame()->GetDevToolsFrameToken()
: base::UnguessableToken::Create();
ExecutionContext* context = GetExecutionContext();
ThreadPoolMessagingProxy* proxy = new ThreadPoolMessagingProxy(context);
std::unique_ptr<WorkerSettings> settings =
std::make_unique<WorkerSettings>(GetFrame()->GetSettings());
// WebWorkerFetchContext is provided later in
// ThreadedMessagingProxyBase::InitializeWorkerThread().
proxy->StartWorker(std::make_unique<GlobalScopeCreationParams>(
context->Url(), mojom::ScriptType::kClassic, context->UserAgent(),
nullptr /* web_worker_fetch_context */,
context->GetContentSecurityPolicy()->Headers(),
network::mojom::ReferrerPolicy::kDefault, context->GetSecurityOrigin(),
context->IsSecureContext(), context->GetHttpsState(),
WorkerClients::Create(), context->GetSecurityContext().AddressSpace(),
OriginTrialContext::GetTokens(context).get(), devtools_worker_token,
std::move(settings), kV8CacheOptionsDefault,
nullptr /* worklet_module_responses_map */,
ConnectToWorkerInterfaceProviderForThreadPool(
context, context->GetSecurityOrigin())));
thread_proxies_.insert(proxy);
return proxy->GetWorkerThread();
}
ThreadPoolThread* ThreadPool::GetLeastBusyThread() {
DCHECK(IsMainThread());
ThreadPoolThread* least_busy_thread = nullptr;
size_t lowest_task_count = std::numeric_limits<std::size_t>::max();
for (auto proxy : thread_proxies_) {
ThreadPoolThread* current_thread = proxy->GetWorkerThread();
size_t current_task_count = current_thread->GetTasksInProgressCount();
// If there's an idle thread, use it.
if (current_task_count == 0)
return current_thread;
if (current_task_count < lowest_task_count) {
least_busy_thread = current_thread;
lowest_task_count = current_task_count;
}
}
if (thread_proxies_.size() == kMaxThreadCount)
return least_busy_thread;
return CreateNewThread();
}
void ThreadPool::ContextDestroyed(ExecutionContext*) {
DCHECK(IsMainThread());
DCHECK(GetExecutionContext()->IsContextThread());
for (auto proxy : thread_proxies_) {
proxy->TerminateGlobalScope();
}
}
void ThreadPool::Trace(blink::Visitor* visitor) {
Supplement<Document>::Trace(visitor);
ContextLifecycleObserver::Trace(visitor);
visitor->Trace(thread_proxies_);
}
} // namespace blink