From 6f5fe8c06d5831daae1f1a4f2412f3185798dca0 Mon Sep 17 00:00:00 2001 From: samlightfoot Date: Fri, 16 Jan 2026 10:43:08 +0000 Subject: [PATCH] Add Direct IO support for compaction reads This change adds support for using O_DIRECT (direct I/O) when reading SSTables during compaction. Direct I/O bypasses the OS page cache, which can be beneficial for compaction workloads where data is typically read once and not accessed again soon after. Key changes: - Add DiskAccessMode.direct option for scan operations - Introduce DirectThreadLocalByteBufferHolder and DirectThreadLocalReadAheadBuffer for aligned buffer management required by O_DIRECT - Add ByteBufferHolder interface to abstract buffer management - Modify FileHandle to support building direct I/O readers via toBuilder() - Add startup check to verify Direct IO support on configured data directories - Enable read-ahead for uncompressed data in Direct CompressedChunkReader - Move direct IO support logic into FileHandle (from SSTableReader) - Add comprehensive tests for direct I/O chunk readers and buffer holders patch by Sam Lightfoot; reviewed by Ariel Weisberg and Maxwell Guo for CASSANDRA-19987 --- CHANGES.txt | 1 + conf/cassandra.yaml | 5 + conf/cassandra_latest.yaml | 5 + .../org/apache/cassandra/config/Config.java | 1 + .../cassandra/config/DatabaseDescriptor.java | 31 +- .../AbstractCompactionStrategy.java | 3 +- .../db/compaction/CompactionManager.java | 4 +- .../compaction/LeveledCompactionStrategy.java | 7 +- .../cassandra/index/accord/SSTableIndex.java | 2 +- .../sai/disk/format/IndexDescriptor.java | 4 +- .../index/sai/disk/io/IndexFileUtils.java | 13 +- .../io/sstable/format/IndexComponent.java | 8 +- .../io/sstable/format/SSTableReader.java | 50 +- .../sstable/format/SSTableSimpleScanner.java | 6 +- .../SortedTableReaderLoadingBuilder.java | 2 +- .../io/sstable/format/SortedTableWriter.java | 27 +- .../bti/BtiTableReaderLoadingBuilder.java | 8 +- .../cassandra/io/util/ByteBufferHolder.java | 28 + .../cassandra/io/util/ChannelProxy.java | 36 +- .../io/util/CompressedChunkReader.java | 155 ++++- .../DirectThreadLocalByteBufferHolder.java | 79 +++ .../DirectThreadLocalReadAheadBuffer.java | 49 ++ .../apache/cassandra/io/util/FileHandle.java | 143 ++++- .../apache/cassandra/io/util/FileUtils.java | 54 +- .../cassandra/io/util/RandomAccessReader.java | 26 + .../io/util/ThreadLocalByteBufferHolder.java | 8 +- .../io/util/ThreadLocalReadAheadBuffer.java | 39 +- .../cassandra/service/StartupChecks.java | 55 +- .../distributed/test/FailingRepairTest.java | 11 + .../format/ForwardingSSTableReader.java | 13 + .../commitlog/DirectIOSegmentBytemanTest.java | 2 +- .../CheckpointIntervalArrayIndexTest.java | 2 +- .../CompressedRandomAccessReaderTest.java | 28 +- .../format/bti/PartitionIndexTest.java | 4 +- .../io/sstable/format/bti/RowIndexTest.java | 4 +- .../util/CompressedChunkReaderTestBase.java | 61 ++ .../util/DirectCompressedChunkReaderTest.java | 545 ++++++++++++++++++ ...DirectThreadLocalByteBufferHolderTest.java | 101 ++++ .../DirectThreadLocalReadAheadBufferTest.java | 64 ++ .../io/util/RandomAccessReaderTest.java | 26 +- ...=> StandardCompressedChunkReaderTest.java} | 88 +-- .../util/ThreadLocalReadAheadBufferTest.java | 91 +-- .../cassandra/service/StartupChecksTest.java | 9 + 43 files changed, 1665 insertions(+), 233 deletions(-) create mode 100644 src/java/org/apache/cassandra/io/util/ByteBufferHolder.java create mode 100644 src/java/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolder.java create mode 100644 src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java create mode 100644 test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTestBase.java create mode 100644 test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java create mode 100644 test/unit/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolderTest.java create mode 100644 test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java rename test/unit/org/apache/cassandra/io/util/{CompressedChunkReaderTest.java => StandardCompressedChunkReaderTest.java} (52%) diff --git a/CHANGES.txt b/CHANGES.txt index e6fa54b38f..3233ffe7bf 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Direct I/O support for compaction reads (CASSANDRA-19987) * Support custom StartupCheck implementations via SPI (CASSANDRA-21093) * Make sstableexpiredblockers support human-readable output with SSTable sizes (CASSANDRA-20448) * Enhance nodetool compactionhistory to report more compaction properities (CASSANDRA-20081) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index bbb6d1d044..a1be9ce1a9 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -688,6 +688,11 @@ commitlog_segment_size: 32MiB # The default setting is legacy when the storage compatibility is set to 4 or auto otherwise. commitlog_disk_access_mode: legacy +# Set the disk access mode for reading SSTables during compaction. The allowed values are: +# - auto: inherit from disk_access_mode (default) +# - direct: use direct I/O for compaction reads, bypassing the OS page cache +# compaction_read_disk_access_mode: auto + # Compression to apply to SSTables as they flush for compressed tables. # Note that tables without compression enabled do not respect this flag. # diff --git a/conf/cassandra_latest.yaml b/conf/cassandra_latest.yaml index 891570eb12..90d7ea0d1d 100644 --- a/conf/cassandra_latest.yaml +++ b/conf/cassandra_latest.yaml @@ -695,6 +695,11 @@ commitlog_segment_size: 32MiB # The default setting is legacy when the storage compatibility is set to 4 or auto otherwise. commitlog_disk_access_mode: auto +# Set the disk access mode for reading SSTables during compaction. The allowed values are: +# - auto: inherit from disk_access_mode (default) +# - direct: use direct I/O for compaction reads, bypassing the OS page cache +# compaction_read_disk_access_mode: auto + # Compression to apply to SSTables as they flush for compressed tables. # Note that tables without compression enabled do not respect this flag. # diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index fafcddddc6..43cdf116ea 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -428,6 +428,7 @@ public class Config public FlushCompression flush_compression = FlushCompression.fast; public int commitlog_max_compression_buffers_in_pool = 3; public DiskAccessMode commitlog_disk_access_mode = DiskAccessMode.legacy; + public DiskAccessMode compaction_read_disk_access_mode = DiskAccessMode.auto; @Replaces(oldName = "periodic_commitlog_sync_lag_block_in_ms", converter = Converters.MILLIS_DURATION_INT, deprecated = true) public DurationSpec.IntMillisecondsBound periodic_commitlog_sync_lag_block; public TransparentDataEncryptionOptions transparent_data_encryption_options = new TransparentDataEncryptionOptions(); diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index bd8542cd73..a40db14a6e 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -221,6 +221,8 @@ public class DatabaseDescriptor private static DiskAccessMode commitLogWriteDiskAccessMode; + private static DiskAccessMode compactionReadDiskAccessMode; + private static AbstractCryptoProvider cryptoProvider; private static IAuthenticator authenticator; private static IAuthorizer authorizer; @@ -662,6 +664,21 @@ public class DatabaseDescriptor } logger.info("DiskAccessMode is {}, indexAccessMode is {}", conf.disk_access_mode, indexAccessMode); + if (DiskAccessMode.auto == conf.compaction_read_disk_access_mode) + { + compactionReadDiskAccessMode = conf.disk_access_mode; + } + else if (DiskAccessMode.direct == conf.compaction_read_disk_access_mode) + { + compactionReadDiskAccessMode = DiskAccessMode.direct; + } + else + { + throw new IllegalArgumentException("Unsupported disk access mode for compaction_read_disk_access_mode " + + "(options: direct/auto) " + conf.compaction_read_disk_access_mode); + } + logger.info("compaction_read_disk_access_mode resolved to: {}", compactionReadDiskAccessMode); + /* phi convict threshold for FailureDetector */ if (conf.phi_convict_threshold < 5 || conf.phi_convict_threshold > 16) { @@ -1773,7 +1790,7 @@ public class DatabaseDescriptor File commitLogLocationDir = new File(commitLogLocation); PathUtils.createDirectoriesIfNotExists(commitLogLocationDir.toPath()); - directIOSupported = FileUtils.getBlockSize(commitLogLocationDir) > 0; + directIOSupported = FileUtils.isDirectIOSupported(commitLogLocationDir); } catch (IOError | ConfigurationException ex) { @@ -3301,6 +3318,18 @@ public class DatabaseDescriptor conf.commitlog_segment_size = new DataStorageSpec.IntMebibytesBound(sizeMebibytes); } + public static DiskAccessMode getCompactionReadDiskAccessMode() + { + return compactionReadDiskAccessMode; + } + + @VisibleForTesting + public static void setCompactionReadDiskAccessMode(DiskAccessMode scanDiskAccessMode) + { + compactionReadDiskAccessMode = scanDiskAccessMode; + conf.compaction_read_disk_access_mode = scanDiskAccessMode; + } + /** * Return commitlog disk access mode. */ diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java index f52d9f7b07..b93b5bf66b 100644 --- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java @@ -32,6 +32,7 @@ import com.google.common.collect.ImmutableMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.SerializationHeader; @@ -262,7 +263,7 @@ public abstract class AbstractCompactionStrategy try { for (SSTableReader sstable : sstables) - scanners.add(sstable.getScanner(ranges)); + scanners.add(sstable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode())); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 0c0daa5f62..91bdbd653e 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -1767,7 +1767,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan { rangesToScan = Collections2.filter(ranges, range -> !transientRanges.contains(range)); } - return sstable.getScanner(rangesToScan); + return sstable.getScanner(rangesToScan, DatabaseDescriptor.getCompactionReadDiskAccessMode()); } @Override @@ -1790,7 +1790,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan @Override public ISSTableScanner getScanner(SSTableReader sstable) { - return sstable.getScanner(); + return sstable.getScanner(DatabaseDescriptor.getCompactionReadDiskAccessMode()); } @Override diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java index 438c9da182..a22f06e562 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java @@ -44,6 +44,7 @@ import com.google.common.primitives.Doubles; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.db.rows.UnfilteredRowIterator; @@ -343,7 +344,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy { // L0 makes no guarantees about overlapping-ness. Just create a direct scanner for each for (SSTableReader sstable : byLevel.get(level)) - scanners.add(sstable.getScanner(ranges)); + scanners.add(sstable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode())); } else { @@ -445,7 +446,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy sstableIterator = this.sstables.iterator(); assert sstableIterator.hasNext(); // caller should check intersecting first SSTableReader currentSSTable = sstableIterator.next(); - currentScanner = currentSSTable.getScanner(ranges); + currentScanner = currentSSTable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode()); } @Override @@ -498,7 +499,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy return endOfData(); } SSTableReader currentSSTable = sstableIterator.next(); - currentScanner = currentSSTable.getScanner(ranges); + currentScanner = currentSSTable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode()); } } diff --git a/src/java/org/apache/cassandra/index/accord/SSTableIndex.java b/src/java/org/apache/cassandra/index/accord/SSTableIndex.java index eb9845002d..f61bb2ad75 100644 --- a/src/java/org/apache/cassandra/index/accord/SSTableIndex.java +++ b/src/java/org/apache/cassandra/index/accord/SSTableIndex.java @@ -76,7 +76,7 @@ public class SSTableIndex extends SharedCloseableImpl { Map files = new EnumMap<>(IndexComponent.class); for (IndexComponent c : id.getLiveComponents()) - files.put(c, new FileHandle.Builder(id.fileFor(c)).mmapped(true).complete()); + files.put(c, new FileHandle.Builder(id.fileFor(c)).mmapped().complete()); List segments = RouteIndexFormat.readSegments(files); files.remove(IndexComponent.SEGMENT).close(); files.remove(IndexComponent.METADATA).close(); diff --git a/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java b/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java index 21b5ef78b3..4d94084cc3 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java +++ b/src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java @@ -269,7 +269,7 @@ public class IndexDescriptor if (logger.isTraceEnabled()) logger.trace(logMessage("Opening file handle for {} ({})"), file, FBUtilities.prettyPrintMemory(file.length())); - return new FileHandle.Builder(file).mmapped(true).complete(); + return new FileHandle.Builder(file).mmapped().complete(); } catch (Throwable t) { @@ -291,7 +291,7 @@ public class IndexDescriptor if (logger.isTraceEnabled()) logger.trace(logMessage("Opening file handle for {} ({})"), file, FBUtilities.prettyPrintMemory(file.length())); - return new FileHandle.Builder(file).mmapped(true).complete(); + return new FileHandle.Builder(file).mmapped().complete(); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java b/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java index c3501fb23b..6d70ff5dce 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java +++ b/src/java/org/apache/cassandra/index/sai/disk/io/IndexFileUtils.java @@ -84,9 +84,16 @@ public class IndexFileUtils public IndexInput openBlockingInput(File file) { FileHandle fileHandle = new FileHandle.Builder(file).complete(); - RandomAccessReader randomReader = fileHandle.createReader(); - - return IndexInputReader.create(randomReader, fileHandle::close); + try + { + RandomAccessReader randomReader = fileHandle.createReader(); + return IndexInputReader.create(randomReader, fileHandle::close); + } + catch (Throwable t) + { + fileHandle.close(); + throw t; + } } public static ChecksumIndexInput getBufferedChecksumIndexInput(IndexInput indexInput) diff --git a/src/java/org/apache/cassandra/io/sstable/format/IndexComponent.java b/src/java/org/apache/cassandra/io/sstable/format/IndexComponent.java index 45dfc62b2c..c60205ea0a 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/IndexComponent.java +++ b/src/java/org/apache/cassandra/io/sstable/format/IndexComponent.java @@ -29,8 +29,12 @@ public class IndexComponent { public static FileHandle.Builder fileBuilder(File file, IOOptions ioOptions, ChunkCache chunkCache) { - return new FileHandle.Builder(file).withChunkCache(chunkCache) - .mmapped(ioOptions.indexDiskAccessMode); + FileHandle.Builder builder = new FileHandle.Builder(file).withChunkCache(chunkCache); + + if (ioOptions.indexDiskAccessMode != null) + builder.withDiskAccessMode(ioOptions.indexDiskAccessMode); + + return builder; } public static FileHandle.Builder fileBuilder(Component component, SSTable ssTable) diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index 7d8941d80c..37e73d47b4 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -115,6 +115,8 @@ import org.apache.cassandra.utils.concurrent.SharedCloseable; import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; +import static org.apache.cassandra.config.Config.DiskAccessMode; +import static org.apache.cassandra.io.util.FileHandle.OnReaderClose; import static org.apache.cassandra.utils.TimeUUID.unixMicrosToRawTimestamp; import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue; import static org.apache.cassandra.utils.concurrent.SharedCloseable.sharedCopyOrNull; @@ -365,7 +367,7 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, public static SSTableReader open(SSTable.Owner owner, Descriptor desc, TableMetadataRef metadata) { - return open(owner, desc, null, metadata); + return open(owner, desc, null, metadata); } public static SSTableReader open(SSTable.Owner owner, Descriptor descriptor, Set components, TableMetadataRef metadata) @@ -1074,12 +1076,17 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, * @return A Scanner over the full content of the SSTable. */ public ISSTableScanner getScanner() + { + return getScanner(dfile.diskAccessMode()); + } + + public ISSTableScanner getScanner(DiskAccessMode diskAccessMode) { PartitionPositionBounds fullRange = getPositionsForFullRange(); if (fullRange != null) - return new SSTableSimpleScanner(this, Collections.singletonList(fullRange)); + return new SSTableSimpleScanner(this, Collections.singletonList(fullRange), diskAccessMode); else - return new SSTableSimpleScanner(this, Collections.emptyList()); + return new SSTableSimpleScanner(this, Collections.emptyList(), diskAccessMode); } /** @@ -1089,11 +1096,16 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, * @return A Scanner for seeking over the rows of the SSTable. */ public ISSTableScanner getScanner(Collection> ranges) + { + return getScanner(ranges, dfile.diskAccessMode()); + } + + public ISSTableScanner getScanner(Collection> ranges, DiskAccessMode diskAccessMode) { if (ranges != null) - return new SSTableSimpleScanner(this, getPositionsForRanges(ranges)); + return new SSTableSimpleScanner(this, getPositionsForRanges(ranges), diskAccessMode); else - return getScanner(); + return getScanner(diskAccessMode); } /** @@ -1104,13 +1116,13 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, */ public ISSTableScanner getScanner(Iterator> boundsIterator) { - return new SSTableSimpleScanner(this, getPositionsForBoundsIterator(boundsIterator)); + return new SSTableSimpleScanner(this, getPositionsForBoundsIterator(boundsIterator), dfile.diskAccessMode()); } public ISSTableScanner getScanner(AbstractBounds bounds) { PartitionPositionBounds positionBounds = getPositionsForBounds(bounds); - return new SSTableSimpleScanner(this, positionBounds == null ? Collections.emptyList() : Collections.singletonList(positionBounds)); + return new SSTableSimpleScanner(this, positionBounds == null ? Collections.emptyList() : Collections.singletonList(positionBounds), dfile.diskAccessMode()); } @@ -1418,7 +1430,29 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, public RandomAccessReader openDataReaderForScan() { - return dfile.createReaderForScan(); + return openDataReaderForScan(dfile.diskAccessMode()); + } + + public RandomAccessReader openDataReaderForScan(DiskAccessMode diskAccessMode) + { + boolean isSameDiskAccessMode = diskAccessMode == dfile.diskAccessMode(); + boolean isDirectIONotSupported = diskAccessMode == DiskAccessMode.direct && !dfile.supportsDirectIO(); + + if (isSameDiskAccessMode || isDirectIONotSupported) + return dfile.createReaderForScan(OnReaderClose.RETAIN_FILE_OPEN); + + FileHandle dataFile = dfile.toBuilder() + .withDiskAccessMode(diskAccessMode) + .complete(); + try + { + return dataFile.createReaderForScan(OnReaderClose.CLOSE_FILE); + } + catch (Throwable t) + { + dataFile.close(); + throw t; + } } public void trySkipFileCacheBefore(DecoratedKey key) diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java index cf40f4e34b..3555c22186 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java @@ -35,6 +35,7 @@ import org.apache.cassandra.io.sstable.SSTableIdentityIterator; import org.apache.cassandra.io.util.RandomAccessReader; import org.apache.cassandra.schema.TableMetadata; +import static org.apache.cassandra.config.Config.DiskAccessMode; import static org.apache.cassandra.io.sstable.format.SSTableReader.PartitionPositionBounds; /// Simple SSTable scanner that reads sequentially through an SSTable without using the index. @@ -71,11 +72,12 @@ implements ISSTableScanner /// The ranges can be constructed by [SSTableReader#getPositionsForRanges] and similar methods as done by the /// various [SSTableReader#getScanner] variations. public SSTableSimpleScanner(SSTableReader sstable, - Collection boundsList) + Collection boundsList, + DiskAccessMode diskAccessMode) { assert sstable != null; - this.dfile = sstable.openDataReaderForScan(); + this.dfile = sstable.openDataReaderForScan(diskAccessMode); this.sstable = sstable; this.tableMetadata = sstable.metadata(); this.sizeInBytes = boundsList.stream().mapToLong(ppb -> ppb.upperPosition - ppb.lowerPosition).sum(); diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableReaderLoadingBuilder.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableReaderLoadingBuilder.java index 4b647549cd..0c645077dd 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableReaderLoadingBuilder.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableReaderLoadingBuilder.java @@ -62,7 +62,7 @@ extends SSTableReaderLoadingBuilder dataFileBuilder.bufferSize(bufferSize); dataFileBuilder.withChunkCache(chunkCache); - dataFileBuilder.mmapped(ioOptions.defaultDiskAccessMode); + dataFileBuilder.withDiskAccessMode(ioOptions.defaultDiskAccessMode); return dataFileBuilder; } diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java index 0c1ccf8560..a3985d5fbd 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java @@ -380,22 +380,21 @@ public abstract class SortedTableWriter

lastEarlyOpenLength) diff --git a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java index 6ee0406230..6a2c488289 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java +++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java @@ -205,7 +205,9 @@ public class BtiTableReaderLoadingBuilder extends SortedTableReaderLoadingBuilde rowIndexFileBuilder = new FileHandle.Builder(descriptor.fileFor(Components.ROW_INDEX)); rowIndexFileBuilder.withChunkCache(chunkCache); - rowIndexFileBuilder.mmapped(ioOptions.indexDiskAccessMode); + + if (ioOptions.indexDiskAccessMode != null) + rowIndexFileBuilder.withDiskAccessMode(ioOptions.indexDiskAccessMode); return rowIndexFileBuilder; } @@ -218,7 +220,9 @@ public class BtiTableReaderLoadingBuilder extends SortedTableReaderLoadingBuilde partitionIndexFileBuilder = new FileHandle.Builder(descriptor.fileFor(Components.PARTITION_INDEX)); partitionIndexFileBuilder.withChunkCache(chunkCache); - partitionIndexFileBuilder.mmapped(ioOptions.indexDiskAccessMode); + + if (ioOptions.indexDiskAccessMode != null) + partitionIndexFileBuilder.withDiskAccessMode(ioOptions.indexDiskAccessMode); return partitionIndexFileBuilder; } diff --git a/src/java/org/apache/cassandra/io/util/ByteBufferHolder.java b/src/java/org/apache/cassandra/io/util/ByteBufferHolder.java new file mode 100644 index 0000000000..0f70c71b81 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/ByteBufferHolder.java @@ -0,0 +1,28 @@ +/* + * 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.nio.ByteBuffer; + +/** + * Holder for reusable thread-local ByteBuffers. + */ +public interface ByteBufferHolder +{ + ByteBuffer getBuffer(int size); +} diff --git a/src/java/org/apache/cassandra/io/util/ChannelProxy.java b/src/java/org/apache/cassandra/io/util/ChannelProxy.java index 81665beecd..43df27db51 100644 --- a/src/java/org/apache/cassandra/io/util/ChannelProxy.java +++ b/src/java/org/apache/cassandra/io/util/ChannelProxy.java @@ -22,8 +22,11 @@ import java.nio.ByteBuffer; import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; import java.nio.channels.WritableByteChannel; +import java.nio.file.OpenOption; import java.nio.file.StandardOpenOption; +import com.sun.nio.file.ExtendedOpenOption; + import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.utils.NativeLibrary; import org.apache.cassandra.utils.concurrent.RefCounted; @@ -40,15 +43,22 @@ import org.apache.cassandra.utils.concurrent.SharedCloseableImpl; */ public final class ChannelProxy extends SharedCloseableImpl { + + public enum IOMode + { + BUFFERED, + DIRECT + } + private final File file; private final String filePath; private final FileChannel channel; - public static FileChannel openChannel(File file) + public static FileChannel openChannel(File file, OpenOption... openOptions) { try { - return FileChannel.open(file.toPath(), StandardOpenOption.READ); + return FileChannel.open(file.toPath(), openOptions); } catch (IOException e) { @@ -56,14 +66,32 @@ public final class ChannelProxy extends SharedCloseableImpl } } + private static OpenOption[] openOptions(IOMode ioMode) + { + switch (ioMode) + { + case DIRECT: + return new OpenOption[]{ StandardOpenOption.READ, ExtendedOpenOption.DIRECT }; + case BUFFERED: + return new OpenOption[]{ StandardOpenOption.READ }; + default: + throw new IllegalArgumentException("Unknown IOMode " + ioMode); + } + } + public ChannelProxy(String path) { - this (new File(path)); + this(new File(path)); } public ChannelProxy(File file) { - this(file, openChannel(file)); + this(file, IOMode.BUFFERED); + } + + public ChannelProxy(File file, IOMode ioMode) + { + this(file, openChannel(file, openOptions(ioMode))); } public ChannelProxy(File file, FileChannel channel) diff --git a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java index b6b3c9a6a2..53f2a4739a 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java +++ b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java @@ -115,10 +115,58 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl } - + /** + * The returned buffer is only valid until the next call to read(). Callers must consume the data immediately. + */ ByteBuffer read(CompressionMetadata.Chunk chunk, boolean shouldCheckCrc) throws CorruptBlockException; } + private static final class DirectRandomAccessReader implements CompressedReader + { + + private final ChannelProxy channel; + private final int blockSize; + private final DirectThreadLocalByteBufferHolder bufferHolder; + + DirectRandomAccessReader(ChannelProxy ch, int blockSize) + { + this.channel = ch; + this.blockSize = blockSize; + this.bufferHolder = new DirectThreadLocalByteBufferHolder(blockSize); + } + + @Override + public ByteBuffer read(CompressionMetadata.Chunk chunk, boolean shouldCheckCrc) throws CorruptBlockException + { + int length = shouldCheckCrc ? chunk.length + Integer.BYTES // length + checksum length + : chunk.length; + + long alignedPos = chunk.offset & -blockSize; + int delta = (int) (chunk.offset - alignedPos); + + ByteBuffer buffer = bufferHolder.getBuffer(length + delta); + if (channel.read(buffer, alignedPos) < length + delta) + throw new CorruptBlockException(channel.filePath(), chunk); + + buffer.position(delta); + buffer.limit(delta + length); + + ByteBuffer slice = buffer.slice(); + slice.limit(chunk.length); // limit at chunk content end (before CRC) + + if (shouldCheckCrc) + { + int checksum = (int) ChecksumType.CRC32.of(slice); + slice.limit(length); + if (slice.getInt() != checksum) + throw new CorruptBlockException(channel.filePath(), chunk); + + slice.position(0).limit(chunk.length); + } + return slice; + } + } + private static class RandomAccessCompressedReader implements CompressedReader { private final ChannelProxy channel; @@ -155,15 +203,17 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl private static class ScanCompressedReader implements CompressedReader { + private final ChannelProxy channel; - private final ThreadLocalByteBufferHolder bufferHolder; + private final ByteBufferHolder bufferHolder; private final ThreadLocalReadAheadBuffer readAheadBuffer; - private ScanCompressedReader(ChannelProxy channel, CompressionMetadata metadata, int readAheadBufferSize) + private ScanCompressedReader(ChannelProxy channel, ByteBufferHolder bufferHolder, + ThreadLocalReadAheadBuffer readAheadBuffer) { this.channel = channel; - this.bufferHolder = new ThreadLocalByteBufferHolder(metadata.compressor().preferredBufferType()); - this.readAheadBuffer = new ThreadLocalReadAheadBuffer(channel, readAheadBufferSize, metadata.compressor().preferredBufferType()); + this.bufferHolder = bufferHolder; + this.readAheadBuffer = readAheadBuffer; } @Override @@ -216,12 +266,101 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl return readAheadBuffer.hasBuffer(); } + @Override public void close() { readAheadBuffer.close(); } } + public static class Direct extends CompressedChunkReader + { + + private final CompressedReader reader; + private final CompressedReader scanReader; + + public Direct(ChannelProxy channel, CompressionMetadata metadata, Supplier crcCheckChanceSupplier) + { + super(channel, metadata, crcCheckChanceSupplier); + int blockSize = FileUtils.getFileBlockSize(channel.file()); + this.reader = new DirectRandomAccessReader(channel, blockSize); + + int readAheadBufferSize = DatabaseDescriptor.getCompressedReadAheadBufferSize(); + this.scanReader = (readAheadBufferSize > 0 && readAheadBufferSize > metadata.chunkLength()) + ? new ScanCompressedReader(channel, + new DirectThreadLocalByteBufferHolder(blockSize), + new DirectThreadLocalReadAheadBuffer(channel, readAheadBufferSize, blockSize)) + : null; + } + + @Override + public void readChunk(long position, ByteBuffer uncompressed) + { + assert (position & -uncompressed.capacity()) == position; + assert position <= fileLength; + + try + { + CompressionMetadata.Chunk chunk = metadata.chunkFor(position); + boolean shouldCheckCrc = shouldCheckCrc(); + + uncompressed.clear(); + CompressedReader readFrom = (scanReader != null && scanReader.allocated()) ? scanReader : reader; + if (chunk.length < maxCompressedLength) + { + ByteBuffer compressed = readFrom.read(chunk, shouldCheckCrc); + try + { + metadata.compressor().uncompress(compressed, uncompressed); + } + catch (IOException e) + { + throw new CorruptBlockException(channel.filePath(), chunk, e); + } + } + else + { + ByteBuffer buffer = readFrom.read(chunk, shouldCheckCrc); + uncompressed.put(buffer); + } + + uncompressed.flip(); + } + catch (CorruptBlockException e) + { + // Make sure reader does not see stale data. + uncompressed.position(0).limit(0); + throw new CorruptSSTableException(e, channel.filePath()); + } + } + + @Override + protected CompressedChunkReader forScan() + { + if (scanReader != null) + scanReader.allocateResources(); + + return this; + } + + @Override + public void releaseUnderlyingResources() + { + if (scanReader != null) + scanReader.deallocateResources(); + } + + @Override + public void close() + { + reader.close(); + if (scanReader != null) + scanReader.close(); + + super.close(); + } + } + public static class Standard extends CompressedChunkReader { @@ -235,9 +374,12 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl int readAheadBufferSize = DatabaseDescriptor.getCompressedReadAheadBufferSize(); scanReader = (readAheadBufferSize > 0 && readAheadBufferSize > metadata.chunkLength()) - ? new ScanCompressedReader(channel, metadata, readAheadBufferSize) : null; + ? new ScanCompressedReader(channel, + new ThreadLocalByteBufferHolder(metadata.compressor().preferredBufferType()), + new ThreadLocalReadAheadBuffer(channel, readAheadBufferSize, metadata.compressor().preferredBufferType())) : null; } + @Override protected CompressedChunkReader forScan() { if (scanReader != null) @@ -282,6 +424,7 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl } else { + // Read directly into destination buffer for zero-copy uncompressed path uncompressed.position(0).limit(chunk.length); if (channel.read(uncompressed, chunk.offset) != chunk.length) throw new CorruptBlockException(channel.filePath(), chunk); diff --git a/src/java/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolder.java b/src/java/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolder.java new file mode 100644 index 0000000000..ec4c9abb77 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolder.java @@ -0,0 +1,79 @@ +/* + * 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.nio.ByteBuffer; +import java.util.concurrent.ConcurrentHashMap; + +import org.agrona.BitUtil; +import org.agrona.BufferUtil; + +import org.apache.cassandra.utils.memory.MemoryUtil; + +import io.netty.util.concurrent.FastThreadLocal; +import sun.nio.ch.DirectBuffer; + +/** + * Thread-local holder for block-aligned direct ByteBuffers used in Direct I/O operations. + *

+ * Buffers are shared across all holders with the same block size and reused per-thread, + * growing as needed. This mirrors {@link ThreadLocalByteBufferHolder}'s design. + */ +public final class DirectThreadLocalByteBufferHolder implements ByteBufferHolder +{ + private static final ConcurrentHashMap> buffersByBlockSize = new ConcurrentHashMap<>(); + + private final FastThreadLocal threadLocal; + private final int blockSize; + + public DirectThreadLocalByteBufferHolder(int blockSize) + { + this.blockSize = blockSize; + this.threadLocal = buffersByBlockSize.computeIfAbsent(blockSize, k -> new FastThreadLocal<>()); + } + + @Override + public ByteBuffer getBuffer(int size) + { + int alignedSize = BitUtil.align(size, blockSize); + ByteBuffer buffer = threadLocal.get(); + + if (buffer != null && buffer.capacity() >= alignedSize) + { + buffer.clear().limit(alignedSize); + return buffer; + } + + if (buffer != null) + cleanBuffer(buffer); + + buffer = BufferUtil.allocateDirectAligned(alignedSize, blockSize); + threadLocal.set(buffer); + return buffer; + } + + private static void cleanBuffer(ByteBuffer buffer) + { + // Aligned buffers are slices; clean the backing buffer (attachment) + DirectBuffer db = (DirectBuffer) buffer; + ByteBuffer attachment = (ByteBuffer) db.attachment(); + MemoryUtil.clean(attachment != null ? attachment : buffer); + } + +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java new file mode 100644 index 0000000000..a2c66cc0f7 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java @@ -0,0 +1,49 @@ +/* + * 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.nio.ByteBuffer; + +import org.agrona.BitUtil; +import org.agrona.BufferUtil; + +import org.apache.cassandra.io.sstable.CorruptSSTableException; + +public final class DirectThreadLocalReadAheadBuffer extends ThreadLocalReadAheadBuffer +{ + + private final int blockSize; + + public DirectThreadLocalReadAheadBuffer(ChannelProxy channel, int bufferSize, int blockSize) + { + super(channel, () -> BufferUtil.allocateDirectAligned(BitUtil.align(bufferSize, blockSize), blockSize)); + this.blockSize = blockSize; + } + + @Override + protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToRead) + { + int alignedSizeToRead = BitUtil.align(sizeToRead, blockSize); + + blockBuffer.limit(alignedSizeToRead); + + if (channel.read(blockBuffer, blockPosition) < sizeToRead) + throw new CorruptSSTableException(null, channel.filePath()); + } +} diff --git a/src/java/org/apache/cassandra/io/util/FileHandle.java b/src/java/org/apache/cassandra/io/util/FileHandle.java index 7e4b214128..1153dcb046 100644 --- a/src/java/org/apache/cassandra/io/util/FileHandle.java +++ b/src/java/org/apache/cassandra/io/util/FileHandle.java @@ -25,7 +25,6 @@ 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; @@ -34,6 +33,8 @@ import org.apache.cassandra.utils.concurrent.Ref; import org.apache.cassandra.utils.concurrent.RefCounted; import org.apache.cassandra.utils.concurrent.SharedCloseableImpl; +import static org.apache.cassandra.config.Config.DiskAccessMode; + /** * {@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}. @@ -48,6 +49,13 @@ import org.apache.cassandra.utils.concurrent.SharedCloseableImpl; */ public class FileHandle extends SharedCloseableImpl { + + public enum OnReaderClose + { + CLOSE_FILE, + RETAIN_FILE_OPEN + } + public final ChannelProxy channel; public final long onDiskLength; @@ -62,17 +70,41 @@ public class FileHandle extends SharedCloseableImpl */ private final Optional compressionMetadata; + private final DiskAccessMode diskAccessMode; + + // Properties to support unbuilding via toBuilder + private final ChunkCache chunkCache; + private final MmappedRegionsCache mmappedRegionsCache; + private final Supplier crcCheckChanceSupplier; + private final long lengthOverride; + private final int bufferSize; + private final BufferType bufferType; + private FileHandle(Cleanup cleanup, ChannelProxy channel, RebuffererFactory rebuffererFactory, CompressionMetadata compressionMetadata, - long onDiskLength) + long onDiskLength, + DiskAccessMode diskAccessMode, + ChunkCache chunkCache, + MmappedRegionsCache mmappedRegionsCache, + Supplier crcCheckChanceSupplier, + long lengthOverride, + int bufferSize, + BufferType bufferType) { super(cleanup); this.rebuffererFactory = rebuffererFactory; this.channel = channel; this.compressionMetadata = Optional.ofNullable(compressionMetadata); this.onDiskLength = onDiskLength; + this.diskAccessMode = diskAccessMode; + this.chunkCache = chunkCache; + this.mmappedRegionsCache = mmappedRegionsCache; + this.crcCheckChanceSupplier = crcCheckChanceSupplier; + this.lengthOverride = lengthOverride; + this.bufferSize = bufferSize; + this.bufferType = bufferType; } private FileHandle(FileHandle copy) @@ -82,6 +114,26 @@ public class FileHandle extends SharedCloseableImpl rebuffererFactory = copy.rebuffererFactory; compressionMetadata = copy.compressionMetadata; onDiskLength = copy.onDiskLength; + diskAccessMode = copy.diskAccessMode; + chunkCache = copy.chunkCache; + mmappedRegionsCache = copy.mmappedRegionsCache; + crcCheckChanceSupplier = copy.crcCheckChanceSupplier; + lengthOverride = copy.lengthOverride; + bufferSize = copy.bufferSize; + bufferType = copy.bufferType; + } + + public Builder toBuilder() + { + return new FileHandle.Builder(file()) + .withDiskAccessMode(diskAccessMode) + .withCrcCheckChance(crcCheckChanceSupplier) + .withMmappedRegionsCache(mmappedRegionsCache) + .withCompressionMetadata(compressionMetadata.orElse(null)) + .withLengthOverride(lengthOverride) + .withChunkCache(chunkCache) + .bufferSize(bufferSize) + .bufferType(bufferType); } /** @@ -112,6 +164,16 @@ public class FileHandle extends SharedCloseableImpl return compressionMetadata; } + public DiskAccessMode diskAccessMode() + { + return diskAccessMode; + } + + public boolean supportsDirectIO() + { + return FileUtils.isDirectIOSupported(file()) && compressionMetadata.isPresent(); + } + @Override public void addTo(Ref.IdentityCollection identities) { @@ -134,9 +196,9 @@ public class FileHandle extends SharedCloseableImpl return createReader(null); } - public RandomAccessReader createReaderForScan() + public RandomAccessReader createReaderForScan(OnReaderClose onReaderClose) { - return createReader(null, true); + return createReader(null, true, onReaderClose); } /** @@ -153,7 +215,20 @@ public class FileHandle extends SharedCloseableImpl public RandomAccessReader createReader(RateLimiter limiter, boolean forScan) { - return new RandomAccessReader(instantiateRebufferer(limiter, forScan)); + return createReader(limiter, forScan, OnReaderClose.RETAIN_FILE_OPEN); + } + + public RandomAccessReader createReader(RateLimiter limiter, boolean forScan, OnReaderClose onReaderClose) + { + if (onReaderClose == OnReaderClose.CLOSE_FILE) + { + return new RandomAccessReader.RandomAccessReaderWithOwnFile(instantiateRebufferer(limiter, forScan), this); + } + else if (onReaderClose == OnReaderClose.RETAIN_FILE_OPEN) + { + return new RandomAccessReader(instantiateRebufferer(limiter, forScan)); + } + throw new IllegalArgumentException("Unknown close policy: " + onReaderClose); } public FileDataInput createReader(long position) @@ -272,7 +347,7 @@ public class FileHandle extends SharedCloseableImpl private ChunkCache chunkCache; private int bufferSize = RandomAccessReader.DEFAULT_BUFFER_SIZE; private BufferType bufferType = BufferType.OFF_HEAP; - private boolean mmapped = false; + private DiskAccessMode diskAccessMode = DiskAccessMode.standard; private long lengthOverride = -1; private MmappedRegionsCache mmappedRegionsCache; @@ -314,21 +389,15 @@ public class FileHandle extends SharedCloseableImpl return this; } - /** - * Set whether to use mmap for reading - * - * @param mmapped true if using mmap - * @return this instance - */ - public Builder mmapped(boolean mmapped) + public Builder mmapped() { - this.mmapped = mmapped; + withDiskAccessMode(DiskAccessMode.mmap); return this; } - public Builder mmapped(Config.DiskAccessMode diskAccessMode) + public Builder withDiskAccessMode(DiskAccessMode diskAccessMode) { - this.mmapped = diskAccessMode == Config.DiskAccessMode.mmap; + this.diskAccessMode = diskAccessMode; return this; } @@ -380,7 +449,28 @@ public class FileHandle extends SharedCloseableImpl */ public FileHandle complete() { - return complete(ChannelProxy::new); + return complete(file -> new ChannelProxy(file, ioMode())); + } + + private ChannelProxy.IOMode ioMode() + { + switch (diskAccessMode) + { + case mmap: + case standard: + case auto: + case legacy: + case mmap_index_only: + // For mmap and standard, BUFFERED is the correct mode. + // For auto/legacy/mmap_index_only, these are meta-modes that should normally be resolved + // during DatabaseDescriptor initialization. In client/tool mode where full initialization + // didn't occur, BUFFERED (standard) is the safe default. + return ChannelProxy.IOMode.BUFFERED; + case direct: + return ChannelProxy.IOMode.DIRECT; + default: + throw new AssertionError("Unhandled diskAccessMode: " + diskAccessMode); + } } @VisibleForTesting @@ -402,7 +492,7 @@ public class FileHandle extends SharedCloseableImpl { rebuffererFactory = new EmptyRebufferer(channel); } - else if (mmapped) + else if (DiskAccessMode.mmap == diskAccessMode) { if (compressionMetadata != null) { @@ -421,7 +511,16 @@ public class FileHandle extends SharedCloseableImpl { if (compressionMetadata != null) { - rebuffererFactory = maybeCached(new CompressedChunkReader.Standard(channel, compressionMetadata, crcCheckChanceSupplier)); + final CompressedChunkReader compressedChunkReader; + if (DiskAccessMode.direct == diskAccessMode) + { + compressedChunkReader = new CompressedChunkReader.Direct(channel, compressionMetadata, crcCheckChanceSupplier); + } + else + { + compressedChunkReader = new CompressedChunkReader.Standard(channel, compressionMetadata, crcCheckChanceSupplier); + } + rebuffererFactory = maybeCached(compressedChunkReader); } else { @@ -429,10 +528,10 @@ public class FileHandle extends SharedCloseableImpl 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; + Cleanup cleanup = new Cleanup(channel, rebuffererFactory, compressionMetadata, chunkCache); + return new FileHandle(cleanup, channel, rebuffererFactory, compressionMetadata, length, diskAccessMode, chunkCache, + mmappedRegionsCache, crcCheckChanceSupplier, lengthOverride, bufferSize, bufferType); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java index 731c2f4cf0..11dfe2cff1 100644 --- a/src/java/org/apache/cassandra/io/util/FileUtils.java +++ b/src/java/org/apache/cassandra/io/util/FileUtils.java @@ -47,6 +47,7 @@ import java.util.stream.Stream; import com.google.common.base.Preconditions; import com.google.common.util.concurrent.RateLimiter; +import com.sun.nio.file.ExtendedOpenOption; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -740,14 +741,54 @@ public final class FileUtils } } + public static boolean isDirectIOSupported(File file) + { + File testFile = null; + try + { + File dir = file.isDirectory() ? file : file.parent(); + testFile = createTempFile("direct-io-test", ".tmp", dir); + + // Direct IO requires knowing the block size for buffer alignment + if (blockSize(testFile) <= 0) + return false; + + try (FileChannel channel = FileChannel.open(testFile.toPath(), + StandardOpenOption.READ, + ExtendedOpenOption.DIRECT)) + { + return true; + } + } + catch (Exception e) + { + return false; + } + finally + { + if (testFile != null) + testFile.tryDelete(); + } + } + + public static int getFileBlockSize(File file) + { + try + { + return blockSize(file); + } + catch (IOException e) + { + throw new RuntimeException("Failed to get file block size in " + file, e); + } + } + public static int getBlockSize(File directory) { File f = FileUtils.createTempFile("block-size-test", ".tmp", directory); try { - long bs = Files.getFileStore(f.toPath()).getBlockSize(); - assert bs >= 0 && bs <= Integer.MAX_VALUE; - return (int) bs; + return blockSize(f); } catch (IOException e) { @@ -759,6 +800,13 @@ public final class FileUtils } } + private static int blockSize(File file) throws IOException + { + long bs = Files.getFileStore(file.toPath()).getBlockSize(); + assert bs >= 0 && bs <= Integer.MAX_VALUE; + return (int) bs; + } + public static class DuplicateHardlinkException extends RuntimeException { public DuplicateHardlinkException(String message) diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index b89e59eb52..eb5a182402 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -22,6 +22,7 @@ import java.nio.ByteOrder; import javax.annotation.concurrent.NotThreadSafe; +import com.google.common.base.Preconditions; import com.google.common.primitives.Ints; import org.apache.cassandra.io.compress.BufferType; @@ -319,6 +320,31 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa } } + static class RandomAccessReaderWithOwnFile extends RandomAccessReader + { + + private final FileHandle fileHandle; + + RandomAccessReaderWithOwnFile(Rebufferer rebufferer, FileHandle fileHandle) + { + super(rebufferer); + this.fileHandle = Preconditions.checkNotNull(fileHandle, "fileHandle cannot be null"); + } + + @Override + public void close() + { + try + { + super.close(); + } + finally + { + fileHandle.close(); + } + } + } + /** * Open a RandomAccessReader (not compressed, not mmapped, no read throttling) that will own its channel. * diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java b/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java index e22ab14aff..3bea891ae3 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java @@ -29,8 +29,9 @@ import io.netty.util.concurrent.FastThreadLocal; /** * Utility class that allow buffers to be reused by storing them in a thread local instance. */ -public final class ThreadLocalByteBufferHolder +public final class ThreadLocalByteBufferHolder implements ByteBufferHolder { + private static final EnumMap> reusableBBHolder = new EnumMap<>(BufferType.class); // Convenience variable holding a ref to the current resuableBB to avoid map lookups private final FastThreadLocal reusableBB; @@ -39,7 +40,7 @@ public final class ThreadLocalByteBufferHolder { for (BufferType bbType : BufferType.values()) { - reusableBBHolder.put(bbType, new FastThreadLocal() + reusableBBHolder.put(bbType, new FastThreadLocal<>() { protected ByteBuffer initialValue() { @@ -47,7 +48,7 @@ public final class ThreadLocalByteBufferHolder } }); } - }; + } /** * The type of buffer that will be returned @@ -69,6 +70,7 @@ public final class ThreadLocalByteBufferHolder * @param size the buffer size * @return the buffer for the current thread. */ + @Override public ByteBuffer getBuffer(int size) { ByteBuffer buffer = reusableBB.get(); diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index 63dca9bfc4..3a564931a9 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -21,24 +21,27 @@ package org.apache.cassandra.io.util; import java.nio.ByteBuffer; import java.util.HashMap; import java.util.Map; +import java.util.function.Supplier; import org.apache.cassandra.io.compress.BufferType; import org.apache.cassandra.io.sstable.CorruptSSTableException; +import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.memory.MemoryUtil; import io.netty.util.concurrent.FastThreadLocal; -public final class ThreadLocalReadAheadBuffer +public class ThreadLocalReadAheadBuffer implements Closeable { + private static class Block { ByteBuffer buffer = null; int index = -1; } - private final ChannelProxy channel; + protected final ChannelProxy channel; - private final BufferType bufferType; + private final Supplier bufferSupplier; private static final FastThreadLocal> blockMap = new FastThreadLocal<>() { @@ -49,15 +52,19 @@ public final class ThreadLocalReadAheadBuffer } }; - private final int bufferSize; + private volatile int bufferSize = -1; private final long channelSize; public ThreadLocalReadAheadBuffer(ChannelProxy channel, int bufferSize, BufferType bufferType) + { + this(channel, () -> bufferType.allocate(bufferSize)); + } + + public ThreadLocalReadAheadBuffer(ChannelProxy channel, Supplier bufferSupplier) { this.channel = channel; this.channelSize = channel.size(); - this.bufferSize = bufferSize; - this.bufferType = bufferType; + this.bufferSupplier = bufferSupplier; } public boolean hasBuffer() @@ -65,9 +72,6 @@ public final class ThreadLocalReadAheadBuffer return block().buffer != null; } - /** - * Safe to call only if {@link #hasBuffer()} is true - */ public int remaining() { return getBlock().buffer.remaining(); @@ -83,8 +87,10 @@ public final class ThreadLocalReadAheadBuffer Block block = block(); if (block.buffer == null) { - block.buffer = bufferType.allocate(bufferSize); + block.buffer = bufferSupplier.get(); block.buffer.clear(); + if (bufferSize == -1) + bufferSize = block.buffer.capacity(); } return block; } @@ -107,10 +113,7 @@ public final class ThreadLocalReadAheadBuffer if (block.index != blockNo) { blockBuffer.flip(); - blockBuffer.limit(sizeToRead); - if (channel.read(blockBuffer, blockPosition) != sizeToRead) - throw new CorruptSSTableException(null, channel.filePath()); - + loadBlock(blockBuffer, blockPosition, sizeToRead); block.index = blockNo; } @@ -119,6 +122,13 @@ public final class ThreadLocalReadAheadBuffer blockBuffer.position((int) (realPosition - blockPosition)); } + protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToRead) + { + blockBuffer.limit(sizeToRead); + if (channel.read(blockBuffer, blockPosition) != sizeToRead) + throw new CorruptSSTableException(null, channel.filePath()); + } + public int read(ByteBuffer dest, int length) { Block block = getBlock(); @@ -151,6 +161,7 @@ public final class ThreadLocalReadAheadBuffer } } + @Override public void close() { clear(true); diff --git a/src/java/org/apache/cassandra/service/StartupChecks.java b/src/java/org/apache/cassandra/service/StartupChecks.java index b3b33ee93b..31a2686af7 100644 --- a/src/java/org/apache/cassandra/service/StartupChecks.java +++ b/src/java/org/apache/cassandra/service/StartupChecks.java @@ -126,6 +126,7 @@ public class StartupChecks checkMaxMapCount, checkReadAheadKbSetting, checkDataDirs, + checkDirectIOSupport, checkSSTablesFormat, checkSystemKeyspaceState, checkLegacyAuthTables, @@ -258,7 +259,8 @@ public class StartupChecks Set directIOWritePaths = new HashSet<>(); if (DatabaseDescriptor.getCommitLogWriteDiskAccessMode() == Config.DiskAccessMode.direct) directIOWritePaths.add(new File(DatabaseDescriptor.getCommitLogLocation()).toPath()); - // TODO: add data directories when direct IO is supported for flushing and compaction + // Note: Data directories for direct IO compaction reads are checked in checkDirectIOSupport. + // This check is specifically for direct IO writes which are currently only supported for commit log. if (!directIOWritePaths.isEmpty() && IGNORE_KERNEL_BUG_1057843_CHECK.getBoolean()) { @@ -718,6 +720,57 @@ public class StartupChecks } }; + public static final StartupCheck checkDirectIOSupport = new StartupCheck() + { + @Override + public String name() + { + return "directio_support"; + } + + @Override + public void execute(StartupChecksConfiguration configuration) throws StartupException + { + if (configuration.isDisabled(name())) + return; + + // Only check if compaction_read_disk_access_mode is direct + if (DatabaseDescriptor.getCompactionReadDiskAccessMode() != Config.DiskAccessMode.direct) + return; + + List unsupportedLocations = findDirectIOUnsupportedLocations(DatabaseDescriptor.getAllDataFileLocations()); + + if (!unsupportedLocations.isEmpty()) + { + throw new StartupException(StartupException.ERR_WRONG_DISK_STATE, + String.format("Direct I/O is configured for compaction reads (compaction_read_disk_access_mode=direct), " + + "but the following data directories do not support Direct I/O: %s. " + + "Either change compaction_read_disk_access_mode to 'standard' in cassandra.yaml, " + + "or ensure all data directories are on filesystems that support Direct I/O. " + + "Network filesystems (NFS, CIFS) and some virtual filesystems do not support Direct I/O.", + unsupportedLocations)); + } + } + }; + + @VisibleForTesting + static List findDirectIOUnsupportedLocations(String[] dataFileLocations) + { + List unsupportedLocations = new ArrayList<>(); + + for (String dataDir : dataFileLocations) + { + File dir = new File(dataDir); + if (!dir.exists()) + continue; // Directory doesn't exist yet, skip + + if (!FileUtils.isDirectIOSupported(dir)) + unsupportedLocations.add(dataDir); + } + + return unsupportedLocations; + } + public static final StartupCheck checkSSTablesFormat = new StartupCheck() { @Override diff --git a/test/distributed/org/apache/cassandra/distributed/test/FailingRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/FailingRepairTest.java index da3c8fdce4..6b60f3b13d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/FailingRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/FailingRepairTest.java @@ -48,6 +48,7 @@ import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameters; import org.apache.cassandra.Util; +import org.apache.cassandra.config.Config.DiskAccessMode; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DataRange; import org.apache.cassandra.db.Keyspace; @@ -301,6 +302,16 @@ public class FailingRepairTest extends TestBaseImpl implements Serializable return new FailingISSTableScanner(); } + public ISSTableScanner getScanner(DiskAccessMode diskAccessMode) + { + return new FailingISSTableScanner(); + } + + public ISSTableScanner getScanner(Collection> ranges, DiskAccessMode diskAccessMode) + { + return new FailingISSTableScanner(); + } + public ChannelProxy getDataChannel() { throw new RuntimeException(); diff --git a/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java b/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java index ee8adceb60..5d146e5ca1 100644 --- a/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java +++ b/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java @@ -27,6 +27,7 @@ import java.util.Set; import com.google.common.util.concurrent.RateLimiter; +import org.apache.cassandra.config.Config.DiskAccessMode; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DataRange; import org.apache.cassandra.db.DecoratedKey; @@ -344,6 +345,18 @@ public abstract class ForwardingSSTableReader extends SSTableReader return delegate.getScanner(rangeIterator); } + @Override + public ISSTableScanner getScanner(DiskAccessMode diskAccessMode) + { + return delegate.getScanner(diskAccessMode); + } + + @Override + public ISSTableScanner getScanner(Collection> ranges, DiskAccessMode diskAccessMode) + { + return delegate.getScanner(ranges, diskAccessMode); + } + @Override public UnfilteredPartitionIterator partitionIterator(ColumnFilter columns, DataRange dataRange, SSTableReadsListener listener) { diff --git a/test/unit/org/apache/cassandra/db/commitlog/DirectIOSegmentBytemanTest.java b/test/unit/org/apache/cassandra/db/commitlog/DirectIOSegmentBytemanTest.java index b752fac238..4409323229 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/DirectIOSegmentBytemanTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/DirectIOSegmentBytemanTest.java @@ -53,7 +53,7 @@ public class DirectIOSegmentBytemanTest @Test @BMRules(rules = { @BMRule(name = "Commitlog dir do not support direct io", targetClass = "FileUtils", - targetMethod = "getBlockSize", + targetMethod = "blockSize", action = "return 0;") } ) public void testDirectIOUnSupportWithDirectConfig() { diff --git a/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java b/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java index 2ccaabbeba..504a0e47ad 100644 --- a/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java @@ -398,7 +398,7 @@ public class CheckpointIntervalArrayIndexTest Map files = new EnumMap<>(IndexComponent.class); for (IndexComponent c : descriptor.getLiveComponents()) - files.put(c, new FileHandle.Builder(descriptor.fileFor(c)).mmapped(true).complete()); + files.put(c, new FileHandle.Builder(descriptor.fileFor(c)).mmapped().complete()); List segments = RouteIndexFormat.readSegments(files); files.remove(IndexComponent.SEGMENT).close(); files.remove(IndexComponent.METADATA).close(); diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java index 591c2aa433..6d73b987f2 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java @@ -44,6 +44,7 @@ import org.apache.cassandra.io.util.SequentialWriterOption; import org.apache.cassandra.schema.CompressionParams; import org.apache.cassandra.utils.SyncUtil; +import static org.apache.cassandra.config.Config.DiskAccessMode; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -62,40 +63,40 @@ public class CompressedRandomAccessReaderTest public void testResetAndTruncate() throws IOException { // test reset in current buffer or previous one - testResetAndTruncate(FileUtils.createTempFile("normal", "1"), false, false, 10, 0); - testResetAndTruncate(FileUtils.createTempFile("normal", "2"), false, false, CompressionParams.DEFAULT_CHUNK_LENGTH, 0); + testResetAndTruncate(FileUtils.createTempFile("normal", "1"), false, DiskAccessMode.standard, 10); + testResetAndTruncate(FileUtils.createTempFile("normal", "2"), false, DiskAccessMode.standard, CompressionParams.DEFAULT_CHUNK_LENGTH); } @Test public void testResetAndTruncateCompressed() throws IOException { // test reset in current buffer or previous one - testResetAndTruncate(FileUtils.createTempFile("compressed", "1"), true, false, 10, 0); - testResetAndTruncate(FileUtils.createTempFile("compressed", "2"), true, false, CompressionParams.DEFAULT_CHUNK_LENGTH, 0); + testResetAndTruncate(FileUtils.createTempFile("compressed", "1"), true, DiskAccessMode.standard, 10); + testResetAndTruncate(FileUtils.createTempFile("compressed", "2"), true, DiskAccessMode.standard, CompressionParams.DEFAULT_CHUNK_LENGTH); } @Test public void testResetAndTruncateCompressedMmap() throws IOException { // test reset in current buffer or previous one - testResetAndTruncate(FileUtils.createTempFile("compressed_mmap", "1"), true, true, 10, 0); - testResetAndTruncate(FileUtils.createTempFile("compressed_mmap", "2"), true, true, CompressionParams.DEFAULT_CHUNK_LENGTH, 0); + testResetAndTruncate(FileUtils.createTempFile("compressed_mmap", "1"), true, DiskAccessMode.mmap, 10); + testResetAndTruncate(FileUtils.createTempFile("compressed_mmap", "2"), true, DiskAccessMode.mmap, CompressionParams.DEFAULT_CHUNK_LENGTH); } @Test public void testResetAndTruncateCompressedUncompressedChunks() throws IOException { // test reset in current buffer or previous one - testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks", "1"), true, false, 10, 3); - testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks", "2"), true, false, CompressionParams.DEFAULT_CHUNK_LENGTH, 3); + testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks", "1"), true, DiskAccessMode.standard, 10); + testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks", "2"), true, DiskAccessMode.standard, CompressionParams.DEFAULT_CHUNK_LENGTH); } @Test public void testResetAndTruncateCompressedUncompressedChunksMmap() throws IOException { // test reset in current buffer or previous one - testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks_mmap", "1"), true, true, 10, 3); - testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks_mmap", "2"), true, true, CompressionParams.DEFAULT_CHUNK_LENGTH, 3); + testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks_mmap", "1"), true, DiskAccessMode.mmap, 10); + testResetAndTruncate(FileUtils.createTempFile("compressed_uchunks_mmap", "2"), true, DiskAccessMode.mmap, CompressionParams.DEFAULT_CHUNK_LENGTH); } @Test @@ -177,13 +178,16 @@ public class CompressedRandomAccessReaderTest } } - private static void testResetAndTruncate(File f, boolean compressed, boolean usemmap, int junkSize, double minCompressRatio) throws IOException + private static void testResetAndTruncate(File f, boolean compressed, DiskAccessMode diskAccessMode, int junkSize) throws IOException { final String filename = f.absolutePath(); writeSSTable(f, compressed ? CompressionParams.snappy() : null, junkSize); try (CompressionMetadata compressionMetadata = compressed ? CompressionMetadata.open(new File(filename + ".metadata"), f.length(), true) : null; - FileHandle fh = new FileHandle.Builder(f).mmapped(usemmap).withCompressionMetadata(compressionMetadata).complete(); + FileHandle fh = new FileHandle.Builder(f) + .withCompressionMetadata(compressionMetadata) + .withDiskAccessMode(diskAccessMode) + .complete(); RandomAccessReader reader = fh.createReader()) { String expected = "The quick brown fox jumps over the lazy dog"; diff --git a/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java b/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java index d6fadbf20b..df9dc936e0 100644 --- a/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java @@ -101,7 +101,7 @@ public class PartitionIndexTest } @Parameterized.Parameter(value = 0) - public static Config.DiskAccessMode accessMode = Config.DiskAccessMode.standard; + public static Config.DiskAccessMode diskAccessMode = Config.DiskAccessMode.standard; @BeforeClass public static void beforeClass() @@ -878,7 +878,7 @@ public class PartitionIndexTest { return new FileHandle.Builder(file) .bufferSize(PageAware.PAGE_SIZE) - .mmapped(accessMode == Config.DiskAccessMode.mmap) + .withDiskAccessMode(diskAccessMode) .withChunkCache(ChunkCache.instance); } diff --git a/test/unit/org/apache/cassandra/io/sstable/format/bti/RowIndexTest.java b/test/unit/org/apache/cassandra/io/sstable/format/bti/RowIndexTest.java index 1771b10cc4..57d38de58f 100644 --- a/test/unit/org/apache/cassandra/io/sstable/format/bti/RowIndexTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/format/bti/RowIndexTest.java @@ -93,7 +93,7 @@ public class RowIndexTest } @Parameterized.Parameter(value = 0) - public static Config.DiskAccessMode accessMode = Config.DiskAccessMode.standard; + public static Config.DiskAccessMode diskAccessMode = Config.DiskAccessMode.standard; @Test public void testSingletons() throws IOException @@ -201,7 +201,7 @@ public class RowIndexTest { complete(); - FileHandle.Builder builder = new FileHandle.Builder(file).mmapped(accessMode == Config.DiskAccessMode.mmap); + FileHandle.Builder builder = new FileHandle.Builder(file).withDiskAccessMode(diskAccessMode); fh = builder.complete(); try (RandomAccessReader rdr = fh.createReader()) { diff --git a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTestBase.java b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTestBase.java new file mode 100644 index 0000000000..7ee51a1e76 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTestBase.java @@ -0,0 +1,61 @@ +/* + * 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 accord.utils.Gen; +import accord.utils.Gens; + +import org.apache.cassandra.schema.CompressionParams; + +public abstract class CompressedChunkReaderTestBase +{ + + static Gen writerOptions() + { + Gen bufferSizes = Gens.constant(1 << 10); + return rs -> writerOption(bufferSizes.next(rs)); + } + + static SequentialWriterOption writerOption(int bufferSize) + { + return SequentialWriterOption.newBuilder() + .finishOnClose(false) + .bufferSize(bufferSize) + .build(); + } + + private enum CompressionKind { Noop, Snappy, Deflate, Lz4, Zstd } + + static Gen compressionParams(Gen chunkLengths) + { + Gen compressionRatio = Gens.pick(1.1D); + return rs -> { + CompressionKind kind = rs.pick(CompressionKind.values()); + switch (kind) + { + case Noop: return CompressionParams.noop(); + case Snappy: return CompressionParams.snappy(chunkLengths.next(rs), compressionRatio.next(rs)); + case Deflate: return CompressionParams.deflate(chunkLengths.next(rs)); + case Lz4: return CompressionParams.lz4(chunkLengths.next(rs)); + case Zstd: return CompressionParams.zstd(chunkLengths.next(rs)); + default: throw new UnsupportedOperationException(kind.name()); + } + }; + } +} diff --git a/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java new file mode 100644 index 0000000000..0065efc926 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java @@ -0,0 +1,545 @@ +/* + * 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.nio.ByteBuffer; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.channels.ReadableByteChannel; +import java.nio.channels.WritableByteChannel; +import java.nio.file.Files; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.utils.Gen; +import accord.utils.Gens; +import accord.utils.RandomSource; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ClusteringComparator; +import org.apache.cassandra.io.compress.CompressedSequentialWriter; +import org.apache.cassandra.io.compress.CompressionMetadata; +import org.apache.cassandra.io.sstable.CorruptSSTableException; +import org.apache.cassandra.io.sstable.metadata.MetadataCollector; +import org.apache.cassandra.schema.CompressionParams; +import org.apache.cassandra.utils.memory.MemoryUtil; + +import static accord.utils.Property.qt; +import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR; +import static org.apache.cassandra.schema.CompressionParams.DEFAULT_CHUNK_LENGTH; + +public class DirectCompressedChunkReaderTest extends CompressedChunkReaderTestBase +{ + private static final int CHECKSUM_LENGTH = Integer.BYTES; // CRC32 checksum size + + private static final Logger logger = LoggerFactory.getLogger(DirectCompressedChunkReaderTest.class); + private static int seed; + + static + { + DatabaseDescriptor.clientInitialization(); + } + + @BeforeClass + public static void setup() + { + seed = new Random().nextInt(); + logger.info("Seed: {}", seed); + } + + private static Gen mixedChunkLengths() + { + int minLength = 1024; + int maxLength = 1024 * 64; + return Gens.pick(Stream.iterate(minLength, n -> n <= maxLength, n -> n * 2) + .collect(Collectors.toList())); + } + + @Test + public void compressedReads() + { + var optionGen = writerOptions(); + var paramsGen = compressionParams(mixedChunkLengths()); + var fileLengthGen = Gens.longs().between(1, 1 << 16); + + testReads(optionGen, paramsGen, fileLengthGen, false); + } + + @Test + public void scanCompressedReads() + { + var optionGen = writerOptions(); + var paramsGen = compressionParams(mixedChunkLengths()); + var fileLengthGen = Gens.longs().between(1, 1 << 16); + + testReads(optionGen, paramsGen, fileLengthGen, true); + } + + @Test + public void compressedReads_edgeCases() + { + var optionGen = writerOptions(); + int chunkLength = 4096; + var paramsGen = compressionParams(Gens.constant(chunkLength)); + + // Test file smaller than chunk length + testReads(optionGen, paramsGen, Gens.longs().of(1024), true); + + // Test partial trailing chunk + testReads(optionGen, paramsGen, Gens.longs().of(chunkLength + 96), true); + } + + @Test + public void corruptBlock() + { + Gen chunkLength = Gens.constant(4096); + Gen compressionGen = compressionParams(chunkLength); + + qt().withSeed(seed).forAll(Gens.random(), compressionGen).check((rs, params) -> { + long totalBytesToWrite = params.chunkLength() * 2L; // two chunks + + List chunks = generateRandomChunks(rs, params, totalBytesToWrite); + + File dataFile = new File(JAVA_IO_TMPDIR.getString() + "data_corrupt_block.bin"); + + final CompressionMetadata metadata; + + try (CompressedSequentialWriter writer = getCompressedSequentialWriter(writerOption(1 << 10), params, dataFile)) + { + for (ByteBuffer chunk : chunks) + writer.write(chunk.duplicate()); + writer.sync(); + metadata = writer.open(0); + } + + CompressionMetadata.Chunk firstChunkMeta = metadata.chunkFor(0); + + long truncatePoint = firstChunkMeta.offset + (long) firstChunkMeta.length / 2; // Halfway into the chunk + + try (FileChannel fileChannel = FileChannel.open(dataFile.toPath(), StandardOpenOption.WRITE)) + { + fileChannel.truncate(truncatePoint); + } + + boolean exceptionThrown = false; + try + { + readAndVerifyChunks(dataFile, metadata, chunks, totalBytesToWrite, false); + } + catch (CorruptSSTableException exc) + { + exceptionThrown = true; + Assert.assertTrue("Exception message should indicate corrupt SSTable", exc.getMessage().contains(dataFile.name())); + } + finally + { + metadata.close(); + Files.deleteIfExists(dataFile.toPath()); + } + Assert.assertTrue("Expected CorruptSSTableException for truncated chunk, but none was thrown.", exceptionThrown); + }); + } + + @Test + public void uncompressedReads() + { + testUncompressedReads(false); + } + + @Test + public void scanUncompressedReads() + { + testUncompressedReads(true); + } + + private void testUncompressedReads(boolean forScan) + { + int maxCompressedLength = 0; // force uncompressed path (chunk.length > maxCompressedLength) + CompressionParams uncompressed = CompressionParams.lz4(DEFAULT_CHUNK_LENGTH, maxCompressedLength); + + var optionGen = writerOptions(); + var paramsGen = Gens.constant(uncompressed); + var fileLengthGen = Gens.longs().between(1, 1 << 16); + + testReads(optionGen, paramsGen, fileLengthGen, forScan); + } + + @Test + public void largeFilePositionReads() throws Exception + { + long FOUR_GB = 1L << 32; + int chunkLength = DEFAULT_CHUNK_LENGTH; + + // Test critical boundaries for compressed file offsets + long[] testOffsets = { + FOUR_GB - 1024, // Just below 4GB + FOUR_GB, // Exactly 4GB + FOUR_GB + 1024, // Just above 4GB + FOUR_GB + (1L << 30), // 5GB (4GB + 1GB) + FOUR_GB * 2 // 8GB + }; + + CompressionParams params = CompressionParams.lz4(chunkLength); // Production default compressor + + for (long offset : testOffsets) + testReadAtLargeOffset(offset, chunkLength, params); + } + + private static void testReads(Gen optionGen, Gen paramsGen, Gen.LongGen totalBytesGen, boolean forScan) + { + qt().withSeed(seed).forAll(Gens.random(), optionGen, paramsGen).check((rs, option, params) -> { + long totalBytesToWrite = totalBytesGen.nextLong(rs); + List chunks = generateRandomChunks(rs, params, totalBytesToWrite); + + File file = new File(JAVA_IO_TMPDIR.getString() + "data.bin"); + try (CompressedSequentialWriter writer = getCompressedSequentialWriter(option, params, file)) + { + for (ByteBuffer chunk : chunks) + writer.write(chunk.duplicate()); + + writer.sync(); + + CompressionMetadata metadata = writer.open(0); + readAndVerifyChunks(file, metadata, chunks, totalBytesToWrite, forScan); + } + finally + { + Files.deleteIfExists(file.toPath()); + } + }); + } + + private static void readAndVerifyChunks(File file, CompressionMetadata metadata, List expectedChunksData, long totalBytesExpected, + boolean forScan) + { + ByteBuffer readBuffer = ByteBuffer.allocateDirect(metadata.chunkLength()); + + try (ChannelProxy channel = new ChannelProxy(file, ChannelProxy.IOMode.DIRECT); + CompressedChunkReader reader = new CompressedChunkReader.Direct(channel, metadata, () -> 1d); + metadata) + { + if (forScan) + reader.forScan(); + + long currentFileOffset = 0; + long totalBytesRead = 0; + int currentChunkIndex = 0; + + while (totalBytesRead < totalBytesExpected) + { + ByteBuffer currentExpectedChunk = expectedChunksData.get(currentChunkIndex); + + readBuffer.clear(); + reader.readChunk(currentFileOffset, readBuffer); + + Assert.assertTrue("Read buffer is empty unexpectedly at offset " + currentFileOffset, readBuffer.hasRemaining()); + + int actualBytesRead = readBuffer.remaining(); + int expectedBytes = currentExpectedChunk.remaining(); + + Assert.assertTrue("Read buffer remaining (" + actualBytesRead + ") is less than expected (" + expectedBytes + ") at offset " + currentFileOffset, + actualBytesRead >= expectedBytes); + + int originalReadBufferLimit = readBuffer.limit(); + readBuffer.limit(expectedBytes); + Assert.assertEquals("Mismatched data at offset " + currentFileOffset, currentExpectedChunk, readBuffer); + readBuffer.limit(originalReadBufferLimit); + + totalBytesRead += expectedBytes; + currentFileOffset += metadata.chunkLength(); + currentChunkIndex++; + } + } + finally + { + MemoryUtil.clean(readBuffer); + } + } + + private static List generateRandomChunks(RandomSource rs, CompressionParams params, long bytesToWrite) + { + List chunks = new ArrayList<>(); + long bytesGenerated = 0; + while (bytesGenerated < bytesToWrite) + { + ByteBuffer chunkBuffer = ByteBuffer.allocate(params.chunkLength()); + int bytesToFill = (int) Math.min(chunkBuffer.capacity(), bytesToWrite - bytesGenerated); + byte[] tempBytes = new byte[bytesToFill]; + rs.nextBytes(tempBytes); + chunkBuffer.put(tempBytes); + chunkBuffer.flip(); + chunks.add(chunkBuffer); + bytesGenerated += bytesToFill; + } + return chunks; + } + + private static CompressedSequentialWriter getCompressedSequentialWriter(SequentialWriterOption option, CompressionParams params, File dataFile) + { + return new CompressedSequentialWriter(dataFile, new File("file.offset"), new File("file.digest"), option, params, new MetadataCollector(new ClusteringComparator())); + } + + /** + * Test reading from a large file offset to verify no integer overflow. + * Creates a simulated compressed file with a chunk at a very large offset (>4GB) + */ + private static void testReadAtLargeOffset(long chunkOffset, int chunkLength, CompressionParams params) throws Exception + { + byte[] uncompressedData = new byte[chunkLength]; + for (int i = 0; i < chunkLength; i++) + uncompressedData[i] = (byte) (i % 256); // Repeating pattern for realistic compression + ByteBuffer uncompressedBuf = ByteBuffer.wrap(uncompressedData); + + // Compress the data + ByteBuffer compressedBuf = ByteBuffer.allocate(chunkLength * 2); + params.getSstableCompressor().compress(uncompressedBuf, compressedBuf); + compressedBuf.flip(); + byte[] compressedData = new byte[compressedBuf.remaining()]; + compressedBuf.get(compressedData); + + // Create a fake channel that simulates a compressed file with data at a large offset + FakeLargeFileChannel fakeChannel = new FakeLargeFileChannel(compressedData, chunkOffset); + + // Create a temp file for ChannelProxy (the file itself isn't used, only the fake channel) + File tmpFile = FileUtils.createTempFile("large_file_test", ".bin"); + + try + { + // Create metadata indicating uncompressed position 0 maps to compressed offset at chunkOffset + try (ChannelProxy channelProxy = new ChannelProxy(tmpFile, fakeChannel); CompressionMetadata metadata = createLargeFileMetadata(0, chunkLength, chunkOffset, compressedData.length, params); CompressedChunkReader directReader = new CompressedChunkReader.Direct(channelProxy, metadata, () -> 0d)) + { + // Read uncompressed position 0, which will read from the large compressed file offset + ByteBuffer readBuffer = ByteBuffer.allocateDirect(chunkLength); + try + { + directReader.readChunk(0, readBuffer); + + // Verify we got a full chunk back + Assert.assertTrue("Read buffer should have data for chunk at offset " + chunkOffset, readBuffer.hasRemaining()); + Assert.assertEquals("Read buffer should contain full chunk", chunkLength, readBuffer.remaining()); + + // Verify decompressed data matches original + byte[] readData = new byte[chunkLength]; + readBuffer.get(readData); + Assert.assertArrayEquals("Data mismatch for chunk at offset " + chunkOffset, + uncompressedData, + readData); + } + finally + { + MemoryUtil.clean(readBuffer); + } + } + } + finally + { + Files.deleteIfExists(tmpFile.toPath()); + } + } + + private static CompressionMetadata createLargeFileMetadata(long chunkIndex, int chunkLength, long chunkOffset, + int compressedLength, CompressionParams params) + { + // Allocate space for chunk offset index: need current offset + next offset to calculate length + Memory chunkOffsets = Memory.allocate(2 * Long.BYTES); + + // Chunk offset index stores where each compressed chunk starts in the file + chunkOffsets.setLong(0, chunkOffset); // This chunk's compressed offset + chunkOffsets.setLong(8, chunkOffset + compressedLength + CHECKSUM_LENGTH); // Next chunk offset (current + data + checksum) + + long dataLength = (chunkIndex + 1) * chunkLength; // Total uncompressed data (one chunk = 16KB) + long compressedFileLength = chunkOffset + compressedLength + CHECKSUM_LENGTH; // Total compressed file size + + File chunksIndexFile = new File(JAVA_IO_TMPDIR.getString() + "large_file_chunks.index"); + + return new CompressionMetadata(chunksIndexFile, + params, + chunkOffsets, + chunkOffsets.size(), + dataLength, + compressedFileLength, + null); + } + + /** + * Fake FileChannel that simulates a large compressed file (>4GB) without requiring actual disk space. + * This allows testing integer overflow scenarios by: + * - Reporting a large file size (e.g., 8GB) via size() + * - Storing only the actual compressed chunk data in memory (a few KB) + * - Mapping reads at high offsets (>4GB) to the in-memory data + * - Returning zeros for reads outside the compressed data region + * This approach enables efficient testing of large file position handling without creating multi-GB test files. + */ + private static class FakeLargeFileChannel extends FileChannel + { + private final byte[] compressedData; + private final long chunkOffset; + private final long fileSize; + private long position; + + FakeLargeFileChannel(byte[] compressedData, long chunkOffset) + { + this.compressedData = compressedData; + this.chunkOffset = chunkOffset; + // File size = offset to chunk + compressed chunk data + CRC32 checksum + // This represents the minimum file size needed to contain the chunk at the specified offset + this.fileSize = chunkOffset + compressedData.length + CHECKSUM_LENGTH; + } + + @Override + public int read(ByteBuffer dst, long readPosition) + { + int bytesToRead = dst.remaining(); + + // Simulate reading from file: return actual data at chunk offset, zeros elsewhere + for (int i = 0; i < bytesToRead; i++) + { + long fileOffset = readPosition + i; + + if (fileOffset >= chunkOffset && fileOffset < chunkOffset + compressedData.length) + { + // Reading from compressed data region + int dataOffset = (int) (fileOffset - chunkOffset); + dst.put(compressedData[dataOffset]); + } + else if (fileOffset >= chunkOffset + compressedData.length && fileOffset < fileSize) + { + // Reading from checksum region (4 bytes after compressed data) + dst.put((byte) 0); // Dummy checksum + } + else + { + // Reading outside our chunk - return padding + dst.put((byte) 0); + } + } + + return bytesToRead; + } + + @Override + public int read(ByteBuffer dst) + { + int read = read(dst, position); + position += read; + return read; + } + + @Override + public long size() + { + return fileSize; + } + + @Override + public long position() + { + return position; + } + + @Override + public FileChannel position(long newPosition) + { + this.position = newPosition; + return this; + } + + // Unsupported operations + @Override + public long read(ByteBuffer[] dsts, int offset, int length) + { + throw new UnsupportedOperationException(); + } + + @Override + public int write(ByteBuffer src) + { + throw new UnsupportedOperationException(); + } + + @Override + public int write(ByteBuffer src, long position) + { + throw new UnsupportedOperationException(); + } + + @Override + public long write(ByteBuffer[] srcs, int offset, int length) + { + throw new UnsupportedOperationException(); + } + + @Override + public FileChannel truncate(long size) + { + throw new UnsupportedOperationException(); + } + + @Override + public void force(boolean metaData) + { + } + + @Override + public long transferTo(long position, long count, WritableByteChannel target) + { + throw new UnsupportedOperationException(); + } + + @Override + public long transferFrom(ReadableByteChannel src, long position, long count) + { + throw new UnsupportedOperationException(); + } + + @Override + public MappedByteBuffer map(FileChannel.MapMode mode, long position, long size) + { + throw new UnsupportedOperationException(); + } + + @Override + public FileLock lock(long position, long size, boolean shared) + { + throw new UnsupportedOperationException(); + } + + @Override + public FileLock tryLock(long position, long size, boolean shared) + { + throw new UnsupportedOperationException(); + } + + @Override + protected void implCloseChannel() + { + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolderTest.java b/test/unit/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolderTest.java new file mode 100644 index 0000000000..98f1814cb8 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/util/DirectThreadLocalByteBufferHolderTest.java @@ -0,0 +1,101 @@ +/* + * 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.nio.ByteBuffer; + +import org.junit.Assert; +import org.junit.Test; + +public class DirectThreadLocalByteBufferHolderTest +{ + @Test + public void testGetBuffer() + { + int blockSize = 4096; + int bufferSize = blockSize * 4; + + DirectThreadLocalByteBufferHolder holder = new DirectThreadLocalByteBufferHolder(blockSize); + + // Initial buffer creation + ByteBuffer buffer = holder.getBuffer(bufferSize); + Assert.assertEquals(bufferSize, buffer.limit()); + Assert.assertEquals(0, buffer.position()); + buffer.put(new byte[bufferSize]); + + // Reuse same buffer + ByteBuffer sameBuffer = holder.getBuffer(bufferSize); + Assert.assertSame(buffer, sameBuffer); + Assert.assertEquals(0, sameBuffer.position()); + } + + @Test + public void testBufferGrows() + { + int blockSize = 4096; + DirectThreadLocalByteBufferHolder holder = new DirectThreadLocalByteBufferHolder(blockSize); + + ByteBuffer small = holder.getBuffer(blockSize); + ByteBuffer large = holder.getBuffer(blockSize * 4); + + Assert.assertNotSame(small, large); + Assert.assertEquals(blockSize * 4, large.capacity()); + } + + @Test + public void testAlignmentRoundsUp() + { + int blockSize = 4096; + DirectThreadLocalByteBufferHolder holder = new DirectThreadLocalByteBufferHolder(blockSize); + + // Request non-aligned size + ByteBuffer buffer = holder.getBuffer(blockSize + 100); + + // Should be rounded up to next block boundary + Assert.assertEquals(blockSize * 2, buffer.capacity()); + Assert.assertEquals(blockSize * 2, buffer.limit()); + } + + @Test + public void testSharedAcrossHolders() + { + int blockSize = 4096; + + DirectThreadLocalByteBufferHolder holder1 = new DirectThreadLocalByteBufferHolder(blockSize); + ByteBuffer buffer1 = holder1.getBuffer(blockSize); + + DirectThreadLocalByteBufferHolder holder2 = new DirectThreadLocalByteBufferHolder(blockSize); + ByteBuffer buffer2 = holder2.getBuffer(blockSize); + + // Same block size = same underlying thread-local = same buffer + Assert.assertSame(buffer1, buffer2); + } + + @Test + public void testDifferentBlockSizesAreSeparate() + { + DirectThreadLocalByteBufferHolder holder4k = new DirectThreadLocalByteBufferHolder(4096); + DirectThreadLocalByteBufferHolder holder8k = new DirectThreadLocalByteBufferHolder(8192); + + ByteBuffer buffer4k = holder4k.getBuffer(4096); + ByteBuffer buffer8k = holder8k.getBuffer(8192); + + Assert.assertNotSame(buffer4k, buffer8k); + } + +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java new file mode 100644 index 0000000000..84b60bc5c8 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java @@ -0,0 +1,64 @@ +/* + * 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.Arrays; + +import org.junit.Test; + +import org.apache.cassandra.utils.Pair; + +public class DirectThreadLocalReadAheadBufferTest extends ThreadLocalReadAheadBufferTest +{ + + @Test + public void testBufferSizes() + { + int blockSize = FileUtils.getFileBlockSize(files[0]); + + for (Integer bufferSize : Arrays.asList(blockSize, // block equal alignment + blockSize / 2, // sub-block alignment + blockSize + (blockSize / 2), // straddling alignment + blockSize * 16)) // multi-block alignment + { + qt().withFixedSeed(seed).forAll(reads()) + .checkAssert(propertyInputs -> testReads(propertyInputs, bufferSize, blockSize)); + } + } + + @Override + protected void testReads(InputData propertyInputs) + { + int blockSize = FileUtils.getFileBlockSize(propertyInputs.file); + testReads(propertyInputs, blockSize * 64, blockSize); + } + + private void testReads(InputData propertyInputs, int bufferSize, int blockSize) + { + try (ChannelProxy bufferedChannel = new ChannelProxy(propertyInputs.file); + ChannelProxy directChannel = new ChannelProxy(propertyInputs.file, ChannelProxy.IOMode.DIRECT)) + { + ThreadLocalReadAheadBuffer tlrab = new DirectThreadLocalReadAheadBuffer(directChannel, bufferSize, blockSize); + for (Pair read : propertyInputs.positionsAndLengths) + { + testRead(read, bufferedChannel, tlrab); + } + tlrab.close(); + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java index 853564688f..6bac7e7078 100644 --- a/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java @@ -293,7 +293,9 @@ public class RandomAccessReaderTest final File f = writeFile(params); FileHandle.Builder builder = new FileHandle.Builder(f).bufferType(params.bufferType) .bufferSize(params.bufferSize); - builder.mmapped(params.mmappedRegions); + if (params.mmappedRegions) + builder.mmapped(); + try (FileHandle fh = builder.complete(); RandomAccessReader reader = fh.createReader()) { @@ -510,21 +512,6 @@ public class RandomAccessReaderTest testSkipBytes(params, numberOfExpectationsInBufferSize + 1); } - public void testSkipBytesNonPositive() throws IOException - { - Parameters params = new Parameters(8192, 4096); - final File f = writeFile(params); - FileHandle.Builder builder = new FileHandle.Builder(f).bufferType(params.bufferType) - .bufferSize(params.bufferSize) - .mmapped(params.mmappedRegions); - try (FileHandle fh = builder.complete(); - RandomAccessReader reader = fh.createReader()) - { - assertEquals(0, reader.skipBytes(0)); - assertEquals(0, reader.skipBytes(-1)); - } - } - @Test(expected = IOException.class) public void testSkipBytesClosed() throws IOException { @@ -544,8 +531,11 @@ public class RandomAccessReaderTest { final File f = writeFile(params); FileHandle.Builder builder = new FileHandle.Builder(f).bufferType(params.bufferType) - .bufferSize(params.bufferSize) - .mmapped(params.mmappedRegions); + .bufferSize(params.bufferSize); + + if (params.mmappedRegions) + builder.mmapped(); + try (FileHandle fh = builder.complete(); RandomAccessReader reader = fh.createReader()) { diff --git a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java similarity index 52% rename from test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java rename to test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java index 17ce00e421..afad1cd117 100644 --- a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java @@ -26,7 +26,6 @@ import org.assertj.core.api.Assertions; import org.junit.Assert; import org.junit.Test; -import accord.utils.Gen; import accord.utils.Gens; import org.apache.cassandra.config.DatabaseDescriptor; @@ -40,7 +39,7 @@ import org.apache.cassandra.utils.memory.MemoryUtil; import static accord.utils.Property.qt; -public class CompressedChunkReaderTest +public class StandardCompressedChunkReaderTest extends CompressedChunkReaderTestBase { static { @@ -50,17 +49,16 @@ public class CompressedChunkReaderTest @Test public void scanReaderReadsLessThanRAReader() { - var optionGen = options(); - var paramsGen = params(); + var optionGen = writerOptions(); + var paramsGen = compressionParams(Gens.constant(CompressionParams.DEFAULT_CHUNK_LENGTH)); var lengthGen = Gens.longs().between(1, 1 << 16); - qt().withSeed(-1871070464864118891L).forAll(Gens.random(), optionGen, paramsGen).check((rs, option, params) -> { + + qt().forAll(Gens.random(), optionGen, paramsGen).check((rs, option, params) -> { ListenableFileSystem fs = FileSystems.newGlobalInMemoryFileSystem(); - File f = new File("/file.db"); + File f = new File("/file.bin"); AtomicInteger reads = new AtomicInteger(); - fs.onPostRead(f.path::equals, (p, c, pos, dst, r) -> { - reads.incrementAndGet(); - }); + fs.onPostRead(f.path::equals, (p, c, pos, dst, r) -> reads.incrementAndGet()); long length = lengthGen.nextLong(rs); CompressionMetadata metadata1, metadata2; try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, new File("/file.offset"), new File("/file.digest"), option, params, new MetadataCollector(new ClusteringComparator()))) @@ -73,71 +71,45 @@ public class CompressedChunkReaderTest metadata2 = writer.open(0); } - doReads(f, metadata1, length, true); + doReads(f, metadata1, length, false); + int raReads = reads.getAndSet(0); + + doReads(f, metadata2, length, true); int scanReads = reads.getAndSet(0); - doReads(f, metadata2, length, false); - int raReads = reads.getAndSet(0); - if (Files.size(f.toPath()) > DatabaseDescriptor.getCompressedReadAheadBufferSize()) - Assert.assertTrue(scanReads < raReads); + Assert.assertTrue(scanReads <= raReads); }); } - private void doReads(File f, CompressionMetadata metadata, long length, boolean useReadAhead) + protected void doReads(File f, CompressionMetadata metadata, long length, boolean useReadAhead) { ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength()); - try (ChannelProxy channel = new ChannelProxy(f); - CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1.1); - metadata) + try (ChannelProxy channel = new ChannelProxy(f)) { - if (useReadAhead) - reader.forScan(); - - long offset = 0; - long maxOffset = length * Long.BYTES; - do + try (CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1d); + metadata) { - reader.readChunk(offset, buffer); - for (long expected = offset / Long.BYTES; buffer.hasRemaining(); expected++) - Assertions.assertThat(buffer.getLong()).isEqualTo(expected); + if (useReadAhead) + reader.forScan(); - offset += metadata.chunkLength(); + long offset = 0; + long maxOffset = length * Long.BYTES; + do + { + reader.readChunk(offset, buffer); + for (long expected = offset / Long.BYTES; buffer.hasRemaining(); expected++) + Assertions.assertThat(buffer.getLong()).isEqualTo(expected); + + offset += metadata.chunkLength(); + } + while (offset < maxOffset); } - while (offset < maxOffset); } finally { MemoryUtil.clean(buffer); - }} - - private static Gen options() - { - Gen bufferSizes = Gens.constant(1 << 10); //.pickInt(1 << 4, 1 << 10, 1 << 15); - return rs -> SequentialWriterOption.newBuilder() - .finishOnClose(false) - .bufferSize(bufferSizes.next(rs)) - .build(); - } - - private enum CompressionKind { Noop, Snappy, Deflate, Lz4, Zstd } - - private static Gen params() - { - Gen chunkLengths = Gens.constant(CompressionParams.DEFAULT_CHUNK_LENGTH); - Gen compressionRatio = Gens.pick(1.1D); - return rs -> { - CompressionKind kind = rs.pick(CompressionKind.values()); - switch (kind) - { - case Noop: return CompressionParams.noop(); - case Snappy: return CompressionParams.snappy(chunkLengths.next(rs), compressionRatio.next(rs)); - case Deflate: return CompressionParams.deflate(chunkLengths.next(rs)); - case Lz4: return CompressionParams.lz4(chunkLengths.next(rs)); - case Zstd: return CompressionParams.zstd(chunkLengths.next(rs)); - default: throw new UnsupportedOperationException(kind.name()); - } - }; + } } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java index 80031ae5e5..a4a7f8d519 100644 --- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java @@ -46,9 +46,9 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TM public class ThreadLocalReadAheadBufferTest implements WithQuickTheories { private static final int numFiles = 5; - private static final File[] files = new File[numFiles]; private static final Logger logger = LoggerFactory.getLogger(ThreadLocalReadAheadBufferTest.class); - private static Integer seed; + protected static final File[] files = new File[numFiles]; + protected static Integer seed; @BeforeClass public static void setup() @@ -74,7 +74,7 @@ public class ThreadLocalReadAheadBufferTest implements WithQuickTheories } catch (Exception e) { - // ignore + // ignore } } } @@ -89,83 +89,87 @@ public class ThreadLocalReadAheadBufferTest implements WithQuickTheories @Test public void testReadsLikeChannelProxy() { - qt().withFixedSeed(seed).forAll(reads()) .checkAssert(this::testReads); } - private void testReads(InputData propertyInputs) + protected void testReads(InputData propertyInputs) { - try (ChannelProxy channel = new ChannelProxy(propertyInputs.file)) + try (ChannelProxy channel = new ChannelProxy(propertyInputs.file); + ThreadLocalReadAheadBuffer tlrab = new ThreadLocalReadAheadBuffer(channel, new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(), BufferType.OFF_HEAP); ) { - ThreadLocalReadAheadBuffer trlab = new ThreadLocalReadAheadBuffer(channel, new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(), BufferType.OFF_HEAP); for (Pair read : propertyInputs.positionsAndLengths) { - int readSize = Math.min(read.right,(int) (channel.size() - read.left)); - ByteBuffer buf1 = ByteBuffer.allocate(readSize); - channel.read(buf1, read.left); - - ByteBuffer buf2 = ByteBuffer.allocate(readSize); - try - { - int copied = 0; - while (copied < readSize) { - trlab.fill(read.left + copied); - int leftToRead = readSize - copied; - if (trlab.remaining() >= leftToRead) - copied += trlab.read(buf2, leftToRead); - else - copied += trlab.read(buf2, trlab.remaining()); - } - } - catch (CorruptSSTableException e) - { - throw new RuntimeException(e); - } - - Assert.assertEquals(buf1, buf2); + testRead(read, channel, tlrab); } } } - private Gen reads() + protected static void testRead(Pair read, ChannelProxy bufferedChannel, ThreadLocalReadAheadBuffer tlrab) + { + int readSize = Math.min(read.right, (int) (bufferedChannel.size() - read.left)); + ByteBuffer buf1 = ByteBuffer.allocate(readSize); + bufferedChannel.read(buf1, read.left); + + ByteBuffer buf2 = ByteBuffer.allocate(readSize); + try + { + int copied = 0; + while (copied < readSize) + { + tlrab.fill(read.left + copied); + int leftToRead = readSize - copied; + if (tlrab.remaining() >= leftToRead) + copied += tlrab.read(buf2, leftToRead); + else + copied += tlrab.read(buf2, tlrab.remaining()); + } + } + catch (CorruptSSTableException e) + { + throw new RuntimeException(e); + } + + Assert.assertEquals(buf1, buf2); + } + + protected Gen reads() { return arbitrary().pick(List.of(files)) .flatMap((file) -> lists().of(longs().between(0, fileSize(file)).zip(integers().between(1, 100), Pair::create)) .ofSizeBetween(5, 10) .map(positionsAndLengths -> new InputData(file, positionsAndLengths))); - } - private Gen lastBlockReads() + protected Gen lastBlockReads() { int blockSize = new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(); return arbitrary().pick(List.of(files)) - .flatMap((file) -> - lists().of(longs().between(max(0, fileSize(file) - blockSize), fileSize(file)).zip(integers().between(1, 100), Pair::create)) - .ofSizeBetween(5, 10) - .map(positionsAndLengths -> new InputData(file, positionsAndLengths))); - + .flatMap((file) -> + lists().of(longs().between(max(0, fileSize(file) - blockSize), fileSize(file)).zip(integers().between(1, 100), Pair::create)) + .ofSizeBetween(5, 10) + .map(positionsAndLengths -> new InputData(file, positionsAndLengths))); } - // need this becasue generators don't handle the IOException + // need this because generators don't handle the IOException private long fileSize(File file) { try { return Files.size(file.toPath()); - } catch (IOException e) + } + catch (IOException e) { throw new RuntimeException(e); } } - private static class InputData + protected static class InputData { - private final File file; - private final List> positionsAndLengths; + protected final File file; + protected final List> positionsAndLengths; public InputData(File file, List> positionsAndLengths) { @@ -201,5 +205,4 @@ public class ThreadLocalReadAheadBufferTest implements WithQuickTheories return file; } - } diff --git a/test/unit/org/apache/cassandra/service/StartupChecksTest.java b/test/unit/org/apache/cassandra/service/StartupChecksTest.java index 7d5ba0e40c..6141802b57 100644 --- a/test/unit/org/apache/cassandra/service/StartupChecksTest.java +++ b/test/unit/org/apache/cassandra/service/StartupChecksTest.java @@ -529,4 +529,13 @@ public class StartupChecksTest assertTrue(e.getMessage().contains(message)); } } + + @Test + public void testFindDirectIOUnsupportedLocationsSkipsNonExistentDirs() + { + // Non-existent directories should be skipped, not added to unsupported list + List unsupported = StartupChecks.findDirectIOUnsupportedLocations( + new String[] { "/this/path/does/not/exist/for/testing" }); + assertThat(unsupported).isEmpty(); + } }