Add AUTO_RANDOM mode

gcs/CHANGES.md

@@ -1,6 +1,7 @@
 # Release Notes
 
 ## Next
+1. Add AUTO_RANDOM as new fadvise mode.
 
 ## 2.2.25 - 2024-08-01
 1. PR #1227 - Avoid registering subscriber class multiple times

gcs/CONFIGURATION.md

@@ -503,6 +503,12 @@ permissions (not authorized) to execute these requests.
         streaming requests as soon as first backward read or forward read for
         more than `fs.gs.inputstream.inplace.seek.limit` bytes was detected.
 
+    *   `AUTO_RANDOM` - in this mode connector starts with bounded range
+        requests when reading non gzip-encoded object and switches to streaming
+        request, bounded by `fs.gs.block.size`, if previous two requests follows
+        sequential read pattern i.e. forward seeks which are within
+        `fs.gs.inputstream.inplace.seek.limit`.
+
 *   `fs.gs.inputstream.inplace.seek.limit` (default: `8388608`)
 
     If forward seeks are within this many bytes of the current position, seeks

gcs/src/main/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemConfiguration.java

@@ -673,6 +673,7 @@ private static GoogleCloudStorageReadOptions getReadChannelOptions(Configuration
         .setTraceLogTimeThreshold(GCS_TRACE_LOG_TIME_THRESHOLD_MS.get(config, config::getLong))
         .setTraceLogExcludeProperties(
             ImmutableSet.copyOf(GCS_TRACE_LOG_EXCLUDE_PROPERTIES.getStringCollection(config)))
+        .setBlockSize(BLOCK_SIZE.get(config, config::getLong))
         .build();
   }
 

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageClientReadChannel.java

@@ -45,7 +45,9 @@
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.SeekableByteChannel;
 import java.util.ArrayList;
+import java.util.LinkedList;
 import java.util.List;
+import java.util.ListIterator;
 import javax.annotation.Nullable;
 
 /** Provides seekable read access to GCS via java-storage library. */
@@ -210,14 +212,41 @@
     // in-place seeks.
     private byte[] skipBuffer = null;
     private ReadableByteChannel byteChannel = null;
+    // Keeps track of distance between last 2 consecutive request.
+    private LimitedFifoQueue<Long> requestDistance = new LimitedFifoQueue<Long>(2);
+    // Keeps track of last index of last served Request.
+    private long servedRequestLastIndex = -1;
     private boolean randomAccess;
+    private boolean sequentialAccess;
+
+    /* A FIFO queue.*/
+    public class LimitedFifoQueue<E> extends LinkedList<E> {
+      private int limit;
+
+      public LimitedFifoQueue(int limit) {
+        this.limit = limit;
+      }
+
+      @Override
+      public boolean add(E o) {
+        super.add(o);
+        while (size() > limit) {
+          // removed the head of linkedList.
+          super.remove();
+        }
+        return true;
+      }
+    }
 
     public ContentReadChannel(
         GoogleCloudStorageReadOptions readOptions, StorageResourceId resourceId) {
       this.blobId =
           BlobId.of(
               resourceId.getBucketName(), resourceId.getObjectName(), resourceId.getGenerationId());
-      this.randomAccess = readOptions.getFadvise() == Fadvise.RANDOM;
+      this.randomAccess =
+          readOptions.getFadvise() == Fadvise.RANDOM
+              || readOptions.getFadvise() == Fadvise.AUTO_RANDOM;
+      this.sequentialAccess = !this.randomAccess;
     }
 
     public int readContent(ByteBuffer dst) throws IOException {
@@ -304,6 +333,7 @@
           int partialBytes = partiallyReadBytes(remainingBeforeRead, dst);
           totalBytesRead += partialBytes;
           currentPosition += partialBytes;
+          contentChannelCurrentPosition += partialBytes;
           logger.atFine().log(
               "Closing contentChannel after %s exception for '%s'.", e.getMessage(), resourceId);
           closeContentChannel();
@@ -321,12 +351,22 @@
       return partialReadBytes;
     }
 
+    private boolean shouldDetectSequentialAccess() {
+      return !gzipEncoded && !sequentialAccess && readOptions.getFadvise() == Fadvise.AUTO_RANDOM;
+    }
+
     private boolean shouldDetectRandomAccess() {
       return !gzipEncoded && !randomAccess && readOptions.getFadvise() == Fadvise.AUTO;
     }
 
     private void setRandomAccess() {
       randomAccess = true;
+      sequentialAccess = false;
+    }
+
+    private void setSequentialAccess() {
+      sequentialAccess = true;
+      randomAccess = false;
     }
 
     private ReadableByteChannel openByteChannel(long bytesToRead) throws IOException {
@@ -341,6 +381,9 @@
         return serveFooterContent();
       }
 
+      // Should be updated only if content is not served from cached footer
+      updateAccessPattern();
+
       setChannelBoundaries(bytesToRead);
 
       ReadableByteChannel readableByteChannel =
@@ -426,8 +469,14 @@
       if (gzipEncoded) {
         return objectSize;
       }
-
       long endPosition = objectSize;
+
+      if (sequentialAccess) {
+        endPosition = objectSize;
+        if (readOptions.getFadvise() == Fadvise.AUTO_RANDOM) {
+          endPosition = min(startPosition + readOptions.getBlockSize(), objectSize);
+        }
+      }
       if (randomAccess) {
         // opening a channel for whole object doesn't make sense as anyhow it will not be utilized
         // for further reads.
@@ -451,6 +500,7 @@
               "Got an exception on contentChannel.close() for '%s'; ignoring it.", resourceId);
         } finally {
           byteChannel = null;
+          servedRequestLastIndex = contentChannelCurrentPosition;
           reset();
         }
       }
@@ -521,34 +571,70 @@
       if (isInRangeSeek()) {
         skipInPlace();
       } else {
-        if (isRandomAccessPattern()) {
-          setRandomAccess();
-        }
         // close existing contentChannel as requested bytes can't be served from current
         // contentChannel;
         closeContentChannel();
       }
     }
 
+    private void updateAccessPattern() {
+      if (readOptions.getFadvise() == Fadvise.AUTO_RANDOM) {
+        if (isSequentialAccessPattern()) {
+          setSequentialAccess();
+        }
+      } else if (readOptions.getFadvise() == Fadvise.AUTO) {
+        if (isRandomAccessPattern()) {
+          setRandomAccess();
+        }
+      }
+    }
+
+    private boolean isSequentialAccessPattern() {
+      if (servedRequestLastIndex != -1) {
+        requestDistance.add(currentPosition - servedRequestLastIndex);
+      }
+
+      if (!shouldDetectSequentialAccess()) {
+        return false;
+      }
+
+      if (requestDistance.size() < 2) {
+        return false;
+      }
+
+      boolean sequentialRead = true;
+      // if more than two reads qualifies for sequential access pattern
+      ListIterator<Long> iterator = requestDistance.listIterator();
+      while (iterator.hasNext()) {
+        Long distance = iterator.next();
+        if (distance < 0 || distance > readOptions.DEFAULT_INPLACE_SEEK_LIMIT) {
+          sequentialRead = false;
+          break;
+        }
+      }
+      return sequentialRead;
+    }
+
     private boolean isRandomAccessPattern() {
       if (!shouldDetectRandomAccess()) {
         return false;
       }
-      if (currentPosition < contentChannelCurrentPosition) {
+      if (servedRequestLastIndex == -1) {
+        return false;
+      }
+
+      if (currentPosition < servedRequestLastIndex) {
         logger.atFine().log(
             "Detected backward read from %s to %s position, switching to random IO for '%s'",
-            contentChannelCurrentPosition, currentPosition, resourceId);
+            servedRequestLastIndex, currentPosition, resourceId);
         return true;
       }
-      if (contentChannelCurrentPosition >= 0
-          && contentChannelCurrentPosition + readOptions.getInplaceSeekLimit() < currentPosition) {
+      if (servedRequestLastIndex >= 0
+          && servedRequestLastIndex + readOptions.getInplaceSeekLimit() < currentPosition) {
         logger.atFine().log(
             "Detected forward read from %s to %s position over %s threshold,"
                 + " switching to random IO for '%s'",
-            contentChannelCurrentPosition,
-            currentPosition,
-            readOptions.getInplaceSeekLimit(),
-            resourceId);
+            servedRequestLastIndex, currentPosition, readOptions.getInplaceSeekLimit(), resourceId);
         return true;
       }
       return false;

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadOptions.java

@@ -32,7 +32,8 @@ public abstract class GoogleCloudStorageReadOptions {
   public enum Fadvise {
     AUTO,
     RANDOM,
-    SEQUENTIAL
+    SEQUENTIAL,
+    AUTO_RANDOM
   }
 
   public static final int DEFAULT_BACKOFF_INITIAL_INTERVAL_MILLIS = 200;
@@ -43,6 +44,7 @@ public enum Fadvise {
   public static final boolean DEFAULT_FAST_FAIL_ON_NOT_FOUND = true;
   public static final boolean DEFAULT_SUPPORT_GZIP_ENCODING = true;
   public static final long DEFAULT_INPLACE_SEEK_LIMIT = 8 * 1024 * 1024;
+  public static final long BLOCK_SIZE = 64 * 1024 * 1024;
   public static final Fadvise DEFAULT_FADVISE = Fadvise.SEQUENTIAL;
   public static final int DEFAULT_MIN_RANGE_REQUEST_SIZE = 2 * 1024 * 1024;
   public static final boolean GRPC_CHECKSUMS_ENABLED_DEFAULT = false;
@@ -75,6 +77,7 @@ public static Builder builder() {
         .setGrpcReadMessageTimeoutMillis(DEFAULT_GRPC_READ_MESSAGE_TIMEOUT_MILLIS)
         .setTraceLogEnabled(TRACE_LOGGING_ENABLED_DEFAULT)
         .setTraceLogTimeThreshold(0L)
+        .setBlockSize(BLOCK_SIZE)
         .setTraceLogExcludeProperties(ImmutableSet.of());
   }
 
@@ -133,6 +136,8 @@ public static Builder builder() {
   /** See {@link Builder#setTraceLogTimeThreshold(long)} . */
   public abstract long getTraceLogTimeThreshold();
 
+  public abstract long getBlockSize();
+
   /** Mutable builder for GoogleCloudStorageReadOptions. */
   @AutoValue.Builder
   public abstract static class Builder {
@@ -200,6 +205,9 @@ public abstract static class Builder {
      * <ul>
      *   <li>{@code AUTO} - automatically switches to {@code RANDOM} mode if backward read or
      *       forward read for more than {@link #setInplaceSeekLimit} bytes is detected.
+     *   <li>{@code AUTO_RANDOM} - Uses {@code RANDOM} to start with and automatically switches to
+     *       {@code SEQUENTIAL} mode if more than 2 requests fall within {@link
+     *       #setInplaceSeekLimit} limits.
      *   <li>{@code RANDOM} - sends HTTP requests with {@code Range} header set to greater of
      *       provided reade buffer by user.
      *   <li>{@code SEQUENTIAL} - sends HTTP requests with unbounded {@code Range} header.
@@ -242,6 +250,8 @@ public abstract static class Builder {
     /** Sets the property for gRPC read message timeout in milliseconds. */
     public abstract Builder setGrpcReadMessageTimeoutMillis(long grpcMessageTimeout);
 
+    public abstract Builder setBlockSize(long blockSize);
+
     abstract GoogleCloudStorageReadOptions autoBuild();
 
     public GoogleCloudStorageReadOptions build() {