[heap] Keep concurrent marking tasks running until marking is completed.

BUG=chromium:694255

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

src/heap/concurrent-marking.cc

@@ -284,7 +284,9 @@ ConcurrentMarking::ConcurrentMarking(Heap* heap, MarkingWorklist* shared,
       shared_(shared),
       bailout_(bailout),
       pending_task_semaphore_(0),
-      pending_task_count_(0) {
+      pending_task_count_(0),
+      waiting_task_count_(0),
+      task_exit_requested_(false) {
 // The runtime flag should be set only if the compile time flag was set.
 #ifndef V8_CONCURRENT_MARKING
   CHECK(!FLAG_concurrent_marking);
@@ -298,9 +300,15 @@ void ConcurrentMarking::Run(int task_id, base::Mutex* lock) {
   {
     TimedScope scope(&time_ms);
     while (true) {
+      bool worklist_is_empty;
+      {
+        // Process the worklist.
         base::LockGuard<base::Mutex> guard(lock);
         HeapObject* object;
-      if (!shared_->Pop(task_id, &object)) break;
+        if (!shared_->Pop(task_id, &object)) {
+          worklist_is_empty = true;
+        } else {
+          worklist_is_empty = false;
           Address new_space_top = heap_->new_space()->original_top();
           Address new_space_limit = heap_->new_space()->original_limit();
           Address addr = object->address();
@@ -311,6 +319,16 @@ void ConcurrentMarking::Run(int task_id, base::Mutex* lock) {
             bytes_marked += visitor.Visit(map, object);
           }
         }
+      }
+      if (worklist_is_empty) {
+        base::LockGuard<base::Mutex> guard(&wait_lock_);
+        if (task_exit_requested_) break;
+        waiting_task_count_++;
+        wait_condition_.Wait(&wait_lock_);
+        waiting_task_count_--;
+      }
+    }
+
     {
       // Take the lock to synchronize with worklist update after
       // young generation GC.
@@ -331,6 +349,8 @@ void ConcurrentMarking::Start() {
     heap_->isolate()->PrintWithTimestamp("Starting concurrent marking\n");
   }
   pending_task_count_ = kTasks;
+  waiting_task_count_ = 0;
+  task_exit_requested_ = false;
   for (int i = 0; i < kTasks; i++) {
     int task_id = i + 1;
     V8::GetCurrentPlatform()->CallOnBackgroundThread(
@@ -342,12 +362,35 @@ void ConcurrentMarking::Start() {
 
 void ConcurrentMarking::EnsureCompleted() {
   if (!FLAG_concurrent_marking) return;
+  RequestTaskExit();
   while (pending_task_count_ > 0) {
     pending_task_semaphore_.Wait();
     pending_task_count_--;
   }
 }
 
+void ConcurrentMarking::RequestTaskExit() {
+  if (!FLAG_concurrent_marking) return;
+  base::LockGuard<base::Mutex> guard(&wait_lock_);
+  task_exit_requested_ = true;
+  wait_condition_.NotifyAll();
+}
+
+void ConcurrentMarking::NotifyWaitingTasks() {
+  if (!FLAG_concurrent_marking) return;
+  base::LockGuard<base::Mutex> guard(&wait_lock_);
+  if (waiting_task_count_ > 0) {
+    if (!shared_->IsGlobalPoolEmpty()) {
+      wait_condition_.NotifyAll();
+    }
+  }
+}
+
+bool ConcurrentMarking::AllTasksWaitingForTesting() {
+  base::LockGuard<base::Mutex> guard(&wait_lock_);
+  return waiting_task_count_ == kTasks;
+}
+
 ConcurrentMarking::PauseScope::PauseScope(ConcurrentMarking* concurrent_marking)
     : concurrent_marking_(concurrent_marking) {
   if (!FLAG_concurrent_marking) return;

src/heap/concurrent-marking.h

@@ -40,6 +40,12 @@ class ConcurrentMarking {
   void Start();
   bool IsRunning() { return pending_task_count_ > 0; }
   void EnsureCompleted();
+  // Wake up waiting tasks if the shared global pool is not empty.
+  void NotifyWaitingTasks();
+  // Set task exit request flag and wake up waiting tasks.
+  void RequestTaskExit();
+  // Returns true if all tasks are waiting. For testing only.
+  bool AllTasksWaitingForTesting();
 
  private:
   struct TaskLock {
@@ -52,8 +58,16 @@ class ConcurrentMarking {
   MarkingWorklist* shared_;
   MarkingWorklist* bailout_;
   TaskLock task_lock_[kTasks];
+  // Used by the main thread to wait for tasks to exit.
   base::Semaphore pending_task_semaphore_;
   int pending_task_count_;
+  // Used by the tasks to wait for
+  // - more work from the main thread
+  // - or for the exit request.
+  base::Mutex wait_lock_;
+  base::ConditionVariable wait_condition_;
+  int waiting_task_count_;
+  bool task_exit_requested_;
 };
 
 }  // namespace internal

src/heap/incremental-marking.cc

@@ -592,7 +592,7 @@ void IncrementalMarking::StartMarking() {
   heap_->IterateStrongRoots(&visitor, VISIT_ONLY_STRONG);
 
   if (FLAG_concurrent_marking) {
-    heap_->concurrent_marking()->Start();
+    heap()->concurrent_marking()->Start();
   }
 
   // Ready to start incremental marking.
@@ -1056,6 +1056,9 @@ void IncrementalMarking::MarkingComplete(CompletionAction action) {
   // that shouldn't make us do a scavenge and keep being incremental, so we set
   // the should-hurry flag to indicate that there can't be much work left to do.
   set_should_hurry(true);
+  if (FLAG_concurrent_marking) {
+    heap()->concurrent_marking()->RequestTaskExit();
+  }
   if (FLAG_trace_incremental_marking) {
     heap()->isolate()->PrintWithTimestamp(
         "[IncrementalMarking] Complete (normal).\n");
@@ -1236,6 +1239,9 @@ size_t IncrementalMarking::Step(size_t bytes_to_process,
         heap_->local_embedder_heap_tracer()->NotifyV8MarkingWorklistWasEmpty();
       }
     }
+    if (FLAG_concurrent_marking) {
+      heap()->concurrent_marking()->NotifyWaitingTasks();
+    }
   }
 
   double end = heap_->MonotonicallyIncreasingTimeInMs();

test/cctest/heap/test-concurrent-marking.cc

@@ -15,23 +15,46 @@
 
 namespace v8 {
 namespace internal {
+
+void PublishSegment(ConcurrentMarking::MarkingWorklist* worklist,
+                    HeapObject* object) {
+  for (int i = 0; i <= ConcurrentMarking::MarkingWorklist::kSegmentCapacity;
+       i++) {
+    worklist->Push(0, object);
+  }
+  CHECK(worklist->Pop(0, &object));
+}
+
 TEST(ConcurrentMarking) {
   if (!i::FLAG_concurrent_marking) return;
   CcTest::InitializeVM();
   Heap* heap = CcTest::heap();
   ConcurrentMarking::MarkingWorklist shared, bailout;
-  for (int i = 0; i <= ConcurrentMarking::MarkingWorklist::kSegmentCapacity;
-       i++) {
-    shared.Push(0, heap->undefined_value());
-  }
-  HeapObject* object;
-  CHECK(shared.Pop(0, &object));
   ConcurrentMarking* concurrent_marking =
       new ConcurrentMarking(heap, &shared, &bailout);
+  PublishSegment(&shared, heap->undefined_value());
   concurrent_marking->Start();
   concurrent_marking->EnsureCompleted();
   delete concurrent_marking;
 }
 
+TEST(ConcurrentMarkingWaiting) {
+  if (!i::FLAG_concurrent_marking) return;
+  CcTest::InitializeVM();
+  Heap* heap = CcTest::heap();
+  ConcurrentMarking::MarkingWorklist shared, bailout;
+  ConcurrentMarking* concurrent_marking =
+      new ConcurrentMarking(heap, &shared, &bailout);
+  PublishSegment(&shared, heap->undefined_value());
+  concurrent_marking->Start();
+  while (!concurrent_marking->AllTasksWaitingForTesting()) {
+    // Busy wait.
+  }
+  PublishSegment(&shared, heap->undefined_value());
+  concurrent_marking->NotifyWaitingTasks();
+  concurrent_marking->EnsureCompleted();
+  delete concurrent_marking;
+}
+
 }  // namespace internal
 }  // namespace v8