HDFS-16512. Improve oiv tool to parse fsimage file in parallel with XML format

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java

@@ -63,6 +63,7 @@ public class OfflineImageViewerPB {
       + "  * XML: This processor creates an XML document with all elements of\n"
       + "    the fsimage enumerated, suitable for further analysis by XML\n"
       + "    tools.\n"
+      + "    -m  defines multiThread to process sub-sections.\n"
       + "  * ReverseXML: This processor takes an XML file and creates a\n"
       + "    binary fsimage containing the same elements.\n"
       + "  * FileDistribution: This processor analyzes the file size\n"
@@ -81,8 +82,7 @@ public class OfflineImageViewerPB {
       + "    changed via the -delimiter argument.\n"
       + "    -sp print storage policy, used by delimiter only.\n"
       + "    -ec print erasure coding policy, used by delimiter only.\n"
-      + "    -m  defines multiThread to process sub-sections, \n"
-      + "    used by delimiter only.\n"
+      + "    -m  defines multiThread to process sub-sections.\n"
       + "  * DetectCorruption: Detect potential corruption of the image by\n"
       + "    selectively loading parts of it and actively searching for\n"
       + "    inconsistencies. Outputs a summary of the found corruptions\n"
@@ -127,6 +127,8 @@ public class OfflineImageViewerPB {
       + "                       constructs the namespace in memory \n"
       + "                       before outputting text.\n"
       + "-m,--multiThread <arg> Use multiThread to process sub-sections.\n"
+      + "                       This option is used with XML and Delimited\n"
+      + "                       processors.\n"
       + "-h,--help              Display usage information and exit\n";
 
   /**
@@ -223,9 +225,7 @@ public static int run(String[] args) throws Exception {
         }
         break;
       case "XML":
-        try (RandomAccessFile r = new RandomAccessFile(inputFile, "r")) {
-          new PBImageXmlWriter(conf, out).visit(r);
-        }
+        new PBImageXmlWriter(conf, out, threads, outputFile).visit(inputFile);
         break;
       case "REVERSEXML":
         try {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/PBImageXmlWriter.java

@@ -18,18 +18,26 @@
 package org.apache.hadoop.hdfs.tools.offlineImageViewer;
 
 import java.io.BufferedInputStream;
+import java.io.EOFException;
+import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintStream;
 import java.io.RandomAccessFile;
+import java.nio.channels.FileChannel;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Date;
 import java.util.Map;
 import java.util.TimeZone;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.codec.binary.Hex;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -68,7 +76,10 @@
 import org.apache.hadoop.io.erasurecode.ECSchema;
 import org.apache.hadoop.util.LimitInputStream;
 import org.apache.hadoop.util.Lists;
+import org.apache.hadoop.util.Time;
 import org.apache.hadoop.util.VersionInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
 import org.apache.hadoop.thirdparty.protobuf.ByteString;
@@ -86,6 +97,9 @@
  */
 @InterfaceAudience.Private
 public final class PBImageXmlWriter {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(PBImageXmlWriter.class);
+
   public static final String NAME_SECTION_NAME = "NameSection";
   public static final String ERASURE_CODING_SECTION_NAME =
       "ErasureCodingSection";
@@ -270,6 +284,9 @@ public final class PBImageXmlWriter {
   private final Configuration conf;
   private final PrintStream out;
   private final SimpleDateFormat isoDateFormat;
+  private final int numThreads;
+  private final String parallelOutputFile;
+  private File filename;
   private SerialNumberManager.StringTable stringTable;
 
   public static SimpleDateFormat createSimpleDateFormat() {
@@ -280,11 +297,25 @@ public static SimpleDateFormat createSimpleDateFormat() {
   }
 
   public PBImageXmlWriter(Configuration conf, PrintStream out) {
+    this(conf, out, 1, "-");
+  }
+
+  PBImageXmlWriter(Configuration conf, PrintStream out, int numThreads,
+      String parallelOutputFile) {
     this.conf = conf;
     this.out = out;
+    this.numThreads = numThreads;
+    this.parallelOutputFile = parallelOutputFile;
     this.isoDateFormat = createSimpleDateFormat();
   }
 
+  public void visit(String filePath) throws IOException {
+    filename = new File(filePath);
+    try (RandomAccessFile file = new RandomAccessFile(filename, "r")) {
+      visit(file);
+    }
+  }
+
   public void visit(RandomAccessFile file) throws IOException {
     if (!FSImageUtil.checkFileFormat(file)) {
       throw new IOException("Unrecognized FSImage");
@@ -320,6 +351,8 @@ public int compare(FileSummary.Section s1, FileSummary.Section s2) {
           }
         }
       });
+      ArrayList<FileSummary.Section> inodeSubSections =
+          getSubSections(sections, SectionName.INODE_SUB);
 
       for (FileSummary.Section s : sections) {
         fin.getChannel().position(s.getOffset());
@@ -342,7 +375,11 @@ public int compare(FileSummary.Section s1, FileSummary.Section s2) {
           dumpErasureCodingSection(is);
           break;
         case INODE:
-          dumpINodeSection(is);
+          if (shouldOutputINodeSectionInParallel(inodeSubSections)) {
+            dumpINodeSectionInParallel(summary, inodeSubSections);
+          } else {
+            dumpINodeSection(is);
+          }
           break;
         case INODE_REFERENCE:
           dumpINodeReferenceSection(is);
@@ -618,12 +655,151 @@ private void dumpINodeSection(InputStream in) throws IOException {
     out.print("<" + INODE_SECTION_NAME + ">");
     o(INODE_SECTION_LAST_INODE_ID, s.getLastInodeId());
     o(INODE_SECTION_NUM_INODES, s.getNumInodes());
-    for (int i = 0; i < s.getNumInodes(); ++i) {
+    dumpINodes(in, s.getNumInodes());
+    out.print("</" + INODE_SECTION_NAME + ">\n");
+  }
+
+  private long dumpINodes(InputStream in) throws IOException {
+    long count = 0;
+    while (true) {
       INodeSection.INode p = INodeSection.INode.parseDelimitedFrom(in);
-      out.print("<" + INODE_SECTION_INODE + ">");
-      dumpINodeFields(p);
-      out.print("</" + INODE_SECTION_INODE + ">\n");
+      if (p == null) {
+        break;
+      }
+      dumpINode(p);
+      count++;
+    }
+    return count;
+  }
+
+  private long dumpINodes(InputStream in, long numINodes) throws IOException {
+    long count = 0;
+    for (long i = 0; i < numINodes; ++i) {
+      INodeSection.INode p = INodeSection.INode.parseDelimitedFrom(in);
+      if (p == null) {
+        throw new EOFException("Unexpected end of INode section.");
+      }
+      dumpINode(p);
+      count++;
+    }
+    return count;
+  }
+
+  private void dumpINode(INodeSection.INode p) {
+    out.print("<" + INODE_SECTION_INODE + ">");
+    dumpINodeFields(p);
+    out.print("</" + INODE_SECTION_INODE + ">\n");
+  }
+
+  private boolean shouldOutputINodeSectionInParallel(
+      ArrayList<FileSummary.Section> subSections) {
+    boolean parallel = numThreads > 1 && !parallelOutputFile.equals("-") &&
+        filename != null && subSections.size() > 1;
+    if (!parallel) {
+      LOG.info("Serial XML output due to threads num: {}, parallel output " +
+          "file: {}, input file: {}, subSections: {}.", numThreads,
+          parallelOutputFile, filename, subSections.size());
+    }
+    return parallel;
+  }
+
+  /**
+   * Process INode sub-sections in parallel and merge the XML inode elements
+   * back into the current INodeSection in sub-section order.
+   */
+  private void dumpINodeSectionInParallel(FileSummary summary,
+      ArrayList<FileSummary.Section> subSections) throws IOException {
+    int nThreads = Integer.min(numThreads, subSections.size());
+    LOG.info("Outputting XML INode section in parallel with {} sub-sections " +
+        "using {} threads", subSections.size(), nThreads);
+    final CopyOnWriteArrayList<IOException> exceptions =
+        new CopyOnWriteArrayList<>();
+    CountDownLatch latch = new CountDownLatch(subSections.size());
+    ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
+    AtomicLong expectedINodes = new AtomicLong(0);
+    AtomicLong lastInodeId = new AtomicLong(0);
+    AtomicLong totalParsed = new AtomicLong(0);
+    String[] paths = new String[subSections.size()];
+
+    for (int i = 0; i < subSections.size(); i++) {
+      paths[i] = parallelOutputFile + ".tmp." + i;
+      int index = i;
+      executorService.submit(() -> {
+        LOG.info("Output XML INodes of section-{}", index);
+        InputStream is = null;
+        try (PrintStream outStream = new PrintStream(paths[index], "UTF-8")) {
+          long startTime = Time.monotonicNow();
+          is = getInputStreamForSection(subSections.get(index),
+              summary.getCodec());
+          if (index == 0) {
+            INodeSection s = INodeSection.parseDelimitedFrom(is);
+            lastInodeId.set(s.getLastInodeId());
+            expectedINodes.set(s.getNumInodes());
+          }
+          PBImageXmlWriter writer = new PBImageXmlWriter(conf, outStream);
+          writer.stringTable = stringTable;
+          totalParsed.addAndGet(writer.dumpINodes(is));
+          outStream.flush();
+          if (outStream.checkError()) {
+            throw new IOException("Failed to write parallel XML output file "
+                + paths[index]);
+          }
+          long timeTaken = Time.monotonicNow() - startTime;
+          LOG.info("Time to output XML INodes of section-{}: {} ms",
+              index, timeTaken);
+        } catch (Exception e) {
+          exceptions.add(new IOException(e));
+        } finally {
+          latch.countDown();
+          try {
+            if (is != null) {
+              is.close();
+            }
+          } catch (IOException ioe) {
+            LOG.warn("Failed to close the input stream, ignoring", ioe);
+          }
+        }
+      });
     }
+
+    try {
+      latch.await();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IOException(e);
+    } finally {
+      executorService.shutdown();
+    }
+
+    if (exceptions.size() != 0) {
+      LOG.error("Failed to output XML INode sub-sections, {} exception(s) " +
+          "occurred.", exceptions.size());
+      cleanupTmpFiles(paths);
+      throw exceptions.get(0);
+    }
+    if (totalParsed.get() != expectedINodes.get()) {
+      cleanupTmpFiles(paths);
+      throw new IOException("Expected to parse " + expectedINodes.get() +
+          " XML INodes in parallel, but parsed " + totalParsed.get() +
+          ". The image may be corrupt.");
+    }
+
+    LOG.info("Completed outputting all INode sub-sections to {} tmp files.",
+            subSections.size());
+
+    out.print("<" + INODE_SECTION_NAME + ">");
+    o(INODE_SECTION_LAST_INODE_ID, lastInodeId.get());
+    o(INODE_SECTION_NUM_INODES, expectedINodes.get());
+
+    long startTime = Time.monotonicNow();
+    try {
+      mergeFilesToOutput(paths, out);
+    } finally {
+      cleanupTmpFiles(paths);
+    }
+    long timeTaken = Time.monotonicNow() - startTime;
+    LOG.info("Completed all stages. Time to merge files: {} ms", timeTaken);
+
     out.print("</" + INODE_SECTION_NAME + ">\n");
   }
 
@@ -859,6 +1035,72 @@ private void loadStringTable(InputStream in) throws IOException {
     stringTable = FSImageLoader.loadStringTable(in);
   }
 
+  private ArrayList<FileSummary.Section> getSubSections(
+      ArrayList<FileSummary.Section> sections, SectionName sectionName) {
+    ArrayList<FileSummary.Section> subSections = new ArrayList<>();
+    for (FileSummary.Section section : sections) {
+      if (SectionName.fromString(section.getName()) == sectionName) {
+        subSections.add(section);
+      }
+    }
+    return subSections;
+  }
+
+  /**
+   * Given a FSImage FileSummary.section, return a LimitInput stream set to
+   * the starting position of the section and limited to the section length.
+   */
+  private InputStream getInputStreamForSection(FileSummary.Section section,
+      String compressionCodec) throws IOException {
+    FileInputStream fin = new FileInputStream(filename);
+    try {
+      FileChannel channel = fin.getChannel();
+      channel.position(section.getOffset());
+      InputStream in = new BufferedInputStream(new LimitInputStream(fin,
+          section.getLength()));
+      return FSImageUtil.wrapInputStreamForCompression(conf, compressionCodec,
+          in);
+    } catch (IOException e) {
+      fin.close();
+      throw e;
+    }
+  }
+
+  private void mergeFilesToOutput(String[] srcPaths, PrintStream out)
+          throws IOException {
+    if (srcPaths == null || srcPaths.length < 1) {
+      LOG.warn("no source files to merge.");
+      return;
+    }
+
+    File[] files = new File[srcPaths.length];
+    for (int i = 0; i < srcPaths.length; i++) {
+      files[i] = new File(srcPaths[i]);
+    }
+
+    byte[] buffer = new byte[64 * 1024];
+    for (File file : files) {
+      try (InputStream in = new FileInputStream(file)) {
+        int n;
+        while ((n = in.read(buffer)) > 0) {
+          out.write(buffer, 0, n);
+        }
+      }
+      if (out.checkError()) {
+        throw new IOException("Failed to merge XML output file " + file);
+      }
+    }
+  }
+
+  private void cleanupTmpFiles(String[] paths) {
+    for (String path : paths) {
+      File file = new File(path);
+      if (!file.delete() && file.exists()) {
+        LOG.warn("delete tmp file: {} returned false", file);
+      }
+    }
+  }
+
   private PBImageXmlWriter o(final String e, final Object v) {
     if (v instanceof Boolean) {
       // For booleans, the presence of the element indicates true, and its

hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/HDFSCommands.md

@@ -227,7 +227,7 @@ Usage: `hdfs oiv [OPTIONS] -i INPUT_FILE`
 | `-step` *size* | Specify the granularity of the distribution in bytes (2MB by default). This option is used with FileDistribution processor. |
 | `-format` | Format the output result in a human-readable fashion rather than a number of bytes. (false by default). This option is used with FileDistribution processor. |
 | `-delimiter` *arg* | Delimiting string to use with Delimited processor. |
-| `-sp` | Whether to print Storage policy(default is false). This option is used with Delimited processor only. |
+| `-sp` | Whether to print Storage policy(default is false). This option is used with XML and Delimited processor. |
 | `-ec` | Whether to print Erasure coding policy(default is false). This option is used with Delimited processor only. |
 | `-m`,`--multiThread` *arg* | Specify multiThread to process sub-sections. This option is used with Delimited processor only. |
 | `-t`,`--temp` *temporary dir* | Use temporary dir to cache intermediate result to generate Delimited outputs. If not set, Delimited processor constructs the namespace in memory before outputting text. |