HADOOP-19354. stream factories

Moves all prefetching stream related options into the prefetching stream
factory; the standard ReadOpContext removes them, so
a new PrefetchingOptions is passed around.

Stream factories can now declare how many extra shared threads they
want and whether or not to create a future pool around the bounded pool.
This is used in S3AFileSystem when creating its thread pools -this class
no longer reads in any of the prefetching options.

All tests which enable/disable prefetching, or probe for its state,
now use S3ATestUtils methods for this.
This avoids them having to now explicitly unset two properties,
set the new input stream type, and any more complications in test
setup in future.

Everything under S3AStore is a service, so service lifecycle matches everywhere
-and store just adds to the list of managed services for start/stop/close
integration.

+ adjust assertions in ITestS3AInputStreamLeakage for prefetching
+ update the prefetching.md doc for factory changs
+ javadocs
+ add string values of type names to Constants

Once the analytics stream is in, a full doc on "stream performance"
will be needed.

package for this stuff is now impl.streams

Change-Id: Id6356d2ded2c477ba16cbb9027ac0cfbece2a542

Author: steveloughran

hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml

@@ -30,7 +30,7 @@
   </Match>
   <!-- we are using completable futures, so ignore the Future which submit() returns -->
   <Match>
-    <Class name="org.apache.hadoop.fs.s3a.S3AFileSystem$InputStreamCallbacksImpl" />
+    <Class name="org.apache.hadoop.fs.s3a.impl.InputStreamCallbacksImpl" />
     <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
   </Match>
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -1561,9 +1561,31 @@ private Constants() {
   public static final String AWS_AUTH_CLASS_PREFIX = "com.amazonaws.auth";
 
 
+  /**
+   * Input stream type: {@value}.
+   */
   public static final String INPUT_STREAM_TYPE = "fs.s3a.input.stream.type";
 
-  public static final String INPUT_STREAM_TYPE_DEFAULT = "classic";
+  /**
+   * The classic input stream: {@value}.
+   */
+  public static final String INPUT_STREAM_TYPE_CLASSIC = "classic";
+
+  /**
+   * The prefetching input stream: {@value}.
+   */
+  public static final String INPUT_STREAM_TYPE_PREFETCH = "prefetch";
+
+  /**
+   * The analytics input stream: {@value}.
+   */
+  public static final String INPUT_STREAM_TYPE_ANALYTICS = "analytics";
+
+  /**
+   * The default input stream.
+   * Currently {@link #INPUT_STREAM_TYPE_CLASSIC}
+   */
+  public static final String INPUT_STREAM_TYPE_DEFAULT = INPUT_STREAM_TYPE_CLASSIC;
 
   /**
    * Controls whether the prefetching input stream is enabled.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/InputStreamType.java

@@ -52,14 +52,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nullable;
 
-import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.GetBucketLocationRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
 import software.amazon.awssdk.services.s3.model.HeadBucketResponse;
 import software.amazon.awssdk.services.s3.model.MultipartUpload;
@@ -152,11 +149,9 @@
 import org.apache.hadoop.fs.s3a.impl.StoreContextFactory;
 import org.apache.hadoop.fs.s3a.impl.UploadContentProviders;
 import org.apache.hadoop.fs.s3a.impl.CSEUtils;
-import org.apache.hadoop.fs.s3a.prefetch.PrefetchingInputStreamFactory;
-import org.apache.hadoop.fs.s3a.impl.ClassicObjectInputStreamFactory;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectReadParameters;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStreamFactory;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStreamCallbacks;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamFactory;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamCallbacks;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperations;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperationsImpl;
 import org.apache.hadoop.fs.statistics.DurationTracker;
@@ -172,7 +167,6 @@
 import org.apache.hadoop.fs.store.audit.ActiveThreadSpanSource;
 import org.apache.hadoop.fs.store.audit.AuditSpan;
 import org.apache.hadoop.fs.store.audit.AuditSpanSource;
-import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.token.DelegationTokenIssuer;
@@ -341,18 +335,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
   private ExecutorService boundedThreadPool;
   private ThreadPoolExecutor unboundedThreadPool;
 
-  // S3 reads are prefetched asynchronously using this future pool.
+  /**
+   * Future pool built on the bounded thread pool.
+   */
   private ExecutorServiceFuturePool futurePool;
 
-  // If true, the prefetching input stream is used for reads.
-  private boolean prefetchEnabled;
-
-  // Size in bytes of a single prefetch block.
-  private int prefetchBlockSize;
-
-  // Size of prefetch queue (in number of blocks).
-  private int prefetchBlockCount;
-
   private int executorCapacity;
   private long multiPartThreshold;
   public static final Logger LOG = LoggerFactory.getLogger(S3AFileSystem.class);
@@ -673,22 +660,11 @@ public void initialize(URI name, Configuration originalConf)
       dirOperationsPurgeUploads = conf.getBoolean(DIRECTORY_OPERATIONS_PURGE_UPLOADS,
           s3ExpressStore);
 
-      this.prefetchEnabled = conf.getBoolean(PREFETCH_ENABLED_KEY, PREFETCH_ENABLED_DEFAULT);
-      long prefetchBlockSizeLong =
-          longBytesOption(conf, PREFETCH_BLOCK_SIZE_KEY, PREFETCH_BLOCK_DEFAULT_SIZE, 1);
-      if (prefetchBlockSizeLong > (long) Integer.MAX_VALUE) {
-        throw new IOException("S3A prefatch block size exceeds int limit");
-      }
-      this.prefetchBlockSize = (int) prefetchBlockSizeLong;
-      this.prefetchBlockCount =
-          intOption(conf, PREFETCH_BLOCK_COUNT_KEY, PREFETCH_BLOCK_DEFAULT_COUNT, 1);
       this.isMultipartUploadEnabled = conf.getBoolean(MULTIPART_UPLOADS_ENABLED,
           DEFAULT_MULTIPART_UPLOAD_ENABLED);
       // multipart copy and upload are the same; this just makes it explicit
       this.isMultipartCopyEnabled = isMultipartUploadEnabled;
 
-      initThreadPools(conf);
-
       int listVersion = conf.getInt(LIST_VERSION, DEFAULT_LIST_VERSION);
       if (listVersion < 1 || listVersion > 2) {
         LOG.warn("Configured fs.s3a.list.version {} is invalid, forcing " +
@@ -811,6 +787,10 @@ public void initialize(URI name, Configuration originalConf)
       // directly through the client manager.
       // this is to aid mocking.
       s3Client = getStore().getOrCreateS3Client();
+
+      // thread pool init requires store to be created
+      initThreadPools();
+
       // The filesystem is now ready to perform operations against
       // S3
       // This initiates a probe against S3 for the bucket existing.
@@ -959,12 +939,15 @@ public Statistics getInstanceStatistics() {
   }
 
   /**
-   * Initialize the thread pool.
+   * Initialize the thread pools.
    * This must be re-invoked after replacing the S3Client during test
    * runs.
    * @param conf configuration.
    */
-  private void initThreadPools(Configuration conf) {
+  private void initThreadPools() {
+
+    Configuration conf = getConf();
+
     final String name = "s3a-transfer-" + getBucket();
     int maxThreads = conf.getInt(MAX_THREADS, DEFAULT_MAX_THREADS);
     if (maxThreads < 2) {
@@ -980,7 +963,9 @@ private void initThreadPools(Configuration conf) {
             TimeUnit.SECONDS,
             Duration.ZERO).getSeconds();
 
-    int numPrefetchThreads = this.prefetchEnabled ? this.prefetchBlockCount : 0;
+    final ObjectInputStreamFactory.ThreadOptions requirements =
+        getStore().prefetchThreadRequirements();
+    int numPrefetchThreads = requirements.sharedThreads();
 
     int activeTasksForBoundedThreadPool = maxThreads;
     int waitingTasksForBoundedThreadPool = maxThreads + totalTasks + numPrefetchThreads;
@@ -998,7 +983,7 @@ private void initThreadPools(Configuration conf) {
     unboundedThreadPool.allowCoreThreadTimeOut(true);
     executorCapacity = intOption(conf,
         EXECUTOR_CAPACITY, DEFAULT_EXECUTOR_CAPACITY, 1);
-    if (prefetchEnabled) {
+    if (requirements.createFuturePool()) {
       final S3AInputStreamStatistics s3AInputStreamStatistics =
           statisticsContext.newInputStreamStatistics();
       futurePool = new ExecutorServiceFuturePool(
@@ -1987,9 +1972,8 @@ protected S3AReadOpContext createReadContext(
         fileStatus,
         vectoredIOContext,
         IOStatisticsContext.getCurrentIOStatisticsContext().getAggregator(),
-        futurePool,
-        prefetchBlockSize,
-        prefetchBlockCount)
+        futurePool
+    )
         .withAuditSpan(auditSpan);
     openFileHelper.applyDefaultOptions(roc);
     return roc.build();
@@ -5519,7 +5503,7 @@ public boolean hasPathCapability(final Path path, final String capability)
 
       // stream leak detection.
     case StreamStatisticNames.STREAM_LEAKS:
-      return !prefetchEnabled;
+      return true;
 
     default:
       // is it a performance flag?

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -39,9 +39,8 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.fs.impl.LeakReporter;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStream;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectReadParameters;
-import org.apache.hadoop.fs.statistics.StreamStatisticNames;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStream;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -191,7 +190,7 @@ public S3AInputStream(ObjectReadParameters parameters) {
     S3AReadOpContext context = getContext();
     this.changeTracker = new ChangeTracker(getUri(),
         context.getChangeDetectionPolicy(),
-        getStreamStatistics().getChangeTrackerStatistics(),
+        getS3AStreamStatistics().getChangeTrackerStatistics(),
         getObjectAttributes());
     setReadahead(context.getReadahead());
     this.asyncDrainThreshold = context.getAsyncDrainThreshold();
@@ -873,7 +872,8 @@ public synchronized void readVectored(List<? extends FileRange> ranges,
 
     if (isOrderedDisjoint(sortedRanges, 1, minSeekForVectorReads())) {
       LOG.debug("Not merging the ranges as they are disjoint");
-      getS3AStreamStatistics().readVectoredOperationStarted(sortedRanges.size(), sortedRanges.size());
+      getS3AStreamStatistics().readVectoredOperationStarted(sortedRanges.size(),
+          sortedRanges.size());
       for (FileRange range: sortedRanges) {
         ByteBuffer buffer = allocate.apply(range.getLength());
         getBoundedThreadPool().submit(() -> readSingleRange(range, buffer));
@@ -883,7 +883,8 @@ public synchronized void readVectored(List<? extends FileRange> ranges,
       List<CombinedFileRange> combinedFileRanges = mergeSortedRanges(sortedRanges,
               1, minSeekForVectorReads(),
               maxReadSizeForVectorReads());
-      getS3AStreamStatistics().readVectoredOperationStarted(sortedRanges.size(), combinedFileRanges.size());
+      getS3AStreamStatistics().readVectoredOperationStarted(sortedRanges.size(),
+          combinedFileRanges.size());
       LOG.debug("Number of original ranges size {} , Number of combined ranges {} ",
               ranges.size(), combinedFileRanges.size());
       for (CombinedFileRange combinedFileRange: combinedFileRanges) {
@@ -1289,9 +1290,9 @@ public synchronized void unbuffer() {
   @Override
   public boolean hasCapability(String capability) {
     switch (toLowerCase(capability)) {
-    case StreamCapabilities.IOSTATISTICS:
     case StreamCapabilities.IOSTATISTICS_CONTEXT:
-    case StreamStatisticNames.STREAM_LEAKS:
+    case StreamCapabilities.READAHEAD:
+    case StreamCapabilities.UNBUFFER:
       return true;
     default:
       return super.hasCapability(capability);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java

@@ -75,15 +75,11 @@ public class S3AReadOpContext extends S3AOpContext {
   /** Thread-level IOStatistics aggregator. **/
   private final IOStatisticsAggregator ioStatisticsAggregator;
 
-  // S3 reads are prefetched asynchronously using this future pool.
+  /**
+   * Pool for any future IO.
+   */
   private ExecutorServiceFuturePool futurePool;
 
-  // Size in bytes of a single prefetch block.
-  private final int prefetchBlockSize;
-
-  // Size of prefetch queue (in number of blocks).
-  private final int prefetchBlockCount;
-
   /**
    * Instantiate.
    * @param path path of read
@@ -93,9 +89,7 @@ public class S3AReadOpContext extends S3AOpContext {
    * @param dstFileStatus target file status
    * @param vectoredIOContext context for vectored read operation.
    * @param ioStatisticsAggregator IOStatistics aggregator for each thread.
-   * @param futurePool the ExecutorServiceFuturePool instance used by async prefetches.
-   * @param prefetchBlockSize the size (in number of bytes) of each prefetched block.
-   * @param prefetchBlockCount maximum number of prefetched blocks.
+   * @param futurePool Pool for any future IO
    */
   public S3AReadOpContext(
       final Path path,
@@ -105,22 +99,15 @@ public S3AReadOpContext(
       FileStatus dstFileStatus,
       VectoredIOContext vectoredIOContext,
       IOStatisticsAggregator ioStatisticsAggregator,
-      ExecutorServiceFuturePool futurePool,
-      int prefetchBlockSize,
-      int prefetchBlockCount) {
+      ExecutorServiceFuturePool futurePool) {
 
     super(invoker, stats, instrumentation,
         dstFileStatus);
     this.path = requireNonNull(path);
     this.vectoredIOContext = requireNonNull(vectoredIOContext, "vectoredIOContext");
     this.ioStatisticsAggregator = ioStatisticsAggregator;
     this.futurePool = futurePool;
-    Preconditions.checkArgument(
-        prefetchBlockSize > 0, "invalid prefetchBlockSize %d", prefetchBlockSize);
-    this.prefetchBlockSize = prefetchBlockSize;
-    Preconditions.checkArgument(
-        prefetchBlockCount > 0, "invalid prefetchBlockCount %d", prefetchBlockCount);
-    this.prefetchBlockCount = prefetchBlockCount;
+
   }
 
   /**
@@ -265,23 +252,6 @@ public ExecutorServiceFuturePool getFuturePool() {
     return this.futurePool;
   }
 
-  /**
-   * Gets the size in bytes of a single prefetch block.
-   *
-   * @return the size in bytes of a single prefetch block.
-   */
-  public int getPrefetchBlockSize() {
-    return this.prefetchBlockSize;
-  }
-
-  /**
-   * Gets the size of prefetch queue (in number of blocks).
-   *
-   * @return the size of prefetch queue (in number of blocks).
-   */
-  public int getPrefetchBlockCount() {
-    return this.prefetchBlockCount;
-  }
 
   @Override
   public String toString() {

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AReadOpContext.java

@@ -46,16 +46,14 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.s3a.api.RequestFactory;
 import org.apache.hadoop.fs.s3a.impl.ChangeTracker;
 import org.apache.hadoop.fs.s3a.impl.ClientManager;
 import org.apache.hadoop.fs.s3a.impl.MultiObjectDeleteException;
 import org.apache.hadoop.fs.s3a.impl.S3AFileSystemOperations;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectInputStreamFactory;
-import org.apache.hadoop.fs.s3a.impl.model.ObjectReadParameters;
+import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamFactory;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
@@ -76,7 +74,11 @@
  */
 @InterfaceAudience.LimitedPrivate("Extensions")
 @InterfaceStability.Unstable
-public interface S3AStore extends Service, IOStatisticsSource, ClientManager, ObjectInputStreamFactory {
+public interface S3AStore extends
+    ClientManager,
+    IOStatisticsSource,
+    ObjectInputStreamFactory,
+    Service {
 
   /**
    * Acquire write capacity for operations.