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..28a7fe8566836e839c58c171bec76b53726f4353 |
--- /dev/null |
+++ b/src/heap/page-parallel-job.h |
@@ -0,0 +1,182 @@ |
+// 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; |
+ DCHECK_GE(num_tasks, 1); |
+ 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 { |
+ Item(MemoryChunk* chunk, typename JobTraits::PerPageData data, Item* next) |
+ : chunk(chunk), state(kAvailable), data(data), next(next) {} |
+ MemoryChunk* chunk; |
+ AtomicValue<ProcessingState> state; |
+ typename JobTraits::PerPageData data; |
+ Item* 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_ |