AuroraMiddleware/v8
1
0
Fork 0
Code Issues 2 Pull Requests Releases Wiki Activity

cppgc: Provide jobs support through DefaultPlatform and TestPlatform

Browse Source 
This CL extends cppgc::DefaultPlatform and TestPlatform to emulate
jobs using std::thread and v8::base::Thread respectively.
Jobs using these platform do not yield unless the job as been
cancelled. Additionally, the job priority is ignored.

Bug: chromium:1056170
Change-Id: I72db1eef410d2be3d3e5ea7d4ece9e5584a451f2
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/2416378
Commit-Queue: Omer Katz <omerkatz@chromium.org>
Reviewed-by: Anton Bikineev <bikineev@chromium.org>
Reviewed-by: Michael Lippautz <mlippautz@chromium.org>
Cr-Commit-Position: refs/heads/master@{#70139}
This commit is contained in:
Omer Katz 2020-09-25 16:45:28 +02:00 committed by Commit Bot
parent 4822d3b22a
commit 22c0fc8f2e
9 changed files with 314 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
BUILD.gn
View File

@ -4331,6 +4331,7 @@ v8_source_set("cppgc_base") {
"include/cppgc/visitor.h",
"include/v8config.h",
"src/heap/cppgc/allocation.cc",
"src/heap/cppgc/default-job.h",
"src/heap/cppgc/default-platform.cc",
"src/heap/cppgc/free-list.cc",
"src/heap/cppgc/free-list.h",

9
include/cppgc/default-platform.h
View File

@ -6,7 +6,6 @@
#define INCLUDE_CPPGC_DEFAULT_PLATFORM_H_
#include <memory>
#include <thread> // NOLINT(build/c++11)
#include <vector>
#include "cppgc/platform.h"
@ -14,6 +13,10 @@
namespace cppgc {
namespace internal {
class DefaultJob;
} // namespace internal
/**
* Default task runner implementation. Keep posted tasks in a list that can be
* processed by calling RunSingleTask() or RunUntilIdle().
@ -61,6 +64,8 @@ class V8_EXPORT DefaultPlatform final : public Platform {
std::shared_ptr<cppgc::TaskRunner> GetForegroundTaskRunner() final;
// DefaultPlatform does not support job priorities. All jobs would be
// assigned the same priority regardless of the cppgc::TaskPriority parameter.
std::unique_ptr<cppgc::JobHandle> PostJob(
cppgc::TaskPriority priority,
std::unique_ptr<cppgc::JobTask> job_task) final;
@ -71,7 +76,7 @@ class V8_EXPORT DefaultPlatform final : public Platform {
private:
std::unique_ptr<PageAllocator> page_allocator_;
std::shared_ptr<DefaultTaskRunner> foreground_task_runner_;
std::vector<std::shared_ptr<std::thread>> job_threads_;
std::vector<std::shared_ptr<internal::DefaultJob>> jobs_;
};
} // namespace cppgc

186
src/heap/cppgc/default-job.h Normal file
View File

@ -0,0 +1,186 @@
// Copyright 2020 the V8 project 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 V8_HEAP_CPPGC_DEFAULT_JOB_H_
#define V8_HEAP_CPPGC_DEFAULT_JOB_H_
#include <atomic>
#include <map>
#include <memory>
#include <unordered_set>
#include <vector>
#include "include/cppgc/platform.h"
#include "src/base/logging.h"
#include "src/base/platform/mutex.h"
namespace cppgc {
namespace internal {
template <typename Job>
class DefaultJobFactory {
public:
static std::shared_ptr<Job> Create(std::unique_ptr<cppgc::JobTask> job_task) {
std::shared_ptr<Job> job =
std::make_shared<Job>(typename Job::Key(), std::move(job_task));
job->NotifyConcurrencyIncrease();
return job;
}
};
template <typename Thread>
class DefaultJobImpl {
public:
class JobDelegate;
class JobHandle;
class Key {
private:
Key() {}
template <typename Job>
friend class DefaultJobFactory;
};
DefaultJobImpl(Key, std::unique_ptr<cppgc::JobTask> job_task)
: job_task_(std::move(job_task)) {}
~DefaultJobImpl() {
Cancel();
DCHECK_EQ(0, active_threads_.load(std::memory_order_relaxed));
}
void NotifyConcurrencyIncrease();
void Join() {
for (std::shared_ptr<Thread>& thread : job_threads_) thread->Join();
job_threads_.clear();
can_run_.store(false, std::memory_order_relaxed);
}
void Cancel() {
can_run_.store(false, std::memory_order_relaxed);
Join();
}
bool IsCompleted() const { return !IsRunning(); }
bool IsRunning() const {
uint8_t active_threads = active_threads_.load(std::memory_order_relaxed);
return (active_threads + job_task_->GetMaxConcurrency(active_threads)) > 0;
}
bool CanRun() const { return can_run_.load(std::memory_order_relaxed); }
protected:
virtual std::shared_ptr<Thread> CreateThread(DefaultJobImpl*) = 0;
void RunJobTask() {
DCHECK_NOT_NULL(job_task_);
NotifyJobThreadStart();
JobDelegate delegate(this);
job_task_->Run(&delegate);
NotifyJobThreadEnd();
}
void NotifyJobThreadStart() {
active_threads_.fetch_add(1, std::memory_order_relaxed);
}
void NotifyJobThreadEnd() {
active_threads_.fetch_sub(1, std::memory_order_relaxed);
}
void GuaranteeAvailableIds(uint8_t max_threads) {
if (max_threads <= highest_thread_count_) return;
v8::base::MutexGuard guard(&ids_lock_);
while (highest_thread_count_ < max_threads) {
available_ids_.push_back(++highest_thread_count_);
}
}
std::unique_ptr<cppgc::JobTask> job_task_;
std::vector<std::shared_ptr<Thread>> job_threads_;
std::atomic_bool can_run_{true};
std::atomic_uint8_t active_threads_{0};
// Task id management.
v8::base::Mutex ids_lock_;
std::vector<uint8_t> available_ids_;
uint8_t highest_thread_count_ = -1;
};
template <typename Thread>
class DefaultJobImpl<Thread>::JobDelegate final : public cppgc::JobDelegate {
public:
explicit JobDelegate(DefaultJobImpl* job) : job_(job) {}
~JobDelegate() { ReleaseTaskId(); }
bool ShouldYield() override { return !job_->CanRun(); }
void NotifyConcurrencyIncrease() override {
job_->NotifyConcurrencyIncrease();
}
uint8_t GetTaskId() override {
AcquireTaskId();
return job_thread_id_;
}
private:
void AcquireTaskId() {
if (job_thread_id_ != kInvalidTaskId) return;
v8::base::MutexGuard guard(&job_->ids_lock_);
job_thread_id_ = job_->available_ids_.back();
DCHECK_NE(kInvalidTaskId, job_thread_id_);
job_->available_ids_.pop_back();
}
void ReleaseTaskId() {
if (job_thread_id_ == kInvalidTaskId) return;
v8::base::MutexGuard guard(&job_->ids_lock_);
job_->available_ids_.push_back(job_thread_id_);
}
DefaultJobImpl* const job_;
static constexpr uint8_t kInvalidTaskId = std::numeric_limits<uint8_t>::max();
uint8_t job_thread_id_ = kInvalidTaskId;
};
template <typename Thread>
void DefaultJobImpl<Thread>::NotifyConcurrencyIncrease() {
DCHECK(CanRun());
static const size_t kMaxThreads = Thread::GetMaxSupportedConcurrency();
uint8_t current_active_threads =
active_threads_.load(std::memory_order_relaxed);
size_t max_threads = std::min(
kMaxThreads, job_task_->GetMaxConcurrency(current_active_threads));
if (current_active_threads >= max_threads) return;
DCHECK_LT(max_threads, std::numeric_limits<uint8_t>::max());
GuaranteeAvailableIds(max_threads);
for (uint8_t new_threads = max_threads - current_active_threads;
new_threads > 0; --new_threads) {
std::shared_ptr<Thread> thread = CreateThread(this);
job_threads_.push_back(thread);
}
}
template <typename Thread>
class DefaultJobImpl<Thread>::JobHandle final : public cppgc::JobHandle {
public:
explicit JobHandle(std::shared_ptr<DefaultJobImpl> job)
: job_(std::move(job)) {
DCHECK_NOT_NULL(job_);
}
void NotifyConcurrencyIncrease() override {
job_->NotifyConcurrencyIncrease();
}
void Join() override { job_->Join(); }
void Cancel() override { job_->Cancel(); }
bool IsCompleted() override { return job_->IsCompleted(); }
bool IsRunning() override { return job_->IsRunning(); }
private:
std::shared_ptr<DefaultJobImpl> job_;
};
} // namespace internal
} // namespace cppgc
#endif // V8_HEAP_CPPGC_DEFAULT_JOB_H_

66
src/heap/cppgc/default-platform.cc
View File

@ -9,30 +9,44 @@
#include "src/base/logging.h"
#include "src/base/page-allocator.h"
#include "src/base/sys-info.h"
#include "src/heap/cppgc/default-job.h"
namespace cppgc {
namespace internal {
// Default implementation of Jobs based on std::thread.
namespace {
// Simple implementation of JobTask based on std::thread.
class DefaultJobHandle : public JobHandle {
class DefaultJobThread final : private std::thread {
public:
explicit DefaultJobHandle(std::shared_ptr<std::thread> thread)
: thread_(std::move(thread)) {}
template <typename Function>
explicit DefaultJobThread(Function function)
: std::thread(std::move(function)) {}
~DefaultJobThread() { DCHECK(!joinable()); }
void NotifyConcurrencyIncrease() override {}
void Join() override {
if (thread_->joinable()) thread_->join();
void Join() { join(); }
static size_t GetMaxSupportedConcurrency() {
return v8::base::SysInfo::NumberOfProcessors() - 1;
}
void Cancel() override { Join(); }
bool IsCompleted() override { return !IsRunning(); }
bool IsRunning() override { return thread_->joinable(); }
};
} // namespace
private:
std::shared_ptr<std::thread> thread_;
class DefaultJob final : public DefaultJobImpl<DefaultJobThread> {
public:
DefaultJob(Key key, std::unique_ptr<cppgc::JobTask> job_task)
: DefaultJobImpl(key, std::move(job_task)) {}
std::shared_ptr<DefaultJobThread> CreateThread(DefaultJobImpl* job) final {
return std::make_shared<DefaultJobThread>([job = this] {
DCHECK_NOT_NULL(job);
job->RunJobTask();
});
}
};
} // namespace
} // namespace internal
void DefaultTaskRunner::PostTask(std::unique_ptr<cppgc::Task> task) {
tasks_.push_back(std::move(task));
@ -108,19 +122,11 @@ std::shared_ptr<cppgc::TaskRunner> DefaultPlatform::GetForegroundTaskRunner() {
std::unique_ptr<cppgc::JobHandle> DefaultPlatform::PostJob(
cppgc::TaskPriority priority, std::unique_ptr<cppgc::JobTask> job_task) {
auto thread = std::make_shared<std::thread>([task = std::move(job_task)] {
class SimpleDelegate final : public cppgc::JobDelegate {
public:
bool ShouldYield() override { return false; }
void NotifyConcurrencyIncrease() override {}
uint8_t GetTaskId() override { return 0; }
bool IsJoiningThread() const override { return false; }
} delegate;
if (task) task->Run(&delegate);
});
job_threads_.push_back(thread);
return std::make_unique<DefaultJobHandle>(std::move(thread));
std::shared_ptr<internal::DefaultJob> job =
internal::DefaultJobFactory<internal::DefaultJob>::Create(
std::move(job_task));
jobs_.push_back(job);
return std::make_unique<internal::DefaultJob::JobHandle>(std::move(job));
}
void DefaultPlatform::WaitAllForegroundTasks() {
@ -128,10 +134,10 @@ void DefaultPlatform::WaitAllForegroundTasks() {
}
void DefaultPlatform::WaitAllBackgroundTasks() {
for (auto& thread : job_threads_) {
thread->join();
for (auto& job : jobs_) {
job->Join();
}
job_threads_.clear();
jobs_.clear();
}
} // namespace cppgc

2
src/heap/cppgc/marker.cc
View File

@ -146,7 +146,7 @@ MarkerBase::IncrementalMarkingTask::IncrementalMarkingTask(
// static
MarkerBase::IncrementalMarkingTask::Handle
MarkerBase::IncrementalMarkingTask::Post(v8::TaskRunner* runner,
MarkerBase::IncrementalMarkingTask::Post(cppgc::TaskRunner* runner,
MarkerBase* marker) {
// Incremental GC is possible only via the GCInvoker, so getting here
// guarantees that either non-nestable tasks or conservative stack

6
src/heap/cppgc/marker.h
View File

@ -99,13 +99,13 @@ class V8_EXPORT_PRIVATE MarkerBase {
bool IncrementalMarkingStepForTesting(MarkingConfig::StackState);
class IncrementalMarkingTask final : public v8::Task {
class IncrementalMarkingTask final : public cppgc::Task {
public:
using Handle = SingleThreadedHandle;
IncrementalMarkingTask(MarkerBase*, MarkingConfig::StackState);
static Handle Post(v8::TaskRunner*, MarkerBase*);
static Handle Post(cppgc::TaskRunner*, MarkerBase*);
private:
void Run() final;
@ -156,7 +156,7 @@ class V8_EXPORT_PRIVATE MarkerBase {
MarkingConfig config_ = MarkingConfig::Default();
cppgc::Platform* platform_;
std::shared_ptr<v8::TaskRunner> foreground_task_runner_;
std::shared_ptr<cppgc::TaskRunner> foreground_task_runner_;
IncrementalMarkingTask::Handle incremental_marking_handle_;
MarkingWorklists marking_worklists_;

14
src/heap/cppgc/sweeper.cc
View File

@ -384,14 +384,14 @@ class MutatorThreadSweeper final : private HeapVisitor<MutatorThreadSweeper> {
cppgc::Platform* platform_;
};
class ConcurrentSweepTask final : public v8::JobTask,
class ConcurrentSweepTask final : public cppgc::JobTask,
private HeapVisitor<ConcurrentSweepTask> {
friend class HeapVisitor<ConcurrentSweepTask>;
public:
explicit ConcurrentSweepTask(SpaceStates* states) : states_(states) {}
void Run(v8::JobDelegate* delegate) final {
void Run(cppgc::JobDelegate* delegate) final {
for (SpaceState& state : *states_) {
while (auto page = state.unswept_pages.Pop()) {
Traverse(*page);
@ -525,14 +525,14 @@ class Sweeper::SweeperImpl final {
}
private:
class IncrementalSweepTask : public v8::IdleTask {
class IncrementalSweepTask : public cppgc::IdleTask {
public:
using Handle = SingleThreadedHandle;
explicit IncrementalSweepTask(SweeperImpl* sweeper)
: sweeper_(sweeper), handle_(Handle::NonEmptyTag{}) {}
static Handle Post(SweeperImpl* sweeper, v8::TaskRunner* runner) {
static Handle Post(SweeperImpl* sweeper, cppgc::TaskRunner* runner) {
auto task = std::make_unique<IncrementalSweepTask>(sweeper);
auto handle = task->GetHandle();
runner->PostIdleTask(std::move(task));
@ -573,7 +573,7 @@ class Sweeper::SweeperImpl final {
if (!platform_) return;
concurrent_sweeper_handle_ = platform_->PostJob(
v8::TaskPriority::kUserVisible,
cppgc::TaskPriority::kUserVisible,
std::make_unique<ConcurrentSweepTask>(&space_states_));
}
@ -593,9 +593,9 @@ class Sweeper::SweeperImpl final {
StatsCollector* stats_collector_;
SpaceStates space_states_;
cppgc::Platform* platform_;
std::shared_ptr<v8::TaskRunner> foreground_task_runner_;
std::shared_ptr<cppgc::TaskRunner> foreground_task_runner_;
IncrementalSweepTask::Handle incremental_sweeper_handle_;
std::unique_ptr<v8::JobHandle> concurrent_sweeper_handle_;
std::unique_ptr<cppgc::JobHandle> concurrent_sweeper_handle_;
bool is_in_progress_ = false;
};

89
test/unittests/heap/cppgc/test-platform.cc
View File

@ -6,29 +6,71 @@
#include "src/base/platform/platform.h"
#include "src/base/platform/time.h"
#include "src/heap/cppgc/default-job.h"
namespace cppgc {
namespace internal {
namespace testing {
void TestTaskRunner::PostTask(std::unique_ptr<v8::Task> task) {
namespace {
class TestJobThread final : public v8::base::Thread {
public:
using id = uint8_t;
explicit TestJobThread(TestJob* job) : Thread(Options("job")), job_(job) {}
void Run() final;
static size_t GetMaxSupportedConcurrency() { return 4u; }
private:
TestJob* const job_;
};
} // namespace
// Default implementation of Jobs based on std::thread.
class TestJob final : public DefaultJobImpl<TestJobThread> {
public:
explicit TestJob(Key key, std::unique_ptr<cppgc::JobTask> job_task)
: DefaultJobImpl(key, std::move(job_task)) {}
std::shared_ptr<TestJobThread> CreateThread(DefaultJobImpl* job) final {
std::shared_ptr<TestJobThread> thread =
std::make_shared<TestJobThread>(this);
const bool thread_started = thread->Start();
USE(thread_started);
DCHECK(thread_started);
return thread;
}
private:
friend class TestJobThread;
};
void TestJobThread::Run() {
DCHECK_NOT_NULL(job_);
job_->RunJobTask();
}
void TestTaskRunner::PostTask(std::unique_ptr<cppgc::Task> task) {
tasks_.push_back(std::move(task));
}
void TestTaskRunner::PostNonNestableTask(std::unique_ptr<v8::Task> task) {
void TestTaskRunner::PostNonNestableTask(std::unique_ptr<cppgc::Task> task) {
PostTask(std::move(task));
}
void TestTaskRunner::PostDelayedTask(std::unique_ptr<v8::Task> task, double) {
void TestTaskRunner::PostDelayedTask(std::unique_ptr<cppgc::Task> task,
double) {
PostTask(std::move(task));
}
void TestTaskRunner::PostNonNestableDelayedTask(std::unique_ptr<v8::Task> task,
double) {
void TestTaskRunner::PostNonNestableDelayedTask(
std::unique_ptr<cppgc::Task> task, double) {
PostTask(std::move(task));
}
void TestTaskRunner::PostIdleTask(std::unique_ptr<v8::IdleTask> task) {
void TestTaskRunner::PostIdleTask(std::unique_ptr<cppgc::IdleTask> task) {
idle_tasks_.push_back(std::move(task));
}
@ -62,36 +104,19 @@ void TestTaskRunner::RunUntilIdle() {
idle_tasks_.clear();
}
class TestPlatform::TestJobHandle : public v8::JobHandle {
public:
explicit TestJobHandle(const std::shared_ptr<JobThread>& thread)
: thread_(thread) {
const bool success = thread_->Start();
USE(success);
}
void NotifyConcurrencyIncrease() override {}
void Join() override { thread_->Join(); }
void Cancel() override { Join(); }
bool IsCompleted() override { return true; }
bool IsRunning() override { return true; }
private:
std::shared_ptr<JobThread> thread_;
};
TestPlatform::TestPlatform()
: foreground_task_runner_(std::make_unique<TestTaskRunner>()) {}
TestPlatform::~TestPlatform() V8_NOEXCEPT { WaitAllBackgroundTasks(); }
std::unique_ptr<v8::JobHandle> TestPlatform::PostJob(
v8::TaskPriority, std::unique_ptr<v8::JobTask> job_task) {
std::unique_ptr<cppgc::JobHandle> TestPlatform::PostJob(
cppgc::TaskPriority, std::unique_ptr<cppgc::JobTask> job_task) {
if (AreBackgroundTasksDisabled()) return {};
auto thread = std::make_shared<JobThread>(std::move(job_task));
job_threads_.push_back(thread);
return std::make_unique<TestJobHandle>(std::move(thread));
std::shared_ptr<TestJob> job =
DefaultJobFactory<TestJob>::Create(std::move(job_task));
jobs_.push_back(job);
return std::make_unique<TestJob::JobHandle>(std::move(job));
}
double TestPlatform::MonotonicallyIncreasingTime() {
@ -104,10 +129,10 @@ void TestPlatform::WaitAllForegroundTasks() {
}
void TestPlatform::WaitAllBackgroundTasks() {
for (auto& thread : job_threads_) {
thread->Join();
for (auto& job : jobs_) {
job->Join();
}
job_threads_.clear();
jobs_.clear();
}
TestPlatform::DisableBackgroundTasksScope::DisableBackgroundTasksScope(

64
test/unittests/heap/cppgc/test-platform.h
View File

@ -16,20 +16,22 @@ namespace cppgc {
namespace internal {
namespace testing {
class TestTaskRunner : public v8::TaskRunner {
class TestJob;
class TestTaskRunner : public cppgc::TaskRunner {
public:
void PostTask(std::unique_ptr<v8::Task> task) override;
void PostDelayedTask(std::unique_ptr<v8::Task> task, double) override;
void PostTask(std::unique_ptr<cppgc::Task> task) override;
void PostDelayedTask(std::unique_ptr<cppgc::Task> task, double) override;
bool NonNestableTasksEnabled() const override { return true; }
void PostNonNestableTask(std::unique_ptr<v8::Task> task) override;
void PostNonNestableTask(std::unique_ptr<cppgc::Task> task) override;
bool NonNestableDelayedTasksEnabled() const override { return true; }
void PostNonNestableDelayedTask(std::unique_ptr<v8::Task> task,
void PostNonNestableDelayedTask(std::unique_ptr<cppgc::Task> task,
double) override;
bool IdleTasksEnabled() override { return true; }
void PostIdleTask(std::unique_ptr<v8::IdleTask> task) override;
void PostIdleTask(std::unique_ptr<cppgc::IdleTask> task) override;
bool RunSingleTask();
bool RunSingleIdleTask(double duration_in_seconds);
@ -37,8 +39,8 @@ class TestTaskRunner : public v8::TaskRunner {
void RunUntilIdle();
private:
std::vector<std::unique_ptr<v8::Task>> tasks_;
std::vector<std::unique_ptr<v8::IdleTask>> idle_tasks_;
std::vector<std::unique_ptr<cppgc::Task>> tasks_;
std::vector<std::unique_ptr<cppgc::IdleTask>> idle_tasks_;
};
class TestPlatform : public Platform {
@ -57,12 +59,14 @@ class TestPlatform : public Platform {
PageAllocator* GetPageAllocator() override { return &page_allocator_; }
std::shared_ptr<v8::TaskRunner> GetForegroundTaskRunner() override {
std::shared_ptr<cppgc::TaskRunner> GetForegroundTaskRunner() override {
return foreground_task_runner_;
}
std::unique_ptr<v8::JobHandle> PostJob(
v8::TaskPriority, std::unique_ptr<v8::JobTask> job_task) override;
// TestPlatform does not support job priorities. All jobs would be assigned
// the same priority regardless of the cppgc::TaskPriority parameter.
std::unique_ptr<cppgc::JobHandle> PostJob(
cppgc::TaskPriority, std::unique_ptr<cppgc::JobTask> job_task) override;
double MonotonicallyIncreasingTime() override;
@ -70,49 +74,13 @@ class TestPlatform : public Platform {
void WaitAllBackgroundTasks();
private:
class TestJobHandle;
class WorkerThread : public v8::base::Thread {
public:
explicit WorkerThread(std::unique_ptr<v8::Task> task)
: Thread(Options("worker")), task_(std::move(task)) {}
void Run() override {
if (task_) std::move(task_)->Run();
}
private:
std::unique_ptr<v8::Task> task_;
};
class JobThread : public v8::base::Thread {
public:
explicit JobThread(std::unique_ptr<v8::JobTask> task)
: Thread(Options("job")), task_(std::move(task)) {}
void Run() override {
class JobDelegate : public v8::JobDelegate {
public:
bool ShouldYield() override { return false; }
void NotifyConcurrencyIncrease() override {}
uint8_t GetTaskId() override { return 0; }
bool IsJoiningThread() const override { return false; }
} delegate;
if (task_) task_->Run(&delegate);
}
private:
std::unique_ptr<v8::JobTask> task_;
};
bool AreBackgroundTasksDisabled() const {
return disabled_background_tasks_ > 0;
}
v8::base::PageAllocator page_allocator_;
std::shared_ptr<TestTaskRunner> foreground_task_runner_;
std::vector<std::shared_ptr<JobThread>> job_threads_;
std::vector<std::shared_ptr<TestJob>> jobs_;
size_t disabled_background_tasks_ = 0;
};