[heap] Switch evacuation to ItemParallelJob

Remove PageParallelJob as it is completely replaced now. It served us
well.

Bug: chromium:651354
Change-Id: I620fa7bb0dcaf7cd3710492913c501bcb162c781
Reviewed-on: https://chromium-review.googlesource.com/544950Reviewed-by: Ulan Degenbaev <ulan@chromium.org>
Commit-Queue: Michael Lippautz <mlippautz@chromium.org>
Cr-Commit-Position: refs/heads/master@{#46159}

BUILD.gn

@@ -1636,7 +1636,6 @@ v8_source_set("v8_base") {
     "src/heap/objects-visiting-inl.h",
     "src/heap/objects-visiting.cc",
     "src/heap/objects-visiting.h",
-    "src/heap/page-parallel-job.h",
     "src/heap/remembered-set.h",
     "src/heap/scavenge-job.cc",
     "src/heap/scavenge-job.h",

src/heap/item-parallel-job.h

@@ -128,6 +128,9 @@ class ItemParallelJob {
   // Adds an item to the job. Transfers ownership to the job.
   void AddItem(Item* item) { items_.push_back(item); }
 
+  int NumberOfItems() const { return static_cast<int>(items_.size()); }
+  int NumberOfTasks() const { return static_cast<int>(tasks_.size()); }
+
   void Run() {
     DCHECK_GE(tasks_.size(), 0);
     const size_t num_tasks = tasks_.size();

src/heap/mark-compact.cc

@@ -25,7 +25,6 @@
 #include "src/heap/object-stats.h"
 #include "src/heap/objects-visiting-inl.h"
 #include "src/heap/objects-visiting.h"
-#include "src/heap/page-parallel-job.h"
 #include "src/heap/spaces-inl.h"
 #include "src/heap/workstealing-bag.h"
 #include "src/ic/ic.h"
@@ -3810,24 +3809,36 @@ bool YoungGenerationEvacuator::RawEvacuatePage(Page* page,
   return success;
 }
 
-class EvacuationJobTraits {
+class PageEvacuationItem : public ItemParallelJob::Item {
  public:
-  struct PageData {
-    MarkingState marking_state;
-  };
+  explicit PageEvacuationItem(Page* page) : page_(page) {}
+  virtual ~PageEvacuationItem() {}
+  Page* page() const { return page_; }
 
-  typedef PageData PerPageData;
-  typedef Evacuator* PerTaskData;
+ private:
+  Page* page_;
+};
+
+class PageEvacuationTask : public ItemParallelJob::Task {
+ public:
+  PageEvacuationTask(Isolate* isolate, Evacuator* evacuator)
+      : ItemParallelJob::Task(isolate), evacuator_(evacuator) {}
 
-  static void ProcessPageInParallel(Heap* heap, PerTaskData evacuator,
-                                    MemoryChunk* chunk, PerPageData) {
-    evacuator->EvacuatePage(reinterpret_cast<Page*>(chunk));
+  void RunInParallel() override {
+    PageEvacuationItem* item = nullptr;
+    while ((item = GetItem<PageEvacuationItem>()) != nullptr) {
+      evacuator_->EvacuatePage(item->page());
+      item->MarkFinished();
     }
+  };
+
+ private:
+  Evacuator* evacuator_;
 };
 
 template <class Evacuator, class Collector>
 void MarkCompactCollectorBase::CreateAndExecuteEvacuationTasks(
-    Collector* collector, PageParallelJob<EvacuationJobTraits>* job,
+    Collector* collector, ItemParallelJob* job,
     RecordMigratedSlotVisitor* record_visitor,
     MigrationObserver* migration_observer, const intptr_t live_bytes) {
   // Used for trace summary.
@@ -3843,15 +3854,16 @@ void MarkCompactCollectorBase::CreateAndExecuteEvacuationTasks(
   ProfilingMigrationObserver profiling_observer(heap());
 
   const int wanted_num_tasks =
-      NumberOfParallelCompactionTasks(job->NumberOfPages());
+      NumberOfParallelCompactionTasks(job->NumberOfItems());
   Evacuator** evacuators = new Evacuator*[wanted_num_tasks];
   for (int i = 0; i < wanted_num_tasks; i++) {
     evacuators[i] = new Evacuator(collector, record_visitor);
     if (profiling) evacuators[i]->AddObserver(&profiling_observer);
     if (migration_observer != nullptr)
       evacuators[i]->AddObserver(migration_observer);
+    job->AddTask(new PageEvacuationTask(heap()->isolate(), evacuators[i]));
   }
-  job->Run(wanted_num_tasks, [evacuators](int i) { return evacuators[i]; });
+  job->Run();
   const Address top = heap()->new_space()->top();
   for (int i = 0; i < wanted_num_tasks; i++) {
     evacuators[i]->Finalize();
@@ -3872,7 +3884,7 @@ void MarkCompactCollectorBase::CreateAndExecuteEvacuationTasks(
                  "wanted_tasks=%d tasks=%d cores=%" PRIuS
                  " live_bytes=%" V8PRIdPTR " compaction_speed=%.f\n",
                  isolate()->time_millis_since_init(),
-                 FLAG_parallel_compaction ? "yes" : "no", job->NumberOfPages(),
+                 FLAG_parallel_compaction ? "yes" : "no", job->NumberOfItems(),
                  wanted_num_tasks, job->NumberOfTasks(),
                  V8::GetCurrentPlatform()->NumberOfAvailableBackgroundThreads(),
                  live_bytes, compaction_speed);
@@ -3888,14 +3900,13 @@ bool MarkCompactCollectorBase::ShouldMovePage(Page* p, intptr_t live_bytes) {
 }
 
 void MarkCompactCollector::EvacuatePagesInParallel() {
-  PageParallelJob<EvacuationJobTraits> job(
-      heap_, heap_->isolate()->cancelable_task_manager(),
+  ItemParallelJob evacuation_job(isolate()->cancelable_task_manager(),
                                  &page_parallel_job_semaphore_);
   intptr_t live_bytes = 0;
 
   for (Page* page : old_space_evacuation_pages_) {
     live_bytes += MarkingState::Internal(page).live_bytes();
-    job.AddPage(page, {marking_state(page)});
+    evacuation_job.AddItem(new PageEvacuationItem(page));
   }
 
   for (Page* page : new_space_evacuation_pages_) {
@@ -3909,19 +3920,18 @@ void MarkCompactCollector::EvacuatePagesInParallel() {
         EvacuateNewSpacePageVisitor<NEW_TO_NEW>::Move(page);
       }
     }
-    job.AddPage(page, {marking_state(page)});
+    evacuation_job.AddItem(new PageEvacuationItem(page));
   }
-  if (job.NumberOfPages() == 0) return;
+  if (evacuation_job.NumberOfItems() == 0) return;
 
   RecordMigratedSlotVisitor record_visitor(this);
-  CreateAndExecuteEvacuationTasks<FullEvacuator>(this, &job, &record_visitor,
-                                                 nullptr, live_bytes);
+  CreateAndExecuteEvacuationTasks<FullEvacuator>(
+      this, &evacuation_job, &record_visitor, nullptr, live_bytes);
   PostProcessEvacuationCandidates();
 }
 
 void MinorMarkCompactCollector::EvacuatePagesInParallel() {
-  PageParallelJob<EvacuationJobTraits> job(
-      heap_, heap_->isolate()->cancelable_task_manager(),
+  ItemParallelJob evacuation_job(isolate()->cancelable_task_manager(),
                                  &page_parallel_job_semaphore_);
   intptr_t live_bytes = 0;
 
@@ -3936,16 +3946,16 @@ void MinorMarkCompactCollector::EvacuatePagesInParallel() {
         EvacuateNewSpacePageVisitor<NEW_TO_NEW>::Move(page);
       }
     }
-    job.AddPage(page, {marking_state(page)});
+    evacuation_job.AddItem(new PageEvacuationItem(page));
   }
-  if (job.NumberOfPages() == 0) return;
+  if (evacuation_job.NumberOfItems() == 0) return;
 
   YoungGenerationMigrationObserver observer(heap(),
                                             heap()->mark_compact_collector());
   YoungGenerationRecordMigratedSlotVisitor record_visitor(
       heap()->mark_compact_collector());
   CreateAndExecuteEvacuationTasks<YoungGenerationEvacuator>(
-      this, &job, &record_visitor, &observer, live_bytes);
+      this, &evacuation_job, &record_visitor, &observer, live_bytes);
 }
 
 class EvacuationWeakObjectRetainer : public WeakObjectRetainer {

src/heap/mark-compact.h

@@ -28,8 +28,6 @@ class MarkCompactCollector;
 class MinorMarkCompactCollector;
 class MarkingVisitor;
 class MigrationObserver;
-template <typename JobTraits>
-class PageParallelJob;
 class RecordMigratedSlotVisitor;
 class ThreadLocalTop;
 class WorkStealingBag;
@@ -304,7 +302,7 @@ class MarkCompactCollectorBase {
 
   template <class Evacuator, class Collector>
   void CreateAndExecuteEvacuationTasks(
-      Collector* collector, PageParallelJob<EvacuationJobTraits>* job,
+      Collector* collector, ItemParallelJob* job,
       RecordMigratedSlotVisitor* record_visitor,
       MigrationObserver* migration_observer, const intptr_t live_bytes);
 

src/heap/page-parallel-job.h

-// 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 void ProcessPageInParallel(Heap* heap,
-//                                     PerTaskData task_data,
-//                                     MemoryChunk* page,
-//                                     PerPageData page_data)
-template <typename JobTraits>
-class PageParallelJob {
- public:
-  // PageParallelJob cannot dynamically create a semaphore because of a bug in
-  // glibc. See http://crbug.com/609249 and
-  // https://sourceware.org/bugzilla/show_bug.cgi?id=12674.
-  // The caller must provide a semaphore with value 0 and ensure that
-  // the lifetime of the semaphore is the same as the lifetime of the Isolate.
-  // It is guaranteed that the semaphore value will be 0 after Run() call.
-  PageParallelJob(Heap* heap, CancelableTaskManager* cancelable_task_manager,
-                  base::Semaphore* semaphore)
-      : heap_(heap),
-        cancelable_task_manager_(cancelable_task_manager),
-        items_(nullptr),
-        num_items_(0),
-        num_tasks_(0),
-        pending_tasks_(semaphore) {}
-
-  ~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() const { return num_items_; }
-
-  // Returns the number of tasks that were spawned when running the job.
-  int NumberOfTasks() const { return num_tasks_; }
-
-  // Runs the given number of tasks in parallel and processes the previously
-  // 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]) !=
-          CancelableTaskManager::kTaskAborted) {
-        pending_tasks_->Wait();
-      }
-    }
-  }
-
- private:
-  static const int kMaxNumberOfTasks = 32;
-
-  enum ProcessingState { kAvailable, kProcessing, kFinished };
-
-  struct Item : public Malloced {
-    Item(MemoryChunk* chunk, typename JobTraits::PerPageData data, Item* next)
-        : chunk(chunk), state(kAvailable), data(data), next(next) {}
-    MemoryChunk* chunk;
-    base::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)) {
-          JobTraits::ProcessPageInParallel(heap_, data_, current->chunk,
-                                           current->data);
-          current->state.SetValue(kFinished);
-        }
-        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_;
-  int num_tasks_;
-  base::Semaphore* pending_tasks_;
-  DISALLOW_COPY_AND_ASSIGN(PageParallelJob);
-};
-
-}  // namespace internal
-}  // namespace v8
-
-#endif  // V8_HEAP_PAGE_PARALLEL_JOB_

src/v8.gyp

@@ -1093,7 +1093,6 @@
         'heap/objects-visiting-inl.h',
         'heap/objects-visiting.cc',
         'heap/objects-visiting.h',
-        'heap/page-parallel-job.h',
         'heap/remembered-set.h',
         'heap/scavenge-job.h',
         'heap/scavenge-job.cc',