print stream capabilities of input and output streams

...inc latest capabilities for iocontext and vectored io

Author: steveloughran

…e/hadoop/fs/store/CapabilityChecker.java …doop/fs/store/PathCapabilityChecker.javasrc/main/java/org/apache/hadoop/fs/store/CapabilityChecker.java renamed to src/main/java/org/apache/hadoop/fs/store/PathCapabilityChecker.java src/main/java/org/apache/hadoop/fs/store/CapabilityChecker.java renamed to src/main/java/org/apache/hadoop/fs/store/PathCapabilityChecker.java

@@ -22,23 +22,22 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.util.VersionInfo;
 
 import static org.apache.hadoop.fs.store.StoreExitCodes.E_EXCEPTION_THROWN;
 import static org.apache.hadoop.fs.store.StoreExitCodes.E_UNSUPPORTED_VERSION;
 
-public class CapabilityChecker {
+public class PathCapabilityChecker {
 
   private final Method hasPathCapability;
-  private final FileSystem fileSystem;
+  private final Object source;
 
-  public CapabilityChecker(FileSystem fileSystem) {
-    this.fileSystem = fileSystem;
+  public PathCapabilityChecker(Object source) {
+    this.source = source;
     Method method;
     try {
-      method = FileSystem.class.getMethod("hasPathCapability",
+      method = source.getClass().getMethod("hasPathCapability",
           Path.class, String.class);
     } catch (NoSuchMethodException e) {
       method = null;
@@ -51,7 +50,7 @@ public boolean methodAvailable() {
   }
 
   /**
-   * Does a filesystem have a capability?
+   * Does an object have a capability?
    * uses reflection so the jar can compile/run against
    * older hadoop releases.
    * throws StoreExitException(E_UNSUPPORTED_VERSION) if the api isn't found.
@@ -69,7 +68,7 @@ public boolean hasPathCapability(Path path, String capability)
               + VersionInfo.getVersion());
     }
     try {
-      return (Boolean) hasPathCapability.invoke(fileSystem, path, capability);
+      return (Boolean) hasPathCapability.invoke(source, path, capability);
     } catch (IllegalAccessException e) {
       throw new StoreExitException(E_UNSUPPORTED_VERSION,
           "Hadoop version does not support PathCapabilities: "

src/main/java/org/apache/hadoop/fs/store/StoreDurationInfo.java

@@ -112,7 +112,7 @@ public String toString() {
   public void close() {
     finished();
     if (log != null) {
-      log.info(text + ": duration " + this);
+      log.info("Duration of {}: {}", text, this);
     }
   }
 }

src/main/java/org/apache/hadoop/fs/store/commands/PathCapability.java

@@ -26,7 +26,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.store.CapabilityChecker;
+import org.apache.hadoop.fs.store.PathCapabilityChecker;
 import org.apache.hadoop.fs.store.StoreEntryPoint;
 import org.apache.hadoop.util.ToolRunner;
 
@@ -79,7 +79,7 @@ public int run(String[] args) throws Exception {
     FileSystem fs = path.getFileSystem(conf);
     println("Using filesystem %s", fs.getUri());
     Path absPath = path.makeQualified(fs.getUri(), fs.getWorkingDirectory());
-    if (new CapabilityChecker(fs).
+    if (new PathCapabilityChecker(fs).
         hasPathCapability(absPath, capability)) {
 
       println("Path %s has capability %s",

src/main/java/org/apache/hadoop/fs/store/diag/CapabilityKeys.java

@@ -18,7 +18,7 @@
 
 package org.apache.hadoop.fs.store.diag;
 
-import org.apache.hadoop.classification.InterfaceStability;
+
 
 /**
  * Common path capabilities.
@@ -129,7 +129,6 @@ private CapabilityKeys() {
   /**
    * Probe for support for BatchListingOperations.
    */
-  @InterfaceStability.Unstable
   public static final String FS_EXPERIMENTAL_BATCH_LISTING =
       "fs.capability.batch.listing";
 
@@ -179,11 +178,13 @@ private CapabilityKeys() {
    */
   public static final String STORE_CAPABILITY_MAGIC_COMMITTER
       = "fs.s3a.capability.magic.committer";
+
   /**
    * Does the FS Support S3 Select?
    * Value: {@value}.
    */
-  public static final String S3_SELECT_CAPABILITY = "fs.s3a.capability.select.sql";
+  public static final String S3_SELECT_CAPABILITY =
+      "fs.s3a.capability.select.sql";
 
   /**
    * {@code PathCapabilities} probe to indicate that the filesystem
@@ -241,7 +242,8 @@ private CapabilityKeys() {
    * {code createFile()} builder.
    * Value {@value}.
    */
-  public static final String FS_S3A_CREATE_PERFORMANCE = "fs.s3a.create.performance";
+  public static final String FS_S3A_CREATE_PERFORMANCE =
+      "fs.s3a.create.performance";
 
   /**
    * Prefix for adding a header to the object when created.
@@ -252,4 +254,91 @@ private CapabilityKeys() {
    */
   public static final String FS_S3A_CREATE_HEADER = "fs.s3a.create.header";
 
+
+  /**
+   * Streams that support IOStatistics context and capture thread-level
+   * IOStatistics.
+   */
+  public static String IOSTATISTICS_CONTEXT = "fs.capability.iocontext.supported";
+
+
+  /**
+   * IOStatisticsSource API.
+   */
+  public static String IOSTATISTICS = "iostatistics";
+
+  /**
+   * Support for vectored IO api.
+   * See {@code PositionedReadable#readVectored(List, IntFunction)}.
+   */
+  public static String VECTOREDIO = "readvectored";
+
+
+  /**
+   * Stream setReadahead capability implemented by
+   * {@code CanSetReadahead#setReadahead(Long)}.
+   */
+  public static String READAHEAD = "in:readahead";
+
+  /**
+   * Stream setDropBehind capability implemented by
+   * {@code CanSetDropBehind#setDropBehind(Boolean)}.
+   */
+  public static String DROPBEHIND = "dropbehind";
+
+  /**
+   * Stream unbuffer capability implemented by {@code CanUnbuffer#unbuffer()}.
+   */
+  public static String UNBUFFER = "in:unbuffer";
+
+  /**
+   * Stream read(ByteBuffer) capability implemented by
+   * {@code ByteBufferReadable#read(java.nio.ByteBuffer)}.
+   */
+  public static String READBYTEBUFFER = "in:readbytebuffer";
+
+  /**
+   * Stream read(long, ByteBuffer) capability implemented by
+   * {@code ByteBufferPositionedReadable#read(long, java.nio.ByteBuffer)}.
+   */
+  public static String PREADBYTEBUFFER = "in:preadbytebuffer";
+
+  /**
+   * Stream hflush capability implemented by {@code Syncable#hflush()}.
+   *
+   * Use the {@code #HSYNC} probe to check for the support of Syncable;
+   * it's that presence of {@code hsync()} which matters.
+   */
+  @Deprecated
+  public static String HFLUSH = "hflush";
+
+  /**
+   * Stream hsync capability implemented by {@code Syncable#hsync()}.
+   */
+  public static String HSYNC = "hsync";
+  
+  
+  /**
+   * Set of input/output stream capabilities to scan for.
+   */
+  public static String[] INPUTSTREAM_CAPABILITIES = {
+      DROPBEHIND,
+      IOSTATISTICS,
+      IOSTATISTICS_CONTEXT,
+      PREADBYTEBUFFER,
+      READAHEAD,
+      READBYTEBUFFER,
+      UNBUFFER,
+      VECTOREDIO,
+  };
+  /**
+   * Set of input/output stream capabilities to scan for.
+   */
+  public static String[] OUTPUTSTREAM_CAPABILITIES = {
+      ABORTABLE_STREAM,
+      HFLUSH,
+      HSYNC,
+      IOSTATISTICS,
+      IOSTATISTICS_CONTEXT,
+  };
 }

src/main/java/org/apache/hadoop/fs/store/diag/StoreDiag.java

@@ -62,8 +62,9 @@
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.UnsupportedFileSystemException;
-import org.apache.hadoop.fs.store.CapabilityChecker;
+import org.apache.hadoop.fs.store.PathCapabilityChecker;
 import org.apache.hadoop.fs.store.StoreDurationInfo;
 import org.apache.hadoop.fs.store.StoreExitException;
 import org.apache.hadoop.io.IOUtils;
@@ -705,7 +706,7 @@ public void executeFileSystemOperations(final Path baseDir,
 
     final String[] pathCapabilites = storeInfo.getOptionalPathCapabilites();
     if (pathCapabilites.length > 0) {
-      final CapabilityChecker checker = new CapabilityChecker(fs);
+      final PathCapabilityChecker checker = new PathCapabilityChecker(fs);
       if (checker.methodAvailable()) {
         heading("Path Capabilities");
         for (String s : pathCapabilites) {
@@ -763,32 +764,17 @@ public void executeFileSystemOperations(final Path baseDir,
       println("Directory %s does not exist", baseDir);
       baseDirFound = false;
     }
+    heading("Attempt to read a file");
 
     if (firstFile != null) {
       // found a file to read
-      Path firstFilePath = firstFile.getPath();
-      heading("reading file %s", firstFilePath);
-      FSDataInputStream in = null;
-      try (StoreDurationInfo ignored = new StoreDurationInfo(LOG,
-          "Reading file %s", firstFilePath)) {
-        in = fs.open(firstFilePath);
-        // read the first char or -1
-        int c = in.read();
-        println("First character of file %s is 0x%02x: '%s'",
-            firstFilePath,
-            c,
-            (c > ' ') ? Character.toString((char) c) : "(n/a)");
-        in.close();
-      } catch(FileNotFoundException ex) {
-        warn("file %s: not found/readable %s", firstFilePath, ex);
-      } catch(AccessDeniedException ex) {
-        warn("client lacks access to file %s: %s", firstFilePath, ex);
-        accessDenied = true;
-      } finally {
-        IOUtils.closeStream(in);
-      }
+      accessDenied = readFile(fs, firstFile.getPath());
+    } else {
+      println("no file found to attempt to read");
     }
 
+    heading("listfiles(%s, true)", baseDir);
+
     // now work with the full path
     limit = LIST_LIMIT;
     try(StoreDurationInfo ignored = new StoreDurationInfo(LOG,
@@ -920,9 +906,11 @@ public void executeFileSystemOperations(final Path baseDir,
       verifyPathNotFound(fs, file);
 
       try (StoreDurationInfo ignored = new StoreDurationInfo(LOG,
-          "creating a file %s", file)) {
+          "Creating file %s", file)) {
         FSDataOutputStream data = fs.create(file, true);
         data.writeUTF(HELLO);
+        printStreamCapabilities(data, CapabilityKeys.OUTPUTSTREAM_CAPABILITIES);
+
         data.close();
         println("Output stream summary: %s", data);
       }
@@ -932,8 +920,10 @@ public void executeFileSystemOperations(final Path baseDir,
       }
       FSDataInputStream in = null;
       try (StoreDurationInfo ignored = new StoreDurationInfo(LOG,
-          "Reading a file %s", file)) {
+          "Reading file %s", file)) {
         in = fs.open(file);
+        printStreamCapabilities(in, CapabilityKeys.INPUTSTREAM_CAPABILITIES);
+
         String utf = in.readUTF();
         in.close();
         println("input stream summary: %s", in);
@@ -974,6 +964,42 @@ public void executeFileSystemOperations(final Path baseDir,
     }
   }
 
+  /**
+   * Read a file.
+   */
+  private boolean readFile(final FileSystem fs,
+      final Path path) throws IOException {
+    boolean accessWasDenied = false;
+    try (StoreDurationInfo ignored = new StoreDurationInfo(LOG,
+        "Reading file %s", path);
+         FSDataInputStream in = fs.open(path)) {
+      // read the first char or -1
+      int c = in.read();
+      println("First character of file %s is 0x%02x: '%s'",
+          path,
+          c,
+          (c > ' ') ? Character.toString((char) c) : "(n/a)");
+      printStreamCapabilities(in, CapabilityKeys.INPUTSTREAM_CAPABILITIES);
+      println("Stream summary: %s", in);
+    } catch(FileNotFoundException ex) {
+      warn("file %s: not found/readable %s", path, ex);
+    } catch(AccessDeniedException ex) {
+      warn("client lacks access to file %s: %s", path, ex);
+      accessWasDenied = true;
+    }
+    return accessWasDenied;
+  }
+
+  private void printStreamCapabilities(final StreamCapabilities in,
+      final String[] capabilities) {
+    println("Capabilities:");
+    for (String s : capabilities) {
+      if (in.hasCapability(s)) {
+        println("    %s", s);
+      }
+    }
+  }
+
   public void deleteDir(final FileSystem fs, final Path dir) {
     try (StoreDurationInfo ignored = new StoreDurationInfo(LOG,
         "delete directory %s", dir)) {