[Backport] Two phase segmented heap dump

Summary: The minor-pause solution, which is presented in this PR, is a two phase segmented heap dump

1. Phase One(STW): Concurrent threads directly write data to multiple heap files.
2. Phase Two(Non-STW): Merge multiple heap files into one complete heap dump file.

Test Plan: manual, mat verification for merged dumps, test/hotspot/jtreg/serviceability

Reviewed-by : denghui.ddh, lei.yul, sanhong.lsh

Issue: dragonwell11#630

src/hotspot/share/logging/logTag.hpp

@@ -78,6 +78,7 @@
   LOG_TAG(handshake) \
   LOG_TAG(hashtables) \
   LOG_TAG(heap) \
+  LOG_TAG(heapdump) \
   LOG_TAG(humongous) \
   LOG_TAG(ihop) \
   LOG_TAG(iklass) \

src/hotspot/share/runtime/thread.hpp

@@ -416,6 +416,7 @@ class Thread: public ThreadShadow {
   virtual bool is_service_thread() const             { return false; }
   virtual bool is_hidden_from_external_view() const  { return false; }
   virtual bool is_jvmti_agent_thread() const         { return false; }
+  virtual bool is_AttachListener_thread() const      { return false; }
   // True iff the thread can perform GC operations at a safepoint.
   // Generally will be true only of VM thread and parallel GC WorkGang
   // threads.

src/hotspot/share/runtime/vmOperations.hpp

@@ -63,6 +63,7 @@
   template(Verify)                                \
   template(PrintJNI)                              \
   template(HeapDumper)                            \
+  template(HeapDumpMerge)                         \
   template(DeoptimizeTheWorld)                    \
   template(CollectForMetadataAllocation)          \
   template(GC_HeapInspection)                     \

src/hotspot/share/runtime/vmStructs.cpp

@@ -96,6 +96,7 @@
 #include "runtime/thread.inline.hpp"
 #include "runtime/vframeArray.hpp"
 #include "runtime/vmStructs.hpp"
+#include "services/attachListener.hpp"
 #include "utilities/globalDefinitions.hpp"
 #include "utilities/hashtable.hpp"
 #include "utilities/macros.hpp"
@@ -1395,6 +1396,7 @@ typedef PaddedEnd<ObjectMonitor>              PaddedObjectMonitor;
         declare_type(ServiceThread, JavaThread)                           \
         declare_type(CompilerThread, JavaThread)                          \
         declare_type(CodeCacheSweeperThread, JavaThread)                  \
+        declare_type(AttachListenerThread, JavaThread)                    \
   declare_toplevel_type(OSThread)                                         \
   declare_toplevel_type(JavaFrameAnchor)                                  \
                                                                           \

src/hotspot/share/services/attachListener.cpp

@@ -247,7 +247,7 @@ jint dump_heap(AttachOperation* op, outputStream* out) {
     // This helps reduces the amount of unreachable objects in the dump
     // and makes it easier to browse.
     HeapDumper dumper(live_objects_only /* request GC */, false, mini_heap_dump);
-    int res = dumper.dump(op->arg(0));
+    int res = dumper.dump(op->arg(0), NULL, -1, false, HeapDumper::default_num_of_dump_threads());
     if (res == 0) {
       if (mini_heap_dump) {
         out->print_cr("Mini-heap dump file created");
@@ -368,7 +368,7 @@ static AttachOperationFunctionInfo funcs[] = {
 // from the queue, examines the operation name (command), and dispatches
 // to the corresponding function to perform the operation.
 
-static void attach_listener_thread_entry(JavaThread* thread, TRAPS) {
+void AttachListenerThread::thread_entry(JavaThread* thread, TRAPS) {
   os::set_priority(thread, NearMaxPriority);
 
   assert(thread == Thread::current(), "Must be");
@@ -484,7 +484,7 @@ void AttachListener::init() {
   }
 
   { MutexLocker mu(Threads_lock);
-    JavaThread* listener_thread = new JavaThread(&attach_listener_thread_entry);
+    JavaThread* listener_thread = new AttachListenerThread();
 
     // Check that thread and osthread were created
     if (listener_thread == NULL || listener_thread->osthread() == NULL) {

src/hotspot/share/services/attachListener.hpp

@@ -59,6 +59,15 @@ enum AttachListenerState {
 
 template<> struct IsRegisteredEnum<AttachListenerState> : public TrueType {};
 
+class AttachListenerThread : public JavaThread {
+private:
+  static void thread_entry(JavaThread* thread, TRAPS);
+
+public:
+  AttachListenerThread() : JavaThread(&AttachListenerThread::thread_entry) {}
+  bool is_AttachListener_thread() const { return true; }
+};
+
 class AttachListener: AllStatic {
  public:
   static void vm_start() NOT_SERVICES_RETURN;

src/hotspot/share/services/diagnosticCommand.cpp

@@ -523,16 +523,21 @@ HeapDumpDCmd::HeapDumpDCmd(outputStream* output, bool heap) :
                "using the given compression level. 1 (recommended) is the fastest, "
                "9 the strongest compression.", "INT", false, "1"),
   _overwrite("-overwrite", "If specified, the dump file will be overwritten if it exists",
-           "BOOLEAN", false, "false") {
+           "BOOLEAN", false, "false"),
+  _parallel("-parallel", "Number of parallel threads to use for heap dump. The VM "
+                         "will try to use the specified number of threads, but might use fewer.",
+            "INT", false, "1") {
   _dcmdparser.add_dcmd_option(&_all);
   _dcmdparser.add_dcmd_option(&_mini_dump);
   _dcmdparser.add_dcmd_argument(&_filename);
   _dcmdparser.add_dcmd_option(&_gzip);
   _dcmdparser.add_dcmd_option(&_overwrite);
+  _dcmdparser.add_dcmd_option(&_parallel);
 }
 
 void HeapDumpDCmd::execute(DCmdSource source, TRAPS) {
   jlong level = -1; // -1 means no compression.
+  jlong parallel = HeapDumper::default_num_of_dump_threads();
 
   if (_gzip.is_set()) {
     level = _gzip.value();
@@ -543,11 +548,23 @@ void HeapDumpDCmd::execute(DCmdSource source, TRAPS) {
     }
   }
 
+  if (_parallel.is_set()) {
+    parallel = _parallel.value();
+
+    if (parallel < 0) {
+      output()->print_cr("Invalid number of parallel dump threads.");
+      return;
+    } else if (parallel == 0) {
+      // 0 implies to disable parallel heap dump, in such case, we use serial dump instead
+      parallel = 1;
+    }
+  }
+
   // Request a full GC before heap dump if _all is false
   // This helps reduces the amount of unreachable objects in the dump
   // and makes it easier to browse.
   HeapDumper dumper(!_all.value() /* request GC if _all is false*/, false, _mini_dump.value());
-  int res = dumper.dump(_filename.value(), output(), (int) level, _overwrite.value());
+  int res = dumper.dump(_filename.value(), output(), (int) level, _overwrite.value(), (uint)parallel);
   if (res == 0) {
     if (_mini_dump.value()) {
       output()->print_cr("Mini heap dump file created");

src/hotspot/share/services/diagnosticCommand.hpp

@@ -334,6 +334,7 @@ class HeapDumpDCmd : public DCmdWithParser {
   DCmdArgument<bool>  _mini_dump;
   DCmdArgument<jlong> _gzip;
   DCmdArgument<bool> _overwrite;
+  DCmdArgument<jlong> _parallel;
 public:
   HeapDumpDCmd(outputStream* output, bool heap);
   static const char* name() {