natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adopt Unsafe::invokeCleaner for Direct ByteBuffer cleaning 

Patch by Samuel D Lightfoot; Reviewed by Ariel Weisberg and Abe Ratnosfky for CASSANDRA-20677
This commit is contained in:
samlightfoot 2025-05-25 11:04:15 +01:00 committed by Ariel Weisberg
parent 7c9af51cfa
commit 58ee847968
35 changed files with 197 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Adopt Unsafe::invokeCleaner for Direct ByteBuffer cleaning (CASSANDRA-20677)
* Add additional metrics around hints (CASSANDRA-20499)
* Support for add and replace in IntervalTree (CASSANDRA-20513)
* Enable single_sstable_uplevel by default for LCS (CASSANDRA-18509)

4
src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java
View File

@ -27,10 +27,10 @@ import java.nio.file.StandardOpenOption;
import net.openhft.chronicle.core.util.ThrowingFunction;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.SimpleCachedBufferPool;
import org.apache.cassandra.utils.NativeLibrary;
import org.apache.cassandra.utils.SyncUtil;
import org.apache.cassandra.utils.memory.MemoryUtil;
/*
* Memory-mapped segment. Maps the destination channel into an appropriately-sized memory-mapped buffer in which the
@ -108,7 +108,7 @@ public class MemoryMappedSegment extends CommitLogSegment
@Override
protected void internalClose()
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
super.internalClose();
}

8
src/java/org/apache/cassandra/db/streaming/CompressedInputStream.java
View File

@ -30,10 +30,10 @@ import com.google.common.primitives.Ints;
import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RebufferingInputStream;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.utils.ChecksumType;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static java.lang.Math.max;
import static java.lang.String.format;
@ -146,7 +146,7 @@ public class CompressedInputStream extends RebufferingInputStream implements Aut
// with poorly compressible data, it's possible for a compressed chunk to be larger than
// configured uncompressed chunk size - depending on data, min_compress_ratio, and compressor;
// we may need to resize the compressed buffer.
FileUtils.clean(compressedChunk);
MemoryUtil.clean(compressedChunk);
compressedChunk = ByteBuffer.allocateDirect(max((int) (compressedChunk.capacity() * GROWTH_FACTOR), chunkLength));
}
@ -212,13 +212,13 @@ public class CompressedInputStream extends RebufferingInputStream implements Aut
{
if (null != buffer)
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
buffer = null;
}
if (null != compressedChunk)
{
FileUtils.clean(compressedChunk);
MemoryUtil.clean(compressedChunk);
compressedChunk = null;
}
}

4
src/java/org/apache/cassandra/db/tries/InMemoryTrie.java
View File

@ -28,11 +28,11 @@ import com.google.common.annotations.VisibleForTesting;
import org.agrona.concurrent.UnsafeBuffer;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.github.jamm.MemoryMeterStrategy;
/**
@ -191,7 +191,7 @@ public class InMemoryTrie<T> extends InMemoryReadTrie<T>
for (UnsafeBuffer b : buffers)
{
if (b != null)
FileUtils.clean(b.byteBuffer());
MemoryUtil.clean(b.byteBuffer());
}
}

3
src/java/org/apache/cassandra/hints/ChecksummedDataInput.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.NativeLibrary;
import org.apache.cassandra.utils.memory.MemoryUtil;
/**
* A {@link RandomAccessReader} wrapper that calculates the CRC in place.
@ -241,7 +242,7 @@ public class ChecksummedDataInput extends RebufferingInputStream
@Override
public void close()
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
channel.close();
}

4
src/java/org/apache/cassandra/hints/HintsBuffer.java
View File

@ -32,10 +32,10 @@ import java.util.zip.CRC32;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.utils.FBUtilities.updateChecksum;
import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
@ -101,7 +101,7 @@ final class HintsBuffer
void free()
{
FileUtils.clean(slab);
MemoryUtil.clean(slab);
}
/**

4
src/java/org/apache/cassandra/index/sasi/utils/MappedBuffer.java
View File

@ -24,8 +24,8 @@ import java.nio.channels.FileChannel.MapMode;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.io.util.ChannelProxy;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.utils.memory.MemoryUtil;
import com.google.common.annotations.VisibleForTesting;
@ -225,7 +225,7 @@ public class MappedBuffer implements Closeable
try
{
for (MappedByteBuffer segment : pages)
FileUtils.clean(segment);
MemoryUtil.clean(segment);
}
catch (Exception e)
{

9
src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
View File

@ -32,11 +32,11 @@ import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.util.ChecksumWriter;
import org.apache.cassandra.io.util.DataPosition;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.io.util.SequentialWriterOption;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.utils.Throwables.merge;
@ -246,7 +246,7 @@ public class CompressedSequentialWriter extends SequentialWriter
int chunkSize = (int) (metadataWriter.chunkOffsetBy(realMark.nextChunkIndex) - chunkOffset - 4);
if (compressed.capacity() < chunkSize)
{
FileUtils.clean(compressed);
MemoryUtil.clean(compressed);
compressed = compressor.preferredBufferType().allocate(chunkSize);
}
@ -401,7 +401,10 @@ public class CompressedSequentialWriter extends SequentialWriter
accumulate = super.doPreCleanup(accumulate);
if (compressed != null)
{
try { FileUtils.clean(compressed); }
try
{
MemoryUtil.clean(compressed);
}
catch (Throwable t) { accumulate = merge(accumulate, t); }
compressed = null;
}

2
src/java/org/apache/cassandra/io/util/BufferedDataOutputStreamPlus.java
View File

@ -323,7 +323,7 @@ public class BufferedDataOutputStreamPlus extends DataOutputStreamPlus
doFlush(0);
channel.close();
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
buffer = null;
}

4
src/java/org/apache/cassandra/io/util/DataOutputBuffer.java
View File

@ -29,6 +29,7 @@ import com.google.common.base.Preconditions;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.config.CassandraRelevantProperties.DATA_OUTPUT_BUFFER_ALLOCATE_TYPE;
import static org.apache.cassandra.config.CassandraRelevantProperties.DOB_DOUBLING_THRESHOLD_MB;
@ -183,7 +184,8 @@ public class DataOutputBuffer extends BufferedDataOutputStreamPlus
protected void setBuffer(ByteBuffer newBuffer)
{
FileUtils.clean(buffer); // free if direct
// free if direct
MemoryUtil.clean(buffer);
buffer = newBuffer;
}

4
src/java/org/apache/cassandra/io/util/FileInputStreamPlus.java
View File

@ -24,6 +24,8 @@ import java.nio.channels.FileChannel;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import org.apache.cassandra.utils.memory.MemoryUtil;
public class FileInputStreamPlus extends RebufferingInputStream
{
final FileChannel channel;
@ -76,7 +78,7 @@ public class FileInputStreamPlus extends RebufferingInputStream
{
try
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
finally
{

65
src/java/org/apache/cassandra/io/util/FileUtils.java
View File

@ -21,10 +21,6 @@ import java.io.BufferedWriter;
import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
@ -75,39 +71,6 @@ public final class FileUtils
private static final DecimalFormat df = new DecimalFormat("#.##");
private static final Class clsDirectBuffer;
private static final MethodHandle mhDirectBufferCleaner;
private static final MethodHandle mhCleanerClean;
static
{
try
{
clsDirectBuffer = Class.forName("sun.nio.ch.DirectBuffer");
Method mDirectBufferCleaner = clsDirectBuffer.getMethod("cleaner");
mhDirectBufferCleaner = MethodHandles.lookup().unreflect(mDirectBufferCleaner);
Method mCleanerClean = mDirectBufferCleaner.getReturnType().getMethod("clean");
mhCleanerClean = MethodHandles.lookup().unreflect(mCleanerClean);
ByteBuffer buf = ByteBuffer.allocateDirect(1);
clean(buf);
}
catch (IllegalAccessException e)
{
logger.error("FATAL: Cassandra is unable to access required classes. This usually means it has been " +
"run without the aid of the standard startup scripts or the scripts have been edited. If this was " +
"intentional, and you are attempting to use Java 11+ you may need to add the --add-exports and " +
"--add-opens jvm options from either jvm11-server.options or jvm11-client.options", e);
throw new RuntimeException(e); // causes ExceptionInInitializerError, will prevent startup
}
catch (Throwable t)
{
logger.error("FATAL: Cannot initialize optimized memory deallocator.", t);
JVMStabilityInspector.inspectThrowable(t);
throw new RuntimeException(t); // causes ExceptionInInitializerError, will prevent startup
}
}
private static final File tempDir = new File(JAVA_IO_TMPDIR.getString());
private static final AtomicLong tempFileNum = new AtomicLong();
@ -358,34 +321,6 @@ public final class FileUtils
return folder.isAncestorOf(file);
}
public static void clean(ByteBuffer buffer)
{
if (buffer == null || !buffer.isDirect())
return;
// TODO Once we can get rid of Java 8, it's simpler to call sun.misc.Unsafe.invokeCleaner(ByteBuffer),
// but need to take care of the attachment handling (i.e. whether 'buf' is a duplicate or slice) - that
// is different in sun.misc.Unsafe.invokeCleaner and this implementation.
try
{
Object cleaner = mhDirectBufferCleaner.bindTo(buffer).invoke();
if (cleaner != null)
{
// ((DirectBuffer) buf).cleaner().clean();
mhCleanerClean.bindTo(cleaner).invoke();
}
}
catch (RuntimeException e)
{
throw e;
}
catch (Throwable e)
{
throw new RuntimeException(e);
}
}
public static long parseFileSize(String value)
{
long result;

3
src/java/org/apache/cassandra/io/util/MmappedRegions.java
View File

@ -32,6 +32,7 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.RefCounted;
import org.apache.cassandra.utils.concurrent.SharedCloseableImpl;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static java.util.stream.Stream.of;
import static org.apache.cassandra.utils.Throwables.perform;
@ -400,7 +401,7 @@ public class MmappedRegions extends SharedCloseableImpl
() ->
{
if (buffer != null)
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}));
}
}

4
src/java/org/apache/cassandra/io/util/NIODataInputStream.java
View File

@ -24,6 +24,8 @@ import java.nio.channels.SeekableByteChannel;
import com.google.common.base.Preconditions;
import org.apache.cassandra.utils.memory.MemoryUtil;
/**
* Rough equivalent of BufferedInputStream and DataInputStream wrapping the input stream of a File or Socket
* Created to work around the fact that when BIS + DIS delegate to NIO for socket IO they will allocate large
@ -73,7 +75,7 @@ public class NIODataInputStream extends RebufferingInputStream
{
channel.close();
super.close();
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
buffer = null;
}

6
src/java/org/apache/cassandra/io/util/SequentialWriter.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.utils.SyncUtil;
import org.apache.cassandra.utils.concurrent.Transactional;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.utils.Throwables.merge;
@ -77,7 +78,10 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
if (buffer != null)
{
try { FileUtils.clean(buffer); }
try
{
MemoryUtil.clean(buffer);
}
catch (Throwable t) { accumulate = merge(accumulate, t); }
buffer = null;
}

5
src/java/org/apache/cassandra/io/util/SimpleCachedBufferPool.java
View File

@ -23,6 +23,7 @@ import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.jctools.queues.MpmcArrayQueue;
/**
@ -92,7 +93,7 @@ public class SimpleCachedBufferPool
// We use a bounded queue. By consequence if we have reached the maximum size for the buffer pool
// offer will return false and we know that we can simply get rid of the buffer.
if (!bufferPool.offer(buffer))
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
/**
@ -103,7 +104,7 @@ public class SimpleCachedBufferPool
ByteBuffer buffer = bufferPool.poll();
while(buffer != null)
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
buffer = bufferPool.poll();
}
}

3
src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java
View File

@ -24,6 +24,7 @@ import java.util.EnumMap;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.memory.MemoryUtil;
/**
* Utility class that allow buffers to be reused by storing them in a thread local instance.
@ -73,7 +74,7 @@ public final class ThreadLocalByteBufferHolder
ByteBuffer buffer = reusableBB.get();
if (buffer.capacity() < size)
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
buffer = bufferType.allocate(size);
reusableBB.set(buffer);
}

3
src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
View File

@ -25,6 +25,7 @@ import java.util.Map;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.utils.memory.MemoryUtil;
public final class ThreadLocalReadAheadBuffer
{
@ -144,7 +145,7 @@ public final class ThreadLocalReadAheadBuffer
blockBuffer.clear();
if (deallocate)
{
FileUtils.clean(blockBuffer);
MemoryUtil.clean(blockBuffer);
block.buffer = null;
}
}

4
src/java/org/apache/cassandra/journal/ActiveSegment.java
View File

@ -28,11 +28,11 @@ import java.util.concurrent.locks.LockSupport;
import com.codahale.metrics.Timer;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@ -235,7 +235,7 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
@Override
void onUnreferenced()
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
try
{
channel.close();

5
src/java/org/apache/cassandra/journal/OnDiskIndex.java
View File

@ -33,6 +33,7 @@ import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.Crc;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.journal.Journal.validateCRC;
import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
@ -97,7 +98,7 @@ final class OnDiskIndex<K> extends Index<K>
}
catch (Throwable e)
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
FileUtils.closeQuietly(channel);
throw new JournalReadError(descriptor, file, e);
}
@ -117,7 +118,7 @@ final class OnDiskIndex<K> extends Index<K>
{
try
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
buffer = null;
channel.close();
}

5
src/java/org/apache/cassandra/journal/StaticSegment.java
View File

@ -35,6 +35,7 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.Closeable;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.memory.MemoryUtil;
/**
* An immutable data segment that is no longer written to.
@ -207,7 +208,7 @@ public final class StaticSegment<K, V> extends Segment<K, V>
@Override
void onUnreferenced()
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
FileUtils.closeQuietly(channel);
index.close();
if (discard)
@ -335,7 +336,7 @@ public final class StaticSegment<K, V> extends Segment<K, V>
public void close()
{
FileUtils.closeQuietly(channel);
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
public abstract boolean advance();

4
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -58,7 +58,7 @@ import org.apache.cassandra.io.UnversionedSerializer;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.memory.MemoryUtil;
/**
* Utility methods to make ByteBuffers less painful
@ -904,7 +904,7 @@ public class ByteBufferUtil
{
if (!allowBufferResize)
throw new IllegalStateException(String.format("output buffer is not large enough for data: current capacity %d, required %d", buf.capacity(), outputLength));
FileUtils.clean(buf);
MemoryUtil.clean(buf);
buf = bufferType.allocate(outputLength);
}
else

3
src/java/org/apache/cassandra/utils/MerkleTree.java
View File

@ -40,7 +40,6 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.memory.MemoryUtil;
@ -987,7 +986,7 @@ public class MerkleTree
Object attachment = MemoryUtil.getAttachment(buffer);
if (attachment instanceof Ref.DirectBufferRef)
((Ref.DirectBufferRef) attachment).release();
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
abstract int hashBytesOffset();

5
src/java/org/apache/cassandra/utils/memory/BufferPool.java
View File

@ -46,7 +46,6 @@ import org.slf4j.LoggerFactory;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.metrics.BufferPoolMetrics;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.Shared;
@ -816,7 +815,7 @@ public class BufferPool
if (chunk == null)
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
updateOverflowMemoryUsage(-size);
}
else
@ -1563,7 +1562,7 @@ public class BufferPool
if (parent != null)
parent.free(slab);
else
FileUtils.clean(slab);
MemoryUtil.clean(slab);
}
static void unsafeRecycle(Chunk chunk)

13
src/java/org/apache/cassandra/utils/memory/MemoryUtil.java
View File

@ -25,6 +25,7 @@ import java.nio.ByteOrder;
import com.sun.jna.Native;
import sun.misc.Unsafe;
import sun.nio.ch.DirectBuffer;
public abstract class MemoryUtil
{
@ -322,4 +323,16 @@ public abstract class MemoryUtil
{
getBytes(sourceAddress, targetBuffer, 0, length);
}
public static void clean(ByteBuffer buffer)
{
if (buffer == null || !buffer.isDirect())
return;
DirectBuffer db = (DirectBuffer) buffer;
if (db.attachment() != null)
return; // duplicate or slice
unsafe.invokeCleaner(buffer);
}
}

3
src/java/org/apache/cassandra/utils/memory/SlabAllocator.java
View File

@ -26,7 +26,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.OpOrder;
@ -116,7 +115,7 @@ public class SlabAllocator extends MemtableBufferAllocator
public void setDiscarded()
{
for (Region region : offHeapRegions)
FileUtils.clean(region.data);
MemoryUtil.clean(region.data);
super.setDiscarded();
}

4
test/long/org/apache/cassandra/hints/HintsWriteThenReadTest.java
View File

@ -36,11 +36,11 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.memory.MemoryUtil;
import static org.apache.cassandra.Util.dk;
import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
@ -152,7 +152,7 @@ public class HintsWriteThenReadTest
{
write(session, timestamp);
}
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
private void write(HintsWriter.Session session, long timestamp) throws IOException

9
test/unit/org/apache/cassandra/db/MmapFileTest.java
View File

@ -32,6 +32,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.memory.MemoryUtil;
public class MmapFileTest
{
@ -86,8 +87,8 @@ public class MmapFileTest
buffer.putInt(42);
buffer.putInt(42);
buffer.putInt(42);
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
mmapCount = (Long) mbs.getAttribute(bpmName, "Count");
@ -114,7 +115,7 @@ public class MmapFileTest
buffer.putInt(42);
buffer.putInt(42);
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
mmapCount = (Long) mbs.getAttribute(bpmName, "Count");
@ -139,7 +140,7 @@ public class MmapFileTest
buffer.putInt(42);
buffer.putInt(42);
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
mmapCount = (Long) mbs.getAttribute(bpmName, "Count");

4
test/unit/org/apache/cassandra/hints/HintsMaker.java
View File

@ -44,11 +44,11 @@ import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.memory.MemoryUtil;
public class HintsMaker
{
@ -202,7 +202,7 @@ public class HintsMaker
}
finally
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
}

4
test/unit/org/apache/cassandra/hints/HintsReaderTest.java
View File

@ -45,11 +45,11 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.exceptions.UnknownTableException;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaTestUtil;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.hamcrest.Matchers;
import static org.junit.Assert.assertEquals;
@ -99,7 +99,7 @@ public class HintsReaderTest
session.append(Hint.create(m, timestamp));
}
}
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
Assert.assertThat(descriptor.hintsFileSize(directory), Matchers.greaterThan(0L));

4
test/unit/org/apache/cassandra/hints/HintsStoreTest.java
View File

@ -41,8 +41,8 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.hamcrest.Matchers;
import static org.junit.Assert.assertEquals;
@ -176,7 +176,7 @@ public class HintsStoreTest
for (int i = 0; i < hintsCount; i++)
session.append(createHint(i, hintCreationTime));
}
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}
Assert.assertThat(descriptor.hintsFileSize(directory), Matchers.greaterThan(0L));
return new File(directory, descriptor.fileName()).lastModified(); // hint file last modified time

3
test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.io.filesystem.ListenableFileSystem;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.assertj.core.api.Assertions;
import org.junit.Assert;
@ -107,7 +108,7 @@ public class CompressedChunkReaderTest
}
finally
{
FileUtils.clean(buffer);
MemoryUtil.clean(buffer);
}}
private static Gen<SequentialWriterOption> options()

8
test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java
View File

@ -37,8 +37,8 @@ import net.jpountz.lz4.LZ4Factory;
import net.jpountz.lz4.LZ4SafeDecompressor;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.memory.MemoryUtil;
public class StreamCompressionSerializerTest
{
@ -64,9 +64,9 @@ public class StreamCompressionSerializerTest
public void tearDown()
{
if (input != null)
FileUtils.clean(input);
MemoryUtil.clean(input);
if (compressed != null)
FileUtils.clean(compressed);
MemoryUtil.clean(compressed);
if (output != null && output.refCnt() > 0)
output.release(output.refCnt());
}
@ -104,7 +104,7 @@ public class StreamCompressionSerializerTest
StreamCompressionSerializer.serialize(compressor, input, VERSION)
.write(size -> {
if (compressed != null)
FileUtils.clean(compressed);
MemoryUtil.clean(compressed);
return compressed = ByteBuffer.allocateDirect(size);
});
input.flip();

4
test/unit/org/apache/cassandra/transport/WriteBytesTest.java
View File

@ -22,8 +22,8 @@ import org.junit.Test;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.Generators;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.assertj.core.api.Assertions;
import static org.quicktheories.QuickTheory.qt;
@ -51,7 +51,7 @@ public class WriteBytesTest
Assertions.assertThat(buf.writerIndex()).isEqualTo(size);
for (int i = 0; i < size; i++)
Assertions.assertThat(buf.getByte(buf.readerIndex() + i)).describedAs("byte mismatch at index %d", i).isEqualTo(bb.get(bb.position() + i));
FileUtils.clean(bb);
MemoryUtil.clean(bb);
});
}

104
test/unit/org/apache/cassandra/utils/memory/MemoryUtilTest.java Normal file
View File

@ -0,0 +1,104 @@
/*
* 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.utils.memory;
import java.lang.management.BufferPoolMXBean;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.assertj.core.api.Assertions;
public class MemoryUtilTest
{
@Test
public void testClean()
{
// We assert direct pool state changes as no ByteBuffer state changes are observable as a result of cleaner
// executing.
BufferPoolMXBean directPool = getDirectBufferPool();
int bufferSize = 1024 * 1024; // some non-insignificant size
ByteBuffer original = ByteBuffer.allocateDirect(bufferSize);
long memoryUsedBefore = directPool.getMemoryUsed();
MemoryUtil.clean(original);
long memoryUsedAfter = directPool.getMemoryUsed();
Assert.assertEquals("Direct memory used should decrease by buffer capacity",
// Allow 5% tolerance for other activities
memoryUsedBefore - bufferSize, memoryUsedAfter, bufferSize * 0.05);
}
@Test
public void testCleanViewDoesNotThrow()
{
// Use a large buffer to likely get mmap'd memory from the OS. This ensures that if cleaning a view incorrectly
// unmaps the original buffer's memory, subsequent access to 'original' would more reliably fail.
// For context: glibc's mmap threshold is 32MB on 64-bit systems
ByteBuffer original = ByteBuffer.allocateDirect(64 * 1024 * 1024);
ByteBuffer slice = original.slice();
MemoryUtil.clean(slice);
try
{
original.putInt(10);
}
catch (Exception exc)
{
Assertions.fail("Unable to write to original buffer after cleaning (slice). " + exc.getMessage(), exc);
}
ByteBuffer duplicate = original.duplicate();
MemoryUtil.clean(duplicate);
try
{
original.putInt(10);
}
catch (Exception exc)
{
Assertions.fail("Unable to write to original buffer after cleaning (duplicate). " + exc.getMessage(), exc);
}
}
@Test
public void testCleanNonDirectDoesNotThrow()
{
ByteBuffer original = ByteBuffer.allocate(16);
MemoryUtil.clean(original);
}
private static BufferPoolMXBean getDirectBufferPool()
{
List<BufferPoolMXBean> pools = ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class);
for (BufferPoolMXBean pool : pools)
{
if (pool.getName().equals("direct"))
{
return pool;
}
}
throw new IllegalArgumentException("Direct buffer pool not found");
}
}