Introduces a thread-local read-ahead buffer for sequential scan workloads (compaction, range queries) over compressed SSTables. The existing path issued one channel.read() syscall per compressed chunk (~64 KiB); by pre-reading 256 KiB at a time and serving subsequent chunk reads from that in-process buffer, the patch cuts syscall count ~4× for sequential scans without touching the random-access path.
BEFORE (all reads)
══════════════════════════════════════════════════════════
SSTableScanner
│ openDataReader()
▼
RandomAccessReader → BufferManagingRebufferer.Aligned
│
▼
CompressedChunkReader.Standard
│ ONE channel.read() per compressed chunk
│ Scanning 1 GiB ≈ 17,000 syscalls
▼
ChannelProxy → OS kernel → disk
AFTER (scan path)
══════════════════════════════════════════════════════════
SSTableScanner
│ openDataReaderForScan() ← NEW
▼
RandomAccessReader → BufferManagingRebufferer.Aligned
│ └── closeReader() → releaseUnderlyingResources() ← NEW
▼
CompressedChunkReader.Standard
│ forScan() → allocate thread-local block ← NEW
│
│ ScanCompressedReader.read(chunk)
│ └─► ThreadLocalReadAheadBuffer.fill(pos)
│ ├── same 256 KiB block? → 0 syscalls (cache hit)
│ └── new 256 KiB block? → ONE channel.read() for 256 KiB
│ Scanning 1 GiB ≈ 4,096 syscalls (4× fewer)
▼
ChannelProxy → OS kernel → disk
┌─────────────────────────────────────────────────────────┐
│ SSTableScanner │
│ openDataReaderForScan() [NEW] │
└──────────────────────────┬──────────────────────────────┘
│
┌──────────────────────────▼──────────────────────────────┐
│ FileHandle │
│ createReaderForScan() [NEW] │
│ instantiateRebufferer(limiter, forScan) [NEW overload] │
└──────────────────────────┬──────────────────────────────┘
│ RebuffererFactory.instantiateRebufferer(isScan)
┌──────────────────────────▼──────────────────────────────┐
│ CompressedChunkReader (abstract) │
│ forScan() [NEW] │
│ releaseUnderlyingResources() [NEW, via ChunkReader] │
│ │
│ ┌────────────────────────────────────────────────┐ │
│ │ Standard (channel-based) │ │
│ │ reader: RandomAccessCompressedReader │ │
│ │ scanReader: ScanCompressedReader [NEW, null] │ │
│ └────────────────────────────────────────────────┘ │
│ │
│ CompressedReader interface [NEW] │
│ ├── RandomAccessCompressedReader [extracted] │
│ └── ScanCompressedReader [NEW] │
│ └─► ThreadLocalReadAheadBuffer [NEW] │
└─────────────────────────────────────────────────────────┘
Compressed SSTables store variable-length compressed blocks. The old code called channel.read() for exactly the size of each block. Since consecutive blocks are adjacent on disk, the OS fetched many pages per call that Cassandra immediately discarded. The read-ahead buffer fixes this by reading a large aligned window once and serving consecutive chunk reads from memory.
┌──────────────────────────────────────────────────────────┐
│ ThreadLocalReadAheadBuffer (instance) │
│ channel: ChannelProxy (immutable) │
│ channelSize: long (snapshot at construct) │
│ bufferSize: int (e.g. 256 KiB) │
│ bufferType: BufferType │
│ │
│ static FastThreadLocal<Map<filePath → Block>> │
│ Block { │
│ buffer: ByteBuffer (null until allocateBuffer()) │
│ index: int (-1 = no valid data) │
│ } │
│ ← one Block per thread per file path │
└──────────────────────────────────────────────────────────┘
The FastThreadLocal is static — all instances across all open SSTables share the same slot. Within that slot is a HashMap<String, Block> keyed by file path, so each thread holds independent buffers for multiple simultaneous files.
ScanCompressedReader.read(chunk)
│
┌──────────────── while copied < chunk.length ─────────────────┐
│ │
│ readAheadBuffer.fill(chunk.offset + copied) │
│ │ │
│ │ blockNo = position / bufferSize │
│ │ blockPos = blockNo * bufferSize │
│ │ │
│ │ block.index == blockNo? │
│ │ ┌──Yes──────────────────────────────────┐ │
│ │ │ reposition buffer at realPos (0 I/O) │ │
│ │ └───────────────────────────────────────┘ │
│ │ ┌──No───────────────────────────────────┐ │
│ │ │ channel.read(block, blockPos) │ │
│ │ │ (ONE 256 KiB syscall) │ │
│ │ │ block.index = blockNo │ │
│ │ └───────────────────────────────────────┘ │
│ │
│ readAheadBuffer.read(dest, min(remaining, leftToRead)) │
│ (loop handles chunks that span a block boundary) │
└──────────────────────────────────────────────────────────────┘
│
│ [optional CRC check]
│ return compressed ByteBuffer
│ (valid only until next call — buffer is reused!)
Standard.forScan()
│
▼ [ALLOCATED: Block.buffer live, Block.index = -1]
│
per-chunk read/fill loop
│
▼
BufferManagingRebufferer.closeReader()
→ releaseUnderlyingResources()
→ deallocateResources()
→ FileUtils.clean(buffer) ← native memory freed promptly
→ Block.buffer = null
│
▼ [DEALLOCATED]
│
Standard.close() → scanReader.close() → readAheadBuffer.close()
→ blockMap.remove(filePath) ← prevents map growth across open/close cycles
Two-phase shutdown: closeReader() frees native memory at end-of-scan; close() removes the map entry to prevent a growing HashMap across SSTable lifecycle events.
cassandra.yaml:
compressed_read_ahead_buffer_size: 256KiB (0 = disabled)
Constraints:
- 0 → feature disabled (scanReader stays null)
- 1..255 KiB → ConfigurationException (must be ≥ 256 KiB or 0)
- ≤ chunkLength → disabled for that table (no reduction in syscalls)
JMX: StorageServiceMBean.get/setCompressedReadAheadBufferInKB()
(takes effect for newly opened scan readers only)
BEFORE: Standard.readChunk()
─────────────────────────────────────────
getBuffer(chunkLen + 4)
channel.read(compressed, chunk.offset) ← one syscall per chunk
[CRC check]
compressor.uncompress(compressed, buf)
Fields: ThreadLocalByteBufferHolder bufferHolder (one, for all reads)
AFTER: Standard.readChunk()
─────────────────────────────────────────
CompressedReader readFrom =
(scanReader != null && scanReader.allocated()) ? scanReader : reader;
│
┌──────┴──────┐
scan │ │ random access
▼ ▼
ScanCompressed RandomAccessCompressed
Reader Reader (unchanged)
│
└─► ThreadLocalReadAheadBuffer
(0 or 1 syscall per 256 KiB block)
Fields: CompressedReader reader (random access)
CompressedReader scanReader (scan, nullable)
Layer Change
─────────────────────────────────────────────────────────────
RebuffererFactory instantiateRebufferer() → (boolean isScan)
ChunkReader +releaseUnderlyingResources() default no-op
CompressedChunkReader +forScan(), +CompressedReader interface,
readChunk() dispatches via CompressedReader
Standard reader/scanReader dual-path; +close() override
FileHandle +createReaderForScan(), overloaded
instantiateRebufferer(limiter, forScan)
SSTableReader +openDataReaderForScan()
SSTableScanner uses openDataReaderForScan() instead of
openDataReader()
BufferManagingRebufferer closeReader() calls
source.releaseUnderlyingResources() [NEW]
Config +compressed_read_ahead_buffer_size (256KiB default)
DatabaseDescriptor +get/set accessors + validation
StorageService(MBean) +JMX getters/setters
-
Scan detection is explicit, not heuristic.
isScanis threaded fromSSTableScanner→FileHandle→RebuffererFactory→CompressedChunkReader. The call site knows it's scanning; no pattern detection latency. -
Static
FastThreadLocalkeyed by file path means one map slot per thread covers all open SSTables simultaneously, with no synchronization on the hot path. -
close()must be called on the allocating thread.blockMap.get().remove(filePath)operates on the calling thread's map. Ifclose()is called from a different thread (e.g., a cleanup thread), it silently misses and the off-heap buffer leaks. -
channelSizeis snapshotted at construction. The last-block read is capped to this snapshot. If the file grows after construction (not expected for SSTables), the last block may be under-read. -
read()returns a buffer valid only until the next call. The buffer is reused; callers must consume it immediately.BufferManagingRebuffererdoes this correctly today, but it's an easy trap for future callers. -
Test coverage uses property-based testing.
ThreadLocalReadAheadBufferTestuses QuickTheories to verify byte-for-byte identical output vs. directchannel.read().CompressedChunkReaderTestvalidates that the scan path issues strictly fewer syscalls than the random-access path. -
Mmap path is unaffected.
CompressedChunkReader.Mmapreads from memory-mapped regions with no syscalls, so it neither benefits from nor participates in the read-ahead path. -
JMX config changes apply to newly opened scan readers only. Existing
CompressedChunkReaderinstances retain the buffer size from their construction time.
A one-file follow-up to CASSANDRA-15452. The original clear(boolean deallocate) called getBlock(), which used computeIfAbsent internally — creating a new Block entry in the thread-local map even when the buffer had never been used. Any CompressedChunkReader closed without ever scanning would silently allocate a Block object as a side effect of teardown. The fix replaces computeIfAbsent with a plain get and adds two early-exit null guards.
BEFORE: clear(boolean deallocate)
──────────────────────────────────────────────────────────────
Block block = getBlock();
└─► blockMap.get()
.computeIfAbsent(filePath, Block::new) ← ALLOCATES
if absent!
block.index = -1;
ByteBuffer blockBuffer = block.buffer;
if (blockBuffer != null) { ... } ← null if never used → no-op
but Block is now in map
close() on an unused reader:
┌─────────────────────────────────────────────────────┐
│ clear(true) → computeIfAbsent → new Block() added │
│ then remove(filePath) → Block discarded │
│ Net: unnecessary allocation + pointless map churn │
└─────────────────────────────────────────────────────┘
clear(false) on unused reader (no remove follows):
┌─────────────────────────────────────────────────────┐
│ new Block() left in map with index=-1, buffer=null │
│ Stale entry persists until close() removes it │
└─────────────────────────────────────────────────────┘
AFTER: clear(boolean deallocate)
──────────────────────────────────────────────────────────────
Block block = blockMap.get().get(channel.filePath());
^^^
get(), not computeIfAbsent
if (block == null)
return; ← early exit if never allocated
block.index = -1;
if (block.buffer == null)
return; ← early exit if buffer not yet live
ByteBuffer blockBuffer = block.buffer;
blockBuffer.clear();
if (deallocate) { FileUtils.clean(blockBuffer); block.buffer = null; }
The call chain for every CompressedChunkReader teardown is close() → clear(true). With the old code, even a reader that was opened but never used for scanning would allocate a Block on the way out. In a busy cluster with many SSTable opens and closes (compaction, streaming, repairs), this was a steady trickle of allocations on the teardown path.
CASSANDRA-15452 CASSANDRA-20551
──────────────────────────────────────────────────────────
Introduced ThreadLocalReadAheadBuffer Follow-up fix to 15452
+ static FastThreadLocal<Map> Fixes clear() / close()
+ getBlock() uses computeIfAbsent on the never-allocated
path
▲
│ getBlock() in clear() was
│ the root cause — introduced
│ by 15452, fixed here
CASSANDRA-20551 also directly addresses one of the key invariants identified in 15452: the blockMap must not accumulate stale entries. The fix eliminates a source of such entries on the teardown path.