# HG changeset patch
# User Haoyu Li <leihouyju@gmail.com>
# Date 1574820663 -28800
#      Wed Nov 27 10:11:03 2019 +0800
# Node ID 358594a083bddabc040a86cf59929941b769d4b8
# Parent  3e3ba44bd9ce1943fe8d32353f27f41da046aa76
implement shadow region optimization

diff -r 3e3ba44bd9ce -r 358594a083bd src/hotspot/share/gc/parallel/psCompactionManager.cpp
--- a/src/hotspot/share/gc/parallel/psCompactionManager.cpp	Wed Nov 27 10:10:26 2019 +0800
+++ b/src/hotspot/share/gc/parallel/psCompactionManager.cpp	Wed Nov 27 10:11:03 2019 +0800
@@ -50,6 +50,8 @@
 ObjectStartArray*    ParCompactionManager::_start_array = NULL;
 ParMarkBitMap*       ParCompactionManager::_mark_bitmap = NULL;
 RegionTaskQueueSet*  ParCompactionManager::_region_array = NULL;
+GrowableArray<size_t >* ParCompactionManager::_shadow_region_array = NULL;
+Monitor*                ParCompactionManager::_shadow_region_monitor = NULL;
 
 ParCompactionManager::ParCompactionManager() :
     _action(CopyAndUpdate) {
@@ -100,6 +102,11 @@
     "Could not create ParCompactionManager");
   assert(ParallelScavengeHeap::heap()->workers().total_workers() != 0,
     "Not initialized?");
+
+  _shadow_region_array = new (ResourceObj::C_HEAP, mtInternal) GrowableArray<size_t >(10, true);
+
+  _shadow_region_monitor = new Monitor(Mutex::barrier, "CompactionManager monitor",
+                                       Mutex::_allow_vm_block_flag, Monitor::_safepoint_check_never);
 }
 
 void ParCompactionManager::reset_all_bitmap_query_caches() {
@@ -164,3 +171,32 @@
     }
   } while (!region_stack()->is_empty());
 }
+
+size_t ParCompactionManager::pop_shadow_region_mt_safe(PSParallelCompact::RegionData* region_ptr) {
+  while (true) {
+    MutexLocker ml(_shadow_region_monitor, Mutex::_no_safepoint_check_flag);
+    if (_shadow_region_array->is_empty()) {
+      // The corresponding heap region may be available now,
+      // so we don't need to acquire a shadow region anymore,
+      // and we return 0 to indicate this case
+      if (region_ptr->claimed()) {
+        return 0;
+      }
+    } else {
+      return _shadow_region_array->pop();
+    }
+  }
+}
+
+void ParCompactionManager::push_shadow_region_mt_safe(size_t shadow_region) {
+  MutexLocker ml(_shadow_region_monitor, Mutex::_no_safepoint_check_flag);
+  _shadow_region_array->push(shadow_region);
+}
+
+void ParCompactionManager::push_shadow_region(size_t shadow_region) {
+  _shadow_region_array->push(shadow_region);
+}
+
+void ParCompactionManager::remove_all_shadow_regions() {
+  _shadow_region_array->clear();
+}
\ No newline at end of file
diff -r 3e3ba44bd9ce -r 358594a083bd src/hotspot/share/gc/parallel/psCompactionManager.hpp
--- a/src/hotspot/share/gc/parallel/psCompactionManager.hpp	Wed Nov 27 10:10:26 2019 +0800
+++ b/src/hotspot/share/gc/parallel/psCompactionManager.hpp	Wed Nov 27 10:11:03 2019 +0800
@@ -25,6 +25,7 @@
 #ifndef SHARE_GC_PARALLEL_PSCOMPACTIONMANAGER_HPP
 #define SHARE_GC_PARALLEL_PSCOMPACTIONMANAGER_HPP
 
+#include "gc/parallel/psParallelCompact.hpp"
 #include "gc/shared/taskqueue.hpp"
 #include "memory/allocation.hpp"
 #include "utilities/stack.hpp"
@@ -77,6 +78,7 @@
 private:
   OverflowTaskQueue<oop, mtGC>        _marking_stack;
   ObjArrayTaskQueue             _objarray_stack;
+  size_t                        _next_shadow_region;
 
   // Is there a way to reuse the _marking_stack for the
   // saving empty regions?  For now just create a different
@@ -85,6 +87,14 @@
 
   static ParMarkBitMap* _mark_bitmap;
 
+  // Contains currently free shadow regions. We use it in
+  // a LIFO fashion for better data locality and utilization.
+  static GrowableArray<size_t>* _shadow_region_array;
+
+  // Provides mutual exclusive access of _shadow_region_array.
+  // See pop/push_shadow_region_mt_safe() below
+  static Monitor*               _shadow_region_monitor;
+
   Action _action;
 
   HeapWord* _last_query_beg;
@@ -109,6 +119,18 @@
   // marking stack and overflow stack directly.
 
  public:
+  static size_t  pop_shadow_region_mt_safe(PSParallelCompact::RegionData* region_ptr);
+  static void    push_shadow_region_mt_safe(size_t shadow_region);
+  static void    push_shadow_region(size_t shadow_region);
+  static void    remove_all_shadow_regions();
+
+  inline size_t  next_shadow_region() { return _next_shadow_region; }
+  inline void    set_next_shadow_region(size_t record) { _next_shadow_region = record; }
+  inline size_t  move_next_shadow_region_by(size_t workers) {
+    _next_shadow_region += workers;
+    return next_shadow_region();
+  }
+
   void reset_bitmap_query_cache() {
     _last_query_beg = NULL;
     _last_query_obj = NULL;
diff -r 3e3ba44bd9ce -r 358594a083bd src/hotspot/share/gc/parallel/psParallelCompact.cpp
--- a/src/hotspot/share/gc/parallel/psParallelCompact.cpp	Wed Nov 27 10:10:26 2019 +0800
+++ b/src/hotspot/share/gc/parallel/psParallelCompact.cpp	Wed Nov 27 10:11:03 2019 +0800
@@ -1023,6 +1023,7 @@
 void PSParallelCompact::post_compact()
 {
   GCTraceTime(Info, gc, phases) tm("Post Compact", &_gc_timer);
+  ParCompactionManager::remove_all_shadow_regions();
 
   for (unsigned int id = old_space_id; id < last_space_id; ++id) {
     // Clear the marking bitmap, summary data and split info.
@@ -2417,6 +2418,8 @@
     for (size_t cur = end_region - 1; cur + 1 > beg_region; --cur) {
       if (sd.region(cur)->claim_unsafe()) {
         ParCompactionManager* cm = ParCompactionManager::manager_array(worker_id);
+        bool result = sd.region(cur)->mark_normal();
+        assert(result, "Must succeed at this point.");
         cm->region_stack()->push(cur);
         region_logger.handle(cur);
         // Assign regions to tasks in round-robin fashion.
@@ -2602,6 +2605,10 @@
     if (ParCompactionManager::steal(worker_id, region_index)) {
       PSParallelCompact::fill_and_update_region(cm, region_index);
       cm->drain_region_stacks();
+    } else if (PSParallelCompact::steal_unavailable_region(cm, region_index)) {
+      // Fill and update an unavailable region with the help of a shadow region
+      PSParallelCompact::fill_and_update_shadow_region(cm, region_index);
+      cm->drain_region_stacks();
     } else {
       if (terminator->offer_termination()) {
         break;
@@ -2656,6 +2663,7 @@
   //
   // max push count is thus: last_space_id * (active_gc_threads * PAR_OLD_DENSE_PREFIX_OVER_PARTITIONING + 1)
   TaskQueue task_queue(last_space_id * (active_gc_threads * PAR_OLD_DENSE_PREFIX_OVER_PARTITIONING + 1));
+  initialize_shadow_regions(active_gc_threads);
   prepare_region_draining_tasks(active_gc_threads);
   enqueue_dense_prefix_tasks(task_queue, active_gc_threads);
 
@@ -2962,7 +2970,15 @@
     assert(cur->data_size() > 0, "region must have live data");
     cur->decrement_destination_count();
     if (cur < enqueue_end && cur->available() && cur->claim()) {
-      cm->push_region(sd.region(cur));
+      if (cur->mark_normal()) {
+        cm->push_region(sd.region(cur));
+      } else if (cur->mark_copied()) {
+        // Try to copy the content of the shadow region back to its corresponding
+        // heap region if the shadow region is filled
+        copy_back(sd.region_to_addr(cur->shadow_region()), sd.region_to_addr(cur));
+        cm->push_shadow_region_mt_safe(cur->shadow_region());
+        cur->set_completed();
+      }
     }
   }
 }
@@ -3040,7 +3056,7 @@
   return 0;
 }
 
-void PSParallelCompact::fill_region(ParCompactionManager* cm, size_t region_idx)
+void PSParallelCompact::fill_region(ParCompactionManager* cm, MoveAndUpdateClosure& closure, size_t region_idx)
 {
   typedef ParMarkBitMap::IterationStatus IterationStatus;
   const size_t RegionSize = ParallelCompactData::RegionSize;
@@ -3048,20 +3064,12 @@
   ParallelCompactData& sd = summary_data();
   RegionData* const region_ptr = sd.region(region_idx);
 
-  // Get the items needed to construct the closure.
-  HeapWord* dest_addr = sd.region_to_addr(region_idx);
-  SpaceId dest_space_id = space_id(dest_addr);
-  ObjectStartArray* start_array = _space_info[dest_space_id].start_array();
-  HeapWord* new_top = _space_info[dest_space_id].new_top();
-  assert(dest_addr < new_top, "sanity");
-  const size_t words = MIN2(pointer_delta(new_top, dest_addr), RegionSize);
-
   // Get the source region and related info.
   size_t src_region_idx = region_ptr->source_region();
   SpaceId src_space_id = space_id(sd.region_to_addr(src_region_idx));
   HeapWord* src_space_top = _space_info[src_space_id].space()->top();
-
-  MoveAndUpdateClosure closure(bitmap, cm, start_array, dest_addr, words);
+  HeapWord* dest_addr = sd.region_to_addr(region_idx);
+
   closure.set_source(first_src_addr(dest_addr, src_space_id, src_region_idx));
 
   // Adjust src_region_idx to prepare for decrementing destination counts (the
@@ -3080,7 +3088,7 @@
       decrement_destination_counts(cm, src_space_id, src_region_idx,
                                    closure.source());
       region_ptr->set_deferred_obj_addr(NULL);
-      region_ptr->set_completed();
+      closure.complete_region(cm, dest_addr, region_ptr);
       return;
     }
 
@@ -3125,11 +3133,12 @@
       // The last object did not fit.  Note that interior oop updates were
       // deferred, then copy enough of the object to fill the region.
       region_ptr->set_deferred_obj_addr(closure.destination());
+
       status = closure.copy_until_full(); // copies from closure.source()
 
       decrement_destination_counts(cm, src_space_id, src_region_idx,
                                    closure.source());
-      region_ptr->set_completed();
+      closure.complete_region(cm, dest_addr, region_ptr);
       return;
     }
 
@@ -3137,7 +3146,7 @@
       decrement_destination_counts(cm, src_space_id, src_region_idx,
                                    closure.source());
       region_ptr->set_deferred_obj_addr(NULL);
-      region_ptr->set_completed();
+      closure.complete_region(cm, dest_addr, region_ptr);
       return;
     }
 
@@ -3150,6 +3159,96 @@
   } while (true);
 }
 
+void PSParallelCompact::fill_and_update_region(ParCompactionManager* cm, size_t region_idx)
+{
+  MoveAndUpdateClosure cl(mark_bitmap(), cm, region_idx);
+  fill_region(cm, cl, region_idx);
+}
+
+void PSParallelCompact::fill_and_update_shadow_region(ParCompactionManager* cm, size_t region_idx)
+{
+  // Get a shadow region first
+  ParallelCompactData& sd = summary_data();
+  RegionData* const region_ptr = sd.region(region_idx);
+  size_t shadow_region = cm->pop_shadow_region_mt_safe(region_ptr);
+  // The zero return value indicates the corresponding heap region is available,
+  // so use MoveAndUpdateClosure to fill the normal region. Otherwise, use
+  // MoveAndUpdateShadowClosure to fill the acquired shadow region.
+  if (shadow_region == 0) {
+    MoveAndUpdateClosure cl(mark_bitmap(), cm, region_idx);
+    region_ptr->shadow_to_normal();
+    return fill_region(cm, cl, region_idx);
+  } else {
+    MoveAndUpdateShadowClosure cl(mark_bitmap(), cm, region_idx, shadow_region);
+    return fill_region(cm, cl, region_idx);
+  }
+}
+
+void PSParallelCompact::copy_back(HeapWord *shadow_addr, HeapWord *region_addr)
+{
+  Copy::aligned_conjoint_words(shadow_addr, region_addr, _summary_data.RegionSize);
+}
+
+bool PSParallelCompact::steal_unavailable_region(ParCompactionManager* cm, size_t &region_idx)
+{
+  size_t next = cm->next_shadow_region();
+  ParallelCompactData& sd = summary_data();
+  size_t old_new_top = sd.addr_to_region_idx(_space_info[old_space_id].new_top());
+  uint active_gc_threads = ParallelScavengeHeap::heap()->workers().active_workers();
+
+  while (next < old_new_top) {
+    if (sd.region(next)->mark_shadow()) {
+      region_idx = next;
+      return true;
+    }
+    next = cm->move_next_shadow_region_by(active_gc_threads);
+  }
+
+  return false;
+}
+
+// The shadow region is an optimization to address region dependencies in full GC. The basic
+// idea is making more regions available by temporally storing their live objects in empty
+// shadow regions to resolve dependencies between them and the destination regions. Therefore,
+// GC threads need not wait destination regions to be available before processing sources.
+//
+// A typical workflow would be:
+// After draining its own stack and failing to steal from others, a GC worker would pick an
+// unavailable region (destination count > 0) and get a shadow region. Then the worker fills
+// the shadow region by copying live objects from source regions of the unavailable one. Once
+// the unavailable region becomes available, the data in the shadow region will be copied back.
+// Shadow regions are empty regions in the to-space and regions between top and end of other spaces.
+//
+// More more details, please refer to §4.2 of the VEE'19 paper:
+// Haoyu Li, Mingyu Wu, Binyu Zang, and Haibo Chen. 2019. ScissorGC: scalable and efficient
+// compaction for Java full garbage collection. In Proceedings of the 15th ACM SIGPLAN/SIGOPS
+// International Conference on Virtual Execution Environments (VEE 2019). ACM, New York, NY, USA,
+// 108-121. DOI: https://doi.org/10.1145/3313808.3313820
+void PSParallelCompact::initialize_shadow_regions(uint parallel_gc_threads)
+{
+  const ParallelCompactData& sd = PSParallelCompact::summary_data();
+
+  for (unsigned int id = old_space_id; id < last_space_id; ++id) {
+    SpaceInfo* const space_info = _space_info + id;
+    MutableSpace* const space = space_info->space();
+
+    const size_t beg_region =
+      sd.addr_to_region_idx(sd.region_align_up(MAX2(space_info->new_top(), space->top())));
+    const size_t end_region =
+      sd.addr_to_region_idx(sd.region_align_down(space->end()));
+
+    for (size_t cur = beg_region; cur < end_region; ++cur) {
+      ParCompactionManager::push_shadow_region(cur);
+    }
+  }
+
+  size_t beg_region = sd.addr_to_region_idx(_space_info[old_space_id].dense_prefix());
+  for (uint i = 0; i < parallel_gc_threads; i++) {
+    ParCompactionManager *cm = ParCompactionManager::manager_array(i);
+    cm->set_next_shadow_region(beg_region + i);
+  }
+}
+
 void PSParallelCompact::fill_blocks(size_t region_idx)
 {
   // Fill in the block table elements for the specified region.  Each block
@@ -3201,46 +3300,6 @@
   }
 }
 
-void
-PSParallelCompact::move_and_update(ParCompactionManager* cm, SpaceId space_id) {
-  const MutableSpace* sp = space(space_id);
-  if (sp->is_empty()) {
-    return;
-  }
-
-  ParallelCompactData& sd = PSParallelCompact::summary_data();
-  ParMarkBitMap* const bitmap = mark_bitmap();
-  HeapWord* const dp_addr = dense_prefix(space_id);
-  HeapWord* beg_addr = sp->bottom();
-  HeapWord* end_addr = sp->top();
-
-  assert(beg_addr <= dp_addr && dp_addr <= end_addr, "bad dense prefix");
-
-  const size_t beg_region = sd.addr_to_region_idx(beg_addr);
-  const size_t dp_region = sd.addr_to_region_idx(dp_addr);
-  if (beg_region < dp_region) {
-    update_and_deadwood_in_dense_prefix(cm, space_id, beg_region, dp_region);
-  }
-
-  // The destination of the first live object that starts in the region is one
-  // past the end of the partial object entering the region (if any).
-  HeapWord* const dest_addr = sd.partial_obj_end(dp_region);
-  HeapWord* const new_top = _space_info[space_id].new_top();
-  assert(new_top >= dest_addr, "bad new_top value");
-  const size_t words = pointer_delta(new_top, dest_addr);
-
-  if (words > 0) {
-    ObjectStartArray* start_array = _space_info[space_id].start_array();
-    MoveAndUpdateClosure closure(bitmap, cm, start_array, dest_addr, words);
-
-    ParMarkBitMap::IterationStatus status;
-    status = bitmap->iterate(&closure, dest_addr, end_addr);
-    assert(status == ParMarkBitMap::full, "iteration not complete");
-    assert(bitmap->find_obj_beg(closure.source(), end_addr) == end_addr,
-           "live objects skipped because closure is full");
-  }
-}
-
 jlong PSParallelCompact::millis_since_last_gc() {
   // We need a monotonically non-decreasing time in ms but
   // os::javaTimeMillis() does not guarantee monotonicity.
@@ -3262,9 +3321,9 @@
 
 ParMarkBitMap::IterationStatus MoveAndUpdateClosure::copy_until_full()
 {
-  if (source() != destination()) {
+  if (source() != copy_destination()) {
     DEBUG_ONLY(PSParallelCompact::check_new_location(source(), destination());)
-    Copy::aligned_conjoint_words(source(), destination(), words_remaining());
+    Copy::aligned_conjoint_words(source(), copy_destination(), words_remaining());
   }
   update_state(words_remaining());
   assert(is_full(), "sanity");
@@ -3283,13 +3342,19 @@
 
   // This test is necessary; if omitted, the pointer updates to a partial object
   // that crosses the dense prefix boundary could be overwritten.
-  if (source() != destination()) {
+  if (source() != copy_destination()) {
     DEBUG_ONLY(PSParallelCompact::check_new_location(source(), destination());)
-    Copy::aligned_conjoint_words(source(), destination(), words);
+    Copy::aligned_conjoint_words(source(), copy_destination(), words);
   }
   update_state(words);
 }
 
+void MoveAndUpdateClosure::complete_region(ParCompactionManager *cm, HeapWord *dest_addr,
+                                           PSParallelCompact::RegionData *region_ptr) {
+  assert(region_ptr->shadow_state() == ParallelCompactData::RegionData::FinishedShadow, "Region should be finished");
+  region_ptr->set_completed();
+}
+
 ParMarkBitMapClosure::IterationStatus
 MoveAndUpdateClosure::do_addr(HeapWord* addr, size_t words) {
   assert(destination() != NULL, "sanity");
@@ -3308,20 +3373,36 @@
     _start_array->allocate_block(destination());
   }
 
-  if (destination() != source()) {
+  if (copy_destination() != source()) {
     DEBUG_ONLY(PSParallelCompact::check_new_location(source(), destination());)
-    Copy::aligned_conjoint_words(source(), destination(), words);
+    Copy::aligned_conjoint_words(source(), copy_destination(), words);
   }
 
-  oop moved_oop = (oop) destination();
+  oop moved_oop = (oop) copy_destination();
   compaction_manager()->update_contents(moved_oop);
   assert(oopDesc::is_oop_or_null(moved_oop), "Expected an oop or NULL at " PTR_FORMAT, p2i(moved_oop));
 
   update_state(words);
-  assert(destination() == (HeapWord*)moved_oop + moved_oop->size(), "sanity");
+  assert(copy_destination() == (HeapWord*)moved_oop + moved_oop->size(), "sanity");
   return is_full() ? ParMarkBitMap::full : ParMarkBitMap::incomplete;
 }
 
+void MoveAndUpdateShadowClosure::complete_region(ParCompactionManager *cm, HeapWord *dest_addr,
+                                                 PSParallelCompact::RegionData *region_ptr) {
+  assert(region_ptr->shadow_state() == ParallelCompactData::RegionData::Shadow, "Region should be shadow");
+  // Record the shadow region index
+  region_ptr->set_shadow_region(_shadow);
+  // Mark the shadow region filled
+  region_ptr->mark_filled();
+  // Try to copy the content of the shadow region back to its corresponding
+  // heap region if available
+  if (((region_ptr->available() && region_ptr->claim()) || region_ptr->claimed()) && region_ptr->mark_copied()) {
+    region_ptr->set_completed();
+    PSParallelCompact::copy_back(PSParallelCompact::summary_data().region_to_addr(_shadow), dest_addr);
+    cm->push_shadow_region_mt_safe(_shadow);
+  }
+}
+
 UpdateOnlyClosure::UpdateOnlyClosure(ParMarkBitMap* mbm,
                                      ParCompactionManager* cm,
                                      PSParallelCompact::SpaceId space_id) :
diff -r 3e3ba44bd9ce -r 358594a083bd src/hotspot/share/gc/parallel/psParallelCompact.hpp
--- a/src/hotspot/share/gc/parallel/psParallelCompact.hpp	Wed Nov 27 10:10:26 2019 +0800
+++ b/src/hotspot/share/gc/parallel/psParallelCompact.hpp	Wed Nov 27 10:11:03 2019 +0800
@@ -239,6 +239,9 @@
     // The first region containing data destined for this region.
     size_t source_region() const { return _source_region; }
 
+    // Reuse _source_region to store the corresponding shadow region index
+    size_t shadow_region() const { return _source_region; }
+
     // The object (if any) starting in this region and ending in a different
     // region that could not be updated during the main (parallel) compaction
     // phase.  This is different from _partial_obj_addr, which is an object that
@@ -307,6 +310,7 @@
     // These are not atomic.
     void set_destination(HeapWord* addr)       { _destination = addr; }
     void set_source_region(size_t region)      { _source_region = region; }
+    void set_shadow_region(size_t region)      { _source_region = region; }
     void set_deferred_obj_addr(HeapWord* addr) { _partial_obj_addr = addr; }
     void set_partial_obj_addr(HeapWord* addr)  { _partial_obj_addr = addr; }
     void set_partial_obj_size(size_t words)    {
@@ -326,6 +330,32 @@
     inline void decrement_destination_count();
     inline bool claim();
 
+    // Possible values of _shadow_state, and transition is as follows
+    // Normal Path:
+    // UnusedRegion -> mark_normal() -> NormalRegion
+    // Shadow Path:
+    // UnusedRegion -> mark_shadow() -> ShadowRegion ->
+    // mark_filled() -> FilledShadow -> mark_copied() -> CopiedShadow
+    static const int UnusedRegion = 0; // The region is not collected yet
+    static const int ShadowRegion = 1; // Stolen by an idle thread, and a shadow region is created for it
+    static const int FilledShadow = 2; // Its shadow region has been filled and ready to be copied back
+    static const int CopiedShadow = 3; // The data of the shadow region has been copied back
+    static const int NormalRegion = 4; // The region will be collected by the original parallel algorithm
+
+    // Mark the current region as normal or shadow to enter different processing paths
+    inline bool mark_normal();
+    inline bool mark_shadow();
+    // Mark the shadow region as filled and ready to be copied back
+    inline void mark_filled();
+    // Mark the shadow region as copied back to avoid double copying.
+    inline bool mark_copied();
+    // Special case: see the comment in PSParallelCompact::fill_and_update_shadow_region.
+    // Return to the normal path here
+    inline void shadow_to_normal();
+
+
+    int shadow_state() { return _shadow_state; }
+
   private:
     // The type used to represent object sizes within a region.
     typedef uint region_sz_t;
@@ -346,6 +376,7 @@
     region_sz_t          _partial_obj_size;
     region_sz_t volatile _dc_and_los;
     bool        volatile _blocks_filled;
+    int         volatile _shadow_state;
 
 #ifdef ASSERT
     size_t               _blocks_filled_count;   // Number of block table fills.
@@ -596,6 +627,29 @@
   return old == los;
 }
 
+inline bool ParallelCompactData::RegionData::mark_normal() {
+  return Atomic::cmpxchg(NormalRegion, &_shadow_state, UnusedRegion, memory_order_relaxed) == UnusedRegion;
+}
+
+inline bool ParallelCompactData::RegionData::mark_shadow() {
+  if (_shadow_state != UnusedRegion) return false;
+  return Atomic::cmpxchg(ShadowRegion, &_shadow_state, UnusedRegion, memory_order_relaxed) == UnusedRegion;
+}
+
+inline void ParallelCompactData::RegionData::mark_filled() {
+  int old = Atomic::cmpxchg(FilledShadow, &_shadow_state, ShadowRegion, memory_order_relaxed);
+  assert(old == Shadow, "Fail to mark the region as filled");
+}
+
+inline bool ParallelCompactData::RegionData::mark_copied() {
+  return Atomic::cmpxchg(CopiedShadow, &_shadow_state, FilledShadow, memory_order_relaxed) == FilledShadow;
+}
+
+void ParallelCompactData::RegionData::shadow_to_normal() {
+  int old = Atomic::cmpxchg(NormalRegion, &_shadow_state, ShadowRegion, memory_order_relaxed);
+  assert(old == Shadow, "Fail to mark the region as finish");
+}
+
 inline ParallelCompactData::RegionData*
 ParallelCompactData::region(size_t region_idx) const
 {
@@ -1130,9 +1184,6 @@
   static inline HeapWord*         dense_prefix(SpaceId space_id);
   static inline ObjectStartArray* start_array(SpaceId space_id);
 
-  // Move and update the live objects in the specified space.
-  static void move_and_update(ParCompactionManager* cm, SpaceId space_id);
-
   // Process the end of the given region range in the dense prefix.
   // This includes saving any object not updated.
   static void dense_prefix_regions_epilogue(ParCompactionManager* cm,
@@ -1182,11 +1233,16 @@
                                            size_t beg_region,
                                            HeapWord* end_addr);
 
-  // Fill a region, copying objects from one or more source regions.
-  static void fill_region(ParCompactionManager* cm, size_t region_idx);
-  static void fill_and_update_region(ParCompactionManager* cm, size_t region) {
-    fill_region(cm, region);
-  }
+  static void fill_region(ParCompactionManager* cm, MoveAndUpdateClosure& closure, size_t region);
+  static void fill_and_update_region(ParCompactionManager* cm, size_t region);
+
+  static bool steal_unavailable_region(ParCompactionManager* cm, size_t& region_idx);
+  static void fill_and_update_shadow_region(ParCompactionManager* cm, size_t region);
+  // Copy the content of a shadow region back to its corresponding heap region
+  static void copy_back(HeapWord* shadow_addr, HeapWord* region_addr);
+  // Collect empty regions as shadow regions and initialize the
+  // _next_shadow_region filed for each compact manager
+  static void initialize_shadow_regions(uint parallel_gc_threads);
 
   // Fill in the block table for the specified region.
   static void fill_blocks(size_t region_idx);
@@ -1233,19 +1289,20 @@
 };
 
 class MoveAndUpdateClosure: public ParMarkBitMapClosure {
+  static inline size_t calculate_words_remaining(size_t region);
  public:
   inline MoveAndUpdateClosure(ParMarkBitMap* bitmap, ParCompactionManager* cm,
-                              ObjectStartArray* start_array,
-                              HeapWord* destination, size_t words);
+                              size_t region);
 
   // Accessors.
   HeapWord* destination() const         { return _destination; }
+  HeapWord* copy_destination() const    { return _destination + _offset; }
 
   // If the object will fit (size <= words_remaining()), copy it to the current
   // destination, update the interior oops and the start array and return either
   // full (if the closure is full) or incomplete.  If the object will not fit,
   // return would_overflow.
-  virtual IterationStatus do_addr(HeapWord* addr, size_t size);
+  IterationStatus do_addr(HeapWord* addr, size_t size);
 
   // Copy enough words to fill this closure, starting at source().  Interior
   // oops and the start array are not updated.  Return full.
@@ -1256,25 +1313,37 @@
   // array are not updated.
   void copy_partial_obj();
 
- protected:
+  virtual void complete_region(ParCompactionManager* cm, HeapWord* dest_addr,
+                               PSParallelCompact::RegionData* region_ptr);
+
+protected:
   // Update variables to indicate that word_count words were processed.
   inline void update_state(size_t word_count);
 
  protected:
+  HeapWord*               _destination;         // Next addr to be written.
   ObjectStartArray* const _start_array;
-  HeapWord*               _destination;         // Next addr to be written.
+  size_t                  _offset;
 };
 
+inline size_t MoveAndUpdateClosure::calculate_words_remaining(size_t region) {
+  HeapWord* dest_addr = PSParallelCompact::summary_data().region_to_addr(region);
+  PSParallelCompact::SpaceId dest_space_id = PSParallelCompact::space_id(dest_addr);
+  HeapWord* new_top = PSParallelCompact::new_top(dest_space_id);
+  assert(dest_addr < new_top, "sanity");
+
+  return MIN2(pointer_delta(new_top, dest_addr), ParallelCompactData::RegionSize);
+}
+
 inline
 MoveAndUpdateClosure::MoveAndUpdateClosure(ParMarkBitMap* bitmap,
                                            ParCompactionManager* cm,
-                                           ObjectStartArray* start_array,
-                                           HeapWord* destination,
-                                           size_t words) :
-  ParMarkBitMapClosure(bitmap, cm, words), _start_array(start_array)
-{
-  _destination = destination;
-}
+                                           size_t region_idx) :
+  ParMarkBitMapClosure(bitmap, cm, calculate_words_remaining(region_idx)),
+  _destination(PSParallelCompact::summary_data().region_to_addr(region_idx)),
+  _start_array(PSParallelCompact::start_array(PSParallelCompact::space_id(_destination))),
+  _offset(0) { }
+
 
 inline void MoveAndUpdateClosure::update_state(size_t words)
 {
@@ -1283,6 +1352,36 @@
   _destination += words;
 }
 
+class MoveAndUpdateShadowClosure: public MoveAndUpdateClosure {
+  inline size_t calculate_shadow_offset(size_t region_idx, size_t shadow_idx);
+public:
+  inline MoveAndUpdateShadowClosure(ParMarkBitMap* bitmap, ParCompactionManager* cm,
+                       size_t region, size_t shadow);
+
+  virtual void complete_region(ParCompactionManager* cm, HeapWord* dest_addr,
+                               PSParallelCompact::RegionData* region_ptr);
+
+private:
+  size_t _shadow;
+};
+
+inline size_t MoveAndUpdateShadowClosure::calculate_shadow_offset(size_t region_idx, size_t shadow_idx) {
+  ParallelCompactData& sd = PSParallelCompact::summary_data();
+  HeapWord* dest_addr = sd.region_to_addr(region_idx);
+  HeapWord* shadow_addr = sd.region_to_addr(shadow_idx);
+  return pointer_delta(shadow_addr, dest_addr);
+}
+
+inline
+MoveAndUpdateShadowClosure::MoveAndUpdateShadowClosure(ParMarkBitMap *bitmap,
+                                                       ParCompactionManager *cm,
+                                                       size_t region,
+                                                       size_t shadow) :
+  MoveAndUpdateClosure(bitmap, cm, region),
+  _shadow(shadow) {
+  _offset = calculate_shadow_offset(region, shadow);
+}
+
 class UpdateOnlyClosure: public ParMarkBitMapClosure {
  private:
   const PSParallelCompact::SpaceId _space_id;
