Add fadvise: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,22 @@ 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` - It is complementing `AUTO` mode which uses sequential
+        mode to start with and adapts to bounded range requests. `AUTO_RANDOM`
+        mode uses bounded channel initially and adapts to sequential requests if
+        consecutive requests are within `fs.gs.inputstream.min.range.request.size`.
+        gzip-encode object will bypass this adoption, it will always be a
+        streaming(unbounded) channel. This helps in cases where egress limits is
+        getting breached for customer because `AUTO` mode will always lead to
+        one unbounded channel for a file. `AUTO_RANDOM` will avoid such unwanted
+        unbounded channels.
+
+*   `fs.gs.fadvise.request.track.count` (default: `3`)
+
+    Self adaptive fadvise mode uses distance between the served requests to
+    decide the access pattern. This property controls how many such requests
+    need to be tracked.
+
 *   `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

@@ -213,6 +213,13 @@ public class GoogleHadoopFileSystemConfiguration {
   public static final HadoopConfigurationProperty<Integer> GCS_BATCH_THREADS =
       new HadoopConfigurationProperty<>("fs.gs.batch.threads", 15);
 
+  /**
+   * Configuration key for number of request to track for adapting the access pattern i.e. fadvise:
+   * AUTO & AUTO_RANDOM.
+   */
+  public static final HadoopConfigurationProperty<Integer> GCS_FADVISE_REQUEST_TRACK_COUNT =
+      new HadoopConfigurationProperty<>("fs.gs.fadvise.request.track.count", 3);
+
   /**
    * Configuration key for enabling the use of Rewrite requests for copy operations. Rewrite request
    * has the same effect as Copy request, but it can handle moving large objects that may
@@ -673,6 +680,8 @@ 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))
+        .setFadviseRequestTrackCount(GCS_FADVISE_REQUEST_TRACK_COUNT.get(config, config::getInt))
         .build();
   }
 

gcs/src/test/java/com/google/cloud/hadoop/fs/gcs/GoogleHadoopFileSystemConfigurationTest.java

@@ -154,6 +154,7 @@ public class GoogleHadoopFileSystemConfigurationTest {
               "fs.gs.write.parallel.composite.upload.part.file.cleanup.type",
               PartFileCleanupType.ALWAYS);
           put("fs.gs.write.parallel.composite.upload.part.file.name.prefix", "");
+          put("fs.gs.fadvise.request.track.count", 3);
         }
       };
 

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

@@ -0,0 +1,181 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.hadoop.gcsio;
+
+import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions.Fadvise;
+import com.google.common.flogger.GoogleLogger;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.ListIterator;
+
+class AdaptiveFileAccessPattern implements Closeable {
+
+  private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();
+  private final StorageResourceId resourceId;
+  private final GoogleCloudStorageReadOptions readOptions;
+  private boolean isPatternOverriden = false;
+  private boolean randomAccess;
+  private long lastServedIndex = -1;
+  // Keeps track of distance between consecutive requests
+  private BoundedList<Long> consecutiveRequestsDistances;
+
+  @Override
+  public void close() throws IOException {
+    if (consecutiveRequestsDistances != null) {
+      consecutiveRequestsDistances = null;
+    }
+  }
+
+  class BoundedList<E> extends LinkedList<E> {
+    private int limit;
+
+    public BoundedList(int limit) {
+      this.limit = limit;
+    }
+
+    @Override
+    public boolean add(E o) {
+      super.add(o);
+      while (size() > limit) {
+        super.removeFirst();
+      }
+      return true;
+    }
+  }
+
+  public AdaptiveFileAccessPattern(
+      StorageResourceId resourceId, GoogleCloudStorageReadOptions readOptions) {
+    this.resourceId = resourceId;
+    this.readOptions = readOptions;
+    this.randomAccess =
+        readOptions.getFadvise() == Fadvise.AUTO_RANDOM
+            || readOptions.getFadvise() == Fadvise.RANDOM;
+    if (readOptions.getFadvise() == Fadvise.AUTO_RANDOM) {
+      consecutiveRequestsDistances = new BoundedList<>(readOptions.getFadviseRequestTrackCount());
+    }
+  }
+
+  public void updateLastServedIndex(long position) {
+    this.lastServedIndex = position;
+  }
+
+  public boolean isRandomAccessPattern() {
+    return randomAccess;
+  }
+
+  public void updateAccessPattern(long currentPosition) {
+    if (isPatternOverriden) {
+      logger.atFiner().log(
+          "Will bypass computing access pattern as it's overriden for resource :%s", resourceId);
+      return;
+    }
+    if (readOptions.getFadvise() == Fadvise.AUTO_RANDOM) {
+      if (isSequentialAccessPattern(currentPosition)) {
+        unsetRandomAccess();
+      }
+    } else if (readOptions.getFadvise() == Fadvise.AUTO) {
+      if (isRandomAccessPattern(currentPosition)) {
+        setRandomAccess();
+      }
+    }
+  }
+
+  /**
+   * This provides a way to override the access pattern, once overridden it will not be recomputed
+   * for adaptive fadvise types.
+   *
+   * @param pattern, true, to override with random access else false
+   */
+  public void overrideAccessPattern(boolean pattern) {
+    this.isPatternOverriden = true;
+    this.randomAccess = pattern;
+    logger.atInfo().log(
+        "Overriding the random access pattern to %s for fadvise:%s for resource: %s ",
+        pattern, readOptions.getFadvise(), resourceId);
+  }
+
+  private boolean isSequentialAccessPattern(long currentPosition) {
+    if (lastServedIndex != -1 && consecutiveRequestsDistances != null) {
+      consecutiveRequestsDistances.add(currentPosition - lastServedIndex);
+    }
+
+    if (!shouldDetectSequentialAccess()) {
+      return false;
+    }
+
+    if (consecutiveRequestsDistances.size() < readOptions.getFadviseRequestTrackCount()) {
+      return false;
+    }
+
+    ListIterator<Long> iterator = consecutiveRequestsDistances.listIterator();
+    while (iterator.hasNext()) {
+      Long distance = iterator.next();
+      if (distance < 0 || distance > readOptions.DEFAULT_INPLACE_SEEK_LIMIT) {
+        return false;
+      }
+    }
+    logger.atInfo().log(
+        "Detected %d consecutive read request within distance threshold %d with fadvise: %s switching to sequential IO for '%s'",
+        consecutiveRequestsDistances.size(),
+        readOptions.getInplaceSeekLimit(),
+        readOptions.getFadvise(),
+        resourceId);
+    return true;
+  }
+
+  private boolean isRandomAccessPattern(long currentPosition) {
+    if (!shouldDetectRandomAccess()) {
+      return false;
+    }
+    if (lastServedIndex == -1) {
+      return false;
+    }
+
+    if (currentPosition < lastServedIndex) {
+      logger.atFine().log(
+          "Detected backward read from %s to %s position, switching to random IO for '%s'",
+          lastServedIndex, currentPosition, resourceId);
+      return true;
+    }
+    if (lastServedIndex >= 0
+        && lastServedIndex + readOptions.getInplaceSeekLimit() < currentPosition) {
+      logger.atFine().log(
+          "Detected forward read from %s to %s position over %s threshold,"
+              + " switching to random IO for '%s'",
+          lastServedIndex, currentPosition, readOptions.getInplaceSeekLimit(), resourceId);
+      return true;
+    }
+    return false;
+  }
+
+  private boolean shouldDetectSequentialAccess() {
+    return randomAccess && readOptions.getFadvise() == Fadvise.AUTO_RANDOM;
+  }
+
+  private boolean shouldDetectRandomAccess() {
+    return !randomAccess && readOptions.getFadvise() == Fadvise.AUTO;
+  }
+
+  private void setRandomAccess() {
+    randomAccess = true;
+  }
+
+  private void unsetRandomAccess() {
+    randomAccess = false;
+  }
+}