VM: Fix WeakProperty processing during parallel marking.

runtime/vm/gc_marker.cc

 // Copyright (c) 2011, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 #include "vm/gc_marker.h"
 
 #include "vm/allocation.h"
 #include "vm/dart_api_state.h"
 #include "vm/isolate.h"
 #include "vm/log.h"
@@ skipping 159 matching lines @@
   uintptr_t marked_bytes() const { return marked_bytes_; }
 
   intptr_t live_count(intptr_t class_id) {
     return class_stats_count_[class_id];
   }
 
   intptr_t live_size(intptr_t class_id) {
     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 {
         // First drain the marking stacks.
         VisitingOldObject(raw_obj);
         const intptr_t class_id = raw_obj->GetClassId();
         if (class_id != kWeakPropertyCid) {
           marked_bytes_ += raw_obj->VisitPointers(this);
         } else {
           RawWeakProperty* raw_weak =
               reinterpret_cast<RawWeakProperty*>(raw_obj);
           marked_bytes_ += ProcessWeakProperty(raw_weak);
         }
         raw_obj = work_list_.Pop();
       } 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) {
     for (RawObject** current = first; current <= last; current++) {
       MarkObject(*current, current);
     }
   }
 
   bool visit_function_code() const {
     return skipped_code_functions_ == NULL;
@@ skipping 35 matching lines @@
     if (skipped_code_functions_ != NULL) {
       skipped_code_functions_->DetachCode();
     }
     // Clear pending weak properties.
     RawWeakProperty* cur_weak = delayed_weak_properties_;
     delayed_weak_properties_ = NULL;
     intptr_t weak_properties_cleared = 0;
     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.
       cur_weak = reinterpret_cast<RawWeakProperty*>(next_weak);
     }
   }
 
   void VisitingOldObject(RawObject* obj) {
     ASSERT((obj == NULL) || obj->IsOldObject());
     visiting_old_object_ = obj;
@@ skipping 267 matching lines @@
       Thread* thread = Thread::Current();
       TIMELINE_FUNCTION_GC_DURATION(thread, "MarkTask");
       StackZone stack_zone(thread);
       Zone* zone = stack_zone.GetZone();
       SkippedCodeFunctions* skipped_code_functions =
           collect_code_ ? new(zone) SkippedCodeFunctions() : NULL;
       SyncMarkingVisitor visitor(isolate_, heap_, page_space_, marking_stack_,
                                  skipped_code_functions);
       // Phase 1: Iterate over roots and drain marking stack in tasks.
       marker_->IterateRoots(isolate_, &visitor, task_index_, num_tasks_);
+
+      bool more_to_mark = false;
       do {
-        visitor.DrainMarkingStack();
+        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;
+          // 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) {
+          // 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();
 
       // Phase 3: Finalize results from all markers (detach code, etc.).
       if (FLAG_log_marker_tasks) {
         THR_Print("Task %" Pd " marked %" Pd " bytes.\n",
                   task_index_, visitor.marked_bytes());
       }
       marker_->FinalizeResultsFrom(&visitor);
@@ skipping 77 matching lines @@
       uintptr_t num_busy = num_tasks;
       // Phase 1: Iterate over roots and drain marking stack in tasks.
       for (intptr_t i = 0; i < num_tasks; ++i) {
         MarkTask* mark_task =
             new MarkTask(this, isolate, heap_, page_space, &marking_stack,
                          &barrier, collect_code,
                          i, num_tasks, &num_busy);
         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.
       {
         TIMELINE_FUNCTION_GC_DURATION(thread, "WeakHandleProcessing");
         MarkingWeakVisitor mark_weak;
         IterateWeakRoots(isolate, &mark_weak);
       }
       barrier.Sync();
 
       // Phase 3: Finalize results from all markers (detach code, etc.).
       barrier.Exit();
     }
     ProcessWeakTables(page_space);
     ProcessObjectIdTable(isolate);
   }
   Epilogue(isolate, invoke_api_callbacks);
 }
 
 }  // namespace dart