Chromium Code Reviews| Index: src/heap/page-parallel-job.h |
| diff --git a/src/heap/page-parallel-job.h b/src/heap/page-parallel-job.h |
| new file mode 100644 |
| index 0000000000000000000000000000000000000000..9808f7e22eb9835e58f9e8aa6b2fce9f02a6081e |
| --- /dev/null |
| +++ b/src/heap/page-parallel-job.h |
| @@ -0,0 +1,181 @@ |
| +// Copyright 2016 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_PAGE_PARALLEL_JOB_ |
| +#define V8_HEAP_PAGE_PARALLEL_JOB_ |
| + |
| +#include "src/allocation.h" |
| +#include "src/cancelable-task.h" |
| +#include "src/utils.h" |
| +#include "src/v8.h" |
| + |
| +namespace v8 { |
| +namespace internal { |
| + |
| +class Heap; |
| +class Isolate; |
| + |
| +// This class manages background tasks that process set of pages in parallel. |
| +// The JobTraits class needs to define: |
| +// - PerPageData type - state associated with each page. |
| +// - PerTaskData type - state associated with each task. |
| +// - static bool ProcessPageInParallel(Heap* heap, |
| +// PerTaskData task_data, |
| +// MemoryChunk* page, |
| +// PerPageData page_data) |
| +// The function should return true iff processing succeeded. |
| +// - static const bool NeedSequentialFinalization |
| +// - static void FinalizePageSequentially(Heap* heap, |
| +// bool processing_succeeded, |
| +// MemoryChunk* page, |
| +// PerPageData page_data) |
| +template <typename JobTraits> |
| +class PageParallelJob { |
| + public: |
| + PageParallelJob(Heap* heap, CancelableTaskManager* cancelable_task_manager) |
| + : heap_(heap), |
| + cancelable_task_manager_(cancelable_task_manager), |
| + items_(nullptr), |
| + num_items_(0), |
| + pending_tasks_(0) {} |
| + |
| + ~PageParallelJob() { |
| + Item* item = items_; |
| + while (item != nullptr) { |
| + Item* next = item->next; |
| + delete item; |
| + item = next; |
| + } |
| + } |
| + |
| + void AddPage(MemoryChunk* chunk, typename JobTraits::PerPageData data) { |
| + Item* item = new Item(chunk, data, items_); |
| + items_ = item; |
| + ++num_items_; |
| + } |
| + |
| + int NumberOfPages() { return num_items_; } |
| + |
| + // Runs the given number of tasks in parallel and processes the previosly |
| + // added pages. This function blocks until all tasks finish. |
| + // The callback takes the index of a task and returns data for that task. |
| + template <typename Callback> |
| + void Run(int num_tasks, Callback per_task_data_callback) { |
| + if (num_items_ == 0) return; |
|
Michael Lippautz
2016/03/09 09:48:53
Maybe add
DCHECK_GE(num_tasks, 1)
after the bail
ulan
2016/03/09 17:34:18
Done.
|
| + uint32_t task_ids[kMaxNumberOfTasks]; |
| + const int max_num_tasks = Min( |
| + kMaxNumberOfTasks, |
| + static_cast<int>( |
| + V8::GetCurrentPlatform()->NumberOfAvailableBackgroundThreads())); |
| + num_tasks = Max(1, Min(num_tasks, max_num_tasks)); |
| + int items_per_task = (num_items_ + num_tasks - 1) / num_tasks; |
| + int start_index = 0; |
| + Task* main_task = nullptr; |
| + for (int i = 0; i < num_tasks; i++, start_index += items_per_task) { |
| + if (start_index >= num_items_) { |
| + start_index -= num_items_; |
| + } |
| + Task* task = new Task(heap_, items_, num_items_, start_index, |
| + &pending_tasks_, per_task_data_callback(i)); |
| + task_ids[i] = task->id(); |
| + if (i > 0) { |
| + V8::GetCurrentPlatform()->CallOnBackgroundThread( |
| + task, v8::Platform::kShortRunningTask); |
| + } else { |
| + main_task = task; |
| + } |
| + } |
| + // Contribute on main thread. |
| + main_task->Run(); |
| + delete main_task; |
| + // Wait for background tasks. |
| + for (int i = 0; i < num_tasks; i++) { |
| + if (!cancelable_task_manager_->TryAbort(task_ids[i])) { |
| + pending_tasks_.Wait(); |
| + } |
| + } |
| + if (JobTraits::NeedSequentialFinalization) { |
| + Item* item = items_; |
| + while (item != nullptr) { |
| + bool success = (item->state.Value() == kFinished); |
| + JobTraits::FinalizePageSequentially(heap_, item->chunk, success, |
| + item->data); |
| + item = item->next; |
| + } |
| + } |
| + } |
| + |
| + private: |
| + static const int kMaxNumberOfTasks = 10; |
| + |
| + enum ProcessingState { kAvailable, kProcessing, kFinished, kFailed }; |
| + |
| + struct Item : public Malloced { |
| + MemoryChunk* chunk; |
| + AtomicValue<ProcessingState> state; |
| + typename JobTraits::PerPageData data; |
| + Item* next; |
| + Item(MemoryChunk* chunk, typename JobTraits::PerPageData data, Item* next) |
|
Michael Lippautz
2016/03/09 09:48:53
nit: move constructor to top
ulan
2016/03/09 17:34:18
Done.
|
| + : chunk(chunk), state(kAvailable), data(data), next(next) {} |
| + }; |
| + |
| + class Task : public CancelableTask { |
| + public: |
| + Task(Heap* heap, Item* items, int num_items, int start_index, |
| + base::Semaphore* on_finish, typename JobTraits::PerTaskData data) |
| + : CancelableTask(heap->isolate()), |
| + heap_(heap), |
| + items_(items), |
| + num_items_(num_items), |
| + start_index_(start_index), |
| + on_finish_(on_finish), |
| + data_(data) {} |
| + |
| + virtual ~Task() {} |
| + |
| + private: |
| + // v8::internal::CancelableTask overrides. |
| + void RunInternal() override { |
| + // Each task starts at a different index to improve parallelization. |
| + Item* current = items_; |
| + int skip = start_index_; |
| + while (skip-- > 0) { |
| + current = current->next; |
| + } |
| + for (int i = 0; i < num_items_; i++) { |
| + if (current->state.TrySetValue(kAvailable, kProcessing)) { |
| + bool success = JobTraits::ProcessPageInParallel( |
| + heap_, data_, current->chunk, current->data); |
| + current->state.SetValue(success ? kFinished : kFailed); |
| + } |
| + current = current->next; |
| + // Wrap around if needed. |
| + if (current == nullptr) { |
| + current = items_; |
| + } |
| + } |
| + on_finish_->Signal(); |
| + } |
| + |
| + Heap* heap_; |
| + Item* items_; |
| + int num_items_; |
| + int start_index_; |
| + base::Semaphore* on_finish_; |
| + typename JobTraits::PerTaskData data_; |
| + DISALLOW_COPY_AND_ASSIGN(Task); |
| + }; |
| + |
| + Heap* heap_; |
| + CancelableTaskManager* cancelable_task_manager_; |
| + Item* items_; |
| + int num_items_; |
| + base::Semaphore pending_tasks_; |
| + DISALLOW_COPY_AND_ASSIGN(PageParallelJob); |
| +}; |
| + |
| +} // namespace internal |
| +} // namespace v8 |
| + |
| +#endif // V8_HEAP_PAGE_PARALLEL_JOB_ |