VM: Fix WeakProperty processing during parallel marking.

runtime/vm/gc_marker.cc

Index: runtime/vm/gc_marker.cc
diff --git a/runtime/vm/gc_marker.cc b/runtime/vm/gc_marker.cc
index 069f34175990a492b4c75280f2062e1e58ad0153..5004920b71cb1dd6a43beca1a9674a0de79fc410 100644
--- a/runtime/vm/gc_marker.cc
+++ b/runtime/vm/gc_marker.cc
@@ -177,12 +177,43 @@ class MarkingVisitorBase : public ObjectPointerVisitor {
     return class_stats_size_[class_id];
   }
 
-  // Returns true if some non-zero amount of work was performed.
-  bool DrainMarkingStack() {
+  bool ProcessPendingWeakProperties() {
+    bool marked = false;
+    RawWeakProperty* cur_weak = delayed_weak_properties_;
+    delayed_weak_properties_ = NULL;
+    while (cur_weak != NULL) {
+      uword next_weak = cur_weak->ptr()->next_;
+      RawObject* raw_key = cur_weak->ptr()->key_;
+      // Reset the next pointer in the weak property.
+      cur_weak->ptr()->next_ = 0;
+      if (raw_key->IsMarked()) {
+        RawObject* raw_val = cur_weak->ptr()->value_;
+        marked = marked || (raw_val->IsHeapObject() && !raw_val->IsMarked());
+
+        // The key is marked so we make sure to properly visit all pointers
+        // originating from this weak property.
+        VisitingOldObject(cur_weak);
+        cur_weak->VisitPointers(this);
+      } else {
+        // Requeue this weak property to be handled later.
+        EnqueueWeakProperty(cur_weak);
+      }
+      // Advance to next weak property in the queue.
+      cur_weak = reinterpret_cast<RawWeakProperty*>(next_weak);
+    }
+
+    return marked;
+  }
+
+  void DrainMarkingStack() {
     RawObject* raw_obj = work_list_.Pop();
+    if ((raw_obj == NULL) && ProcessPendingWeakProperties()) {
+      raw_obj = work_list_.Pop();
+    }
+
     if (raw_obj == NULL) {
       ASSERT(visiting_old_object_ == NULL);
-      return false;
+      return;
     }
     do {
       do {
@@ -200,33 +231,13 @@ class MarkingVisitorBase : public ObjectPointerVisitor {
       } while (raw_obj != NULL);
 
       // Marking stack is empty.
-      // Process all the pending weak properties in this visitor.
-      RawWeakProperty* cur_weak = delayed_weak_properties_;
-      delayed_weak_properties_ = NULL;
-      while (cur_weak != NULL) {
-        uword next_weak = cur_weak->ptr()->next_;
-        RawObject* raw_key = cur_weak->ptr()->key_;
-        // Reset the next pointer in the weak property.
-        cur_weak->ptr()->next_ = 0;
-        if (raw_key->IsMarked()) {
-          // The key is marked so we make sure to properly visit all pointers
-          // originating from this weak property.
-          VisitingOldObject(cur_weak);
-          cur_weak->VisitPointers(this);
-        } else {
-          // Requeue this weak property to be handled later.
-          EnqueueWeakProperty(cur_weak);
-        }
-        // Advance to next weak property in the queue.
-        cur_weak = reinterpret_cast<RawWeakProperty*>(next_weak);
-      }
+      ProcessPendingWeakProperties();
 
       // Check whether any further work was pushed either by other markers or
       // by the handling of weak properties.
       raw_obj = work_list_.Pop();
     } while (raw_obj != NULL);
     VisitingOldObject(NULL);
-    return true;
   }
 
   void VisitPointers(RawObject** first, RawObject** last) {
@@ -282,6 +293,7 @@ class MarkingVisitorBase : public ObjectPointerVisitor {
     while (cur_weak != NULL) {
       uword next_weak = cur_weak->ptr()->next_;
       cur_weak->ptr()->next_ = 0;
+      RELEASE_ASSERT(!cur_weak->ptr()->key_->IsMarked());
       WeakProperty::Clear(cur_weak);
       weak_properties_cleared++;
       // Advance to next weak property in the queue.
@@ -569,28 +581,54 @@ class MarkTask : public ThreadPool::Task {
                                  skipped_code_functions);
       // Phase 1: Iterate over roots and drain marking stack in tasks.
       marker_->IterateRoots(isolate_, &visitor, task_index_, num_tasks_);
-      do {
-        visitor.DrainMarkingStack();
 
-        // I can't find more work right now. If no other task is busy,
-        // then there will never be more work (NB: 1 is *before* decrement).
-        if (AtomicOperations::FetchAndDecrement(num_busy_) == 1) break;
-
-        // Wait for some work to appear.
-        // TODO(iposva): Replace busy-waiting with a solution using Monitor,
-        // and redraw the boundaries between stack/visitor/task as needed.
-        while (marking_stack_->IsEmpty() &&
-               AtomicOperations::LoadRelaxed(num_busy_) > 0) {
+      bool more_to_mark = false;
+      do {
+        do {
+          visitor.DrainMarkingStack();
+
+          // I can't find more work right now. If no other task is busy,
+          // then there will never be more work (NB: 1 is *before* decrement).
+          if (AtomicOperations::FetchAndDecrement(num_busy_) == 1) break;
+
+          // Wait for some work to appear.
+          // TODO(iposva): Replace busy-waiting with a solution using Monitor,
+          // and redraw the boundaries between stack/visitor/task as needed.
+          while (marking_stack_->IsEmpty() &&
+                 AtomicOperations::LoadRelaxed(num_busy_) > 0) {
+          }
+
+          // If no tasks are busy, there will never be more work.
+          if (AtomicOperations::LoadRelaxed(num_busy_) == 0) break;
+
+          // I saw some work; get busy and compete for it.
+          AtomicOperations::FetchAndIncrement(num_busy_);
+        } while (true);
+        // Wait for all markers to stop.
+        barrier_->Sync();
+        ASSERT(AtomicOperations::LoadRelaxed(num_busy_) == 0);
+
+        // Check if we have any pending properties with marked keys.
+        // Those might have been marked by another marker.
+        more_to_mark = visitor.ProcessPendingWeakProperties();
+        if (more_to_mark) {
+          // We have more work to do. Notify others.
+          AtomicOperations::FetchAndIncrement(num_busy_);
         }
 
-        // If no tasks are busy, there will never be more work.
-        if (AtomicOperations::LoadRelaxed(num_busy_) == 0) break;
-
-        // I saw some work; get busy and compete for it.
-        AtomicOperations::FetchAndIncrement(num_busy_);
-      } while (true);
-      ASSERT(AtomicOperations::LoadRelaxed(num_busy_) == 0);
-      barrier_->Sync();
+        // Wait for all other markers to finish processing their pending
+        // weak properties and decide if they need to continue marking.
+        // Caveat: we need two barriers here to make this decision in lock step
+        // between all markers and the main thread.
+        barrier_->Sync();
+        if (!more_to_mark && (AtomicOperations::LoadRelaxed(num_busy_) > 0)) {
+          // All markers continue to marker as long as any single marker has
+          // some work to do.
+          AtomicOperations::FetchAndIncrement(num_busy_);
+          more_to_mark = true;
+        }
+        barrier_->Sync();

[siva, 2016/06/08 19:45:20]

Why is this barrier_->Sync() necessary? At this point it has been established
that at least one person has more work to do and so everybody needs to iterate
over again, couldn't they just start iterating again without doing the sync?

[Vyacheslav Egorov (Google), 2016/06/09 08:17:05]

Because strictly speaking one thread that has work can rush, while other threads
are descheduled, drain it's marking stack and decrement num_busy_ back. At this
point other threads are resumed check num_busy_ and see zero. This is a possible
though unlikely execution, but when writing concurrent code we have to think
about *all* possible concurrent executions:

   Marker A                  Marker B

1. more_to_mark = true       more_to_mark = false
2. lock inc num_busy         -- not running -----   num_busy = 1
3. Sync                      Sync
4. DrainMarkingStack         -- not running -----
5. lock dec num_busy_        -- not running -----   num_busy = 0 
6. -- not running -----      load num_busy_ -> 0
7. -- not running -----      more_to_mark = false

In this case Marker B does not realize that Marker A was running and potentially
did some work. 

[siva, 2016/06/09 16:39:53]

True if we let the threads run we get into a scenario of wanting the value of
num_busy corresponding to the iteration a thread is in.
With this sync you are ensuring that all threads will only be looking at the
value for the same iteration.

+      } while (more_to_mark);
 
       // Phase 2: Weak processing and follow-up marking on main thread.
       barrier_->Sync();
@@ -688,7 +726,19 @@ void GCMarker::MarkObjects(Isolate* isolate,
         ThreadPool* pool = Dart::thread_pool();
         pool->Run(mark_task);
       }
-      barrier.Sync();
+      bool more_to_mark = false;
+      do {
+        // Wait for all markers to stop.
+        barrier.Sync();
+
+        // Wait for all markers to go through weak properties and verify
+        // that there are no more objects to mark.
+        // Note: we need to have two barriers here because we want all markers
+        // and main thread to make decisions in lock step.
+        barrier.Sync();

[siva, 2016/06/08 19:45:20]

back to back barrier.Sync() calls with no code between them?

[Vyacheslav Egorov (Google), 2016/06/09 08:17:06]

These barriers correspond to barriers in the marker threads - there must be in
one to one correspondence. 

ThreadBarrier::Sync pauses the thread until all threads (that includes the main
thread) reach ::Sync. That is why there must be 1-1 correspondence between the
barriers.

[siva, 2016/06/09 16:39:53]

Acknowledged.

+        more_to_mark = AtomicOperations::LoadRelaxed(&num_busy) > 0;
+        barrier.Sync();
+      } while (more_to_mark);

[siva, 2016/06/08 19:45:20]

Is this loop necessary could we just assert after the first barrier.Sync() that
AtomicOperations::LoadRelaxed(&num_busy) == 0 because the MarkTask.Run method
exits only after AtomicOperations::LoadRelaxed(num_busy_) is 0.

[Vyacheslav Egorov (Google), 2016/06/09 08:17:05]

Yes, the loop is necessary, because there is a loop in the marker task - so we
must guarantee that we execute barriers in 1-1 correspondence. See above.

[siva, 2016/06/09 16:39:53]

Acknowledged.

 
       // Phase 2: Weak processing on main thread.
       {