content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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 org.apache.cassandra.io.util;
  
  import java.util.Optional;
  import java.util.function.Function;
  import java.util.function.Supplier;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.util.concurrent.RateLimiter;
  
  import org.apache.cassandra.cache.ChunkCache;
  import org.apache.cassandra.config.Config;
  import org.apache.cassandra.io.compress.BufferType;
  import org.apache.cassandra.io.compress.CompressionMetadata;
  import org.apache.cassandra.utils.NativeLibrary;
  import org.apache.cassandra.utils.Throwables;
  import org.apache.cassandra.utils.concurrent.Ref;
  import org.apache.cassandra.utils.concurrent.RefCounted;
  import org.apache.cassandra.utils.concurrent.SharedCloseableImpl;
  
  /**
   * {@link FileHandle} provides access to a file for reading, including the ones written by various {@link SequentialWriter}
   * instances, and it is typically used by {@link org.apache.cassandra.io.sstable.format.SSTableReader}.
   * <p>
   * Use {@link FileHandle.Builder} to create an instance, and call {@link #createReader()} (and its variants) to
   * access the readers for the underlying file.
   * <p>
   * You can use {@link Builder#complete()} several times during its lifecycle with different {@code overrideLength}(i.e. early opening file).
   * For that reason, the builder keeps a reference to the file channel and makes a copy for each {@link Builder#complete()} call.
   * Therefore, it is important to close the {@link Builder} when it is no longer needed, as well as any {@link FileHandle}
   * instances.
   */
  public class FileHandle extends SharedCloseableImpl
  {
      public final ChannelProxy channel;
  
      public final long onDiskLength;
  
      /*
       * Rebufferer factory to use when constructing RandomAccessReaders
       */
      private final RebuffererFactory rebuffererFactory;
  
      /*
       * Optional CompressionMetadata when dealing with compressed file
       */
      private final Optional<CompressionMetadata> compressionMetadata;
  
      private FileHandle(Cleanup cleanup,
                         ChannelProxy channel,
                         RebuffererFactory rebuffererFactory,
                         CompressionMetadata compressionMetadata,
                         long onDiskLength)
      {
          super(cleanup);
          this.rebuffererFactory = rebuffererFactory;
          this.channel = channel;
          this.compressionMetadata = Optional.ofNullable(compressionMetadata);
          this.onDiskLength = onDiskLength;
      }
  
      private FileHandle(FileHandle copy)
      {
          super(copy);
          channel = copy.channel;
          rebuffererFactory = copy.rebuffererFactory;
          compressionMetadata = copy.compressionMetadata;
          onDiskLength = copy.onDiskLength;
      }
  
      /**
       * @return file this factory is referencing
       */
      public File file()
      {
          return new File(channel.filePath());
      }
  
      public String path()
      {
          return channel.filePath();
      }
  
      public long dataLength()
      {
          return compressionMetadata.map(c -> c.dataLength).orElseGet(rebuffererFactory::fileLength);
      }
  
      public RebuffererFactory rebuffererFactory()
      {
          return rebuffererFactory;
      }
  
      public Optional<CompressionMetadata> compressionMetadata()
      {
          return compressionMetadata;
      }
  
      @Override
      public void addTo(Ref.IdentityCollection identities)
      {
          super.addTo(identities);
      }
  
      @Override
      public FileHandle sharedCopy()
      {
          return new FileHandle(this);
      }
  
      /**
       * Create {@link RandomAccessReader} with configured method of reading content of the file.
       *
       * @return RandomAccessReader for the file
       */
      public RandomAccessReader createReader()
      {
          return createReader(null);
      }
  
      public RandomAccessReader createReaderForScan()
      {
          return createReader(null, true);
      }
  
      /**
       * Create {@link RandomAccessReader} with configured method of reading content of the file.
       * Reading from file will be rate limited by given {@link RateLimiter}.
       *
       * @param limiter RateLimiter to use for rate limiting read
       * @return RandomAccessReader for the file
       */
      public RandomAccessReader createReader(RateLimiter limiter)
      {
          return createReader(limiter, false);
      }
  
      public RandomAccessReader createReader(RateLimiter limiter, boolean forScan)
      {
         return new RandomAccessReader(instantiateRebufferer(limiter, forScan));
      }
  
      public FileDataInput createReader(long position)
      {
          RandomAccessReader reader = createReader();
          try
          {
              reader.seek(position);
              return reader;
          }
          catch (Throwable t)
          {
              try
              {
                  reader.close();
              }
              catch (Throwable t2)
              {
                  t.addSuppressed(t2);
              }
              throw t;
          }
      }
  
      /**
       * Drop page cache from start to given {@code before}.
       *
       * @param before uncompressed position from start of the file to be dropped from cache. if 0, to end of file.
       */
      public void dropPageCache(long before)
      {
          long position = compressionMetadata.map(metadata -> {
              if (before >= metadata.dataLength)
                  return 0L;
              else
                  return metadata.chunkFor(before).offset;
          }).orElse(before);
          NativeLibrary.trySkipCache(channel.getFileDescriptor(), 0, position, file().absolutePath());
      }
  
      public Rebufferer instantiateRebufferer(RateLimiter limiter)
      {
          return instantiateRebufferer(limiter, false);
      }
  
      public Rebufferer instantiateRebufferer(RateLimiter limiter, boolean forScan)
      {
          Rebufferer rebufferer = rebuffererFactory.instantiateRebufferer(forScan);
  
          if (limiter != null)
              rebufferer = new LimitingRebufferer(rebufferer, limiter, DiskOptimizationStrategy.MAX_BUFFER_SIZE);
          return rebufferer;
      }
  
      /**
       * Perform clean up of all resources held by {@link FileHandle}.
       */
      private static class Cleanup implements RefCounted.Tidy
      {
          final ChannelProxy channel;
          final RebuffererFactory rebufferer;
          final CompressionMetadata compressionMetadata;
          final Optional<ChunkCache> chunkCache;
  
          private Cleanup(ChannelProxy channel,
                          RebuffererFactory rebufferer,
                          CompressionMetadata compressionMetadata,
                          ChunkCache chunkCache)
          {
              this.channel = channel;
              this.rebufferer = rebufferer;
              this.compressionMetadata = compressionMetadata;
              this.chunkCache = Optional.ofNullable(chunkCache);
          }
  
          public String name()
          {
              return channel.filePath();
          }
  
          public void tidy()
          {
              chunkCache.ifPresent(cache -> cache.invalidateFile(name()));
              try
              {
                  if (compressionMetadata != null)
                  {
                      compressionMetadata.close();
                  }
              }
              finally
              {
                  try
                  {
                      channel.close();
                  }
                  finally
                  {
                      rebufferer.close();
                  }
              }
          }
      }
  
      /**
       * Configures how the file will be read (compressed, mmapped, use cache etc.)
       */
      public static class Builder
      {
          public static final long NO_LENGTH_OVERRIDE = -1;
  
          public final File file;
  
          private CompressionMetadata compressionMetadata;
          private Supplier<Double> crcCheckChanceSupplier = () -> 1.0;
          private ChunkCache chunkCache;
          private int bufferSize = RandomAccessReader.DEFAULT_BUFFER_SIZE;
          private BufferType bufferType = BufferType.OFF_HEAP;
          private boolean mmapped = false;
          private long lengthOverride = -1;
          private MmappedRegionsCache mmappedRegionsCache;
  
          public Builder(File file)
          {
              this.file = file;
          }
  
          /**
           * Set {@link ChunkCache} to use.
           *
           * @param chunkCache ChunkCache object to use for caching
           * @return this object
           */
          public Builder withChunkCache(ChunkCache chunkCache)
          {
              this.chunkCache = chunkCache;
              return this;
          }
  
          /**
           * Provide {@link CompressionMetadata} to use when reading compressed file.
           * Upon completion, builder will create a shared copy of this object and that copy will be used in the created
           * instance of {@link FileHandle}. Therefore, the caller is responsible for closing the instance of
           * {@link CompressionMetadata} passed to this method after builder completion.
           *
           * @param metadata CompressionMetadata to use, can be {@code null} if no compression is used
           * @return this object
           */
          public Builder withCompressionMetadata(CompressionMetadata metadata)
          {
              this.compressionMetadata = metadata;
              return this;
          }
  
          public Builder withCrcCheckChance(Supplier<Double> crcCheckChanceSupplier)
          {
              this.crcCheckChanceSupplier = crcCheckChanceSupplier;
              return this;
          }
  
          /**
           * Set whether to use mmap for reading
           *
           * @param mmapped true if using mmap
           * @return this instance
           */
          public Builder mmapped(boolean mmapped)
          {
              this.mmapped = mmapped;
              return this;
          }
  
          public Builder mmapped(Config.DiskAccessMode diskAccessMode)
          {
              this.mmapped = diskAccessMode == Config.DiskAccessMode.mmap;
              return this;
          }
  
          public Builder withMmappedRegionsCache(MmappedRegionsCache mmappedRegionsCache)
          {
              this.mmappedRegionsCache = mmappedRegionsCache;
              return this;
          }
  
          /**
           * Set the buffer size to use (if appropriate).
           *
           * @param bufferSize Buffer size in bytes
           * @return this instance
           */
          public Builder bufferSize(int bufferSize)
          {
              this.bufferSize = bufferSize;
              return this;
          }
  
          /**
           * Set the buffer type (on heap or off heap) to use (if appropriate).
           *
           * @param bufferType Buffer type to use
           * @return this instance
           */
          public Builder bufferType(BufferType bufferType)
          {
              this.bufferType = bufferType;
              return this;
          }
  
          /**
           * Override the file length.
           *
           * @param lengthOverride Override file length (in bytes) so that read cannot go further than this value.
           *                       If the value is less than or equal to 0, then the value is ignored.
           * @return Built file
           */
          public Builder withLengthOverride(long lengthOverride)
          {
              this.lengthOverride = lengthOverride;
              return this;
          }
  
          /**
           * Complete building {@link FileHandle}.
           */
          public FileHandle complete()
          {
              return complete(ChannelProxy::new);
          }
  
          @VisibleForTesting
          public FileHandle complete(Function<File, ChannelProxy> channelProxyFactory)
          {
              ChannelProxy channel = null;
              MmappedRegions regions = null;
              CompressionMetadata compressionMetadata = null;
              try
              {
                  compressionMetadata = this.compressionMetadata != null ? this.compressionMetadata.sharedCopy() : null;
                  channel = channelProxyFactory.apply(file);
  
                  long fileLength = (compressionMetadata != null) ? compressionMetadata.compressedFileLength : channel.size();
                  long length = lengthOverride > 0 ? lengthOverride : fileLength;
  
                  RebuffererFactory rebuffererFactory;
                  if (length == 0)
                  {
                      rebuffererFactory = new EmptyRebufferer(channel);
                  }
                  else if (mmapped)
                  {
                      if (compressionMetadata != null)
                      {
                          regions = mmappedRegionsCache != null ? mmappedRegionsCache.getOrCreate(channel, compressionMetadata, bufferSize)
                                                                : MmappedRegions.map(channel, compressionMetadata);
                          rebuffererFactory = maybeCached(new CompressedChunkReader.Mmap(channel, compressionMetadata, regions, crcCheckChanceSupplier));
                      }
                      else
                      {
                          regions = mmappedRegionsCache != null ? mmappedRegionsCache.getOrCreate(channel, length, bufferSize)
                                                                : MmappedRegions.map(channel, length, bufferSize);
                          rebuffererFactory = new MmapRebufferer(channel, length, regions);
                      }
                  }
                  else
                  {
                      if (compressionMetadata != null)
                      {
                          rebuffererFactory = maybeCached(new CompressedChunkReader.Standard(channel, compressionMetadata, crcCheckChanceSupplier));
                      }
                      else
                      {
                          int chunkSize = DiskOptimizationStrategy.roundForCaching(bufferSize, ChunkCache.roundUp);
                          rebuffererFactory = maybeCached(new SimpleChunkReader(channel, length, bufferType, chunkSize));
                      }
                  }
                  Cleanup cleanup = new Cleanup(channel, rebuffererFactory, compressionMetadata, chunkCache);
  
                  FileHandle fileHandle = new FileHandle(cleanup, channel, rebuffererFactory, compressionMetadata, length);
                  return fileHandle;
              }
              catch (Throwable t)
              {
                  Throwables.closeNonNullAndAddSuppressed(t, regions, channel, compressionMetadata);
                  throw t;
              }
          }
  
          private RebuffererFactory maybeCached(ChunkReader reader)
          {
              if (chunkCache != null && chunkCache.capacity() > 0)
                  return chunkCache.wrap(reader);
              return reader;
          }
      }
  
      @Override
      public String toString()
      {
          return getClass().getSimpleName() + "(path='" + file() + '\'' +
                 ", length=" + rebuffererFactory.fileLength() +
                 ')';
      }
  }
  