Caleb Rackliffe created CASSANDRA-21684:
-------------------------------------------
Summary: Unbounded per-thread cache of read-ahead buffers can
cause slow heap leak
Key: CASSANDRA-21684
URL: https://issues.apache.org/jira/browse/CASSANDRA-21684
Project: Apache Cassandra
Issue Type: Bug
Components: Local/Compaction
Reporter: Caleb Rackliffe
Assignee: Caleb Rackliffe
{{ThreadLocalReadAheadBuffer}} caches one read-ahead Block per scanned file in
a per-thread map (blockMap), keyed by file path, so a thread can reuse its
buffer across repeated scans of the same SSTable. That map has no upper bound
and entries are only removed when the same thread that populated one also
happens to run the file's final close, which rarely lines up, since SSTables
are typically scanned by several different pool threads (compaction, repair,
reads) over their lifetime. The result: every long-lived thread accumulates a
permanent stale entry for every distinct SSTable it has ever scanned, growing
unbounded as compaction continuously creates new SSTable generations.
{noformat}
@Test
public void releasingScanResourcesShouldRemoveThreadLocalBlockMapEntry() throws
Exception
{
SequentialWriterOption writerOption =
SequentialWriterOption.newBuilder().finishOnClose(false).bufferSize(1 <<
10).build();
CompressionParams params = CompressionParams.snappy(4096, 1.1);
FileSystems.newGlobalInMemoryFileSystem();
File f = new File("/scan_release_repro.db");
File offsets = new File("/scan_release_repro.offset");
File digest = new File("/scan_release_repro.digest");
long longsToWrite = 600;
CompressionMetadata metadata;
try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f,
offsets, digest, writerOption, params, new MetadataCollector(new
ClusteringComparator())))
{
for (long i = 0; i < longsToWrite; i++)
writer.writeLong(i);
writer.sync();
metadata = writer.open(0);
}
DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256);
ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength());
try (ChannelProxy channel = new ChannelProxy(f);
CompressedChunkReader reader = new
CompressedChunkReader.Standard(channel, metadata, () -> 1.1);
metadata)
{
reader.forScan();
reader.readChunk(0, buffer);
Assert.assertTrue("expected a read-ahead Block to be present for this
file on this thread after a scan read",
threadLocalBlockMapContainsKey(channel.filePath()));
// Mirrors what BufferManagingRebufferer.closeReader() does at the end
of every scan.
reader.releaseUnderlyingResources();
Assert.assertFalse("a stale Block entry for this file should not remain
in this thread's read-ahead " +
"blockMap once the scan has released its resources
-- otherwise every distinct " +
"file this thread ever scans leaks a permanent
entry",
threadLocalBlockMapContainsKey(channel.filePath()));
}
finally
{
FileUtils.clean(buffer);
}
}
@SuppressWarnings("unchecked")
private static boolean threadLocalBlockMapContainsKey(String filePath) throws
Exception
{
Field field = ThreadLocalReadAheadBuffer.class.getDeclaredField("blockMap");
field.setAccessible(true);
FastThreadLocal<Map<String, ?>> blockMap = (FastThreadLocal<Map<String,
?>>) field.get(null);
return blockMap.get().containsKey(filePath);
}
{noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
