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

Remove FileCacheService, instead pooling buffers 

patch by stefania and benedict for CASSANDRA-8897
This commit is contained in:
stefania 2015-06-09 10:08:24 +01:00 committed by Benedict Elliott Smith
parent e182217b20
commit 17dd4ccc78
33 changed files with 2602 additions and 827 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0:
* Make file buffer cache independent of paths being read (CASSANDRA-8897)
* Remove deprecated legacy Hadoop code (CASSANDRA-9353)
* Decommissioned nodes will not rejoin the cluster (CASSANDRA-8801)
* Change gossip stabilization to use endpoit size (CASSANDRA-9401)

11
conf/cassandra.yaml
View File

@ -328,10 +328,17 @@ concurrent_reads: 32
concurrent_writes: 32
concurrent_counter_writes: 32
# Total memory to use for sstable-reading buffers. Defaults to
# the smaller of 1/4 of heap or 512MB.
# Maximum memory to use for pooling sstable buffers. Defaults to the smaller
# of 1/4 of heap or 512MB. This pool is allocated off-heap, so is in addition
# to the memory allocated for heap. Memory is only allocated as needed.
# file_cache_size_in_mb: 512
# Flag indicating whether to allocate on or off heap when the sstable buffer
# pool is exhausted, that is when it has exceeded the maximum memory
# file_cache_size_in_mb, beyond which it will not cache buffers but allocate on request.
# buffer_pool_use_heap_if_exhausted: true
# Total permitted memory to use for memtables. Cassandra will stop
# accepting writes when the limit is exceeded until a flush completes,
# and will trigger a flush based on memtable_cleanup_threshold

4
src/java/org/apache/cassandra/config/Config.java
View File

@ -222,7 +222,9 @@ public class Config
private static boolean isClientMode = false;
public Integer file_cache_size_in_mb;
public Integer file_cache_size_in_mb = 512;
public boolean buffer_pool_use_heap_if_exhausted = true;
public boolean inter_dc_tcp_nodelay = true;

5
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -1484,6 +1484,11 @@ public class DatabaseDescriptor
return conf.file_cache_size_in_mb;
}
public static boolean getBufferPoolUseHeapIfExhausted()
{
return conf.buffer_pool_use_heap_if_exhausted;
}
public static long getTotalCommitlogSpaceInMB()
{
return conf.commitlog_total_space_in_mb;

43
src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
View File

@ -32,6 +32,7 @@ import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.memory.BufferPool;
/**
* CRAR extends RAR to transparently uncompress blocks from the file into RAR.buffer. Most of the RAR
@ -41,26 +42,12 @@ public class CompressedRandomAccessReader extends RandomAccessReader
{
public static CompressedRandomAccessReader open(ChannelProxy channel, CompressionMetadata metadata)
{
try
{
return new CompressedRandomAccessReader(channel, metadata, null);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
return new CompressedRandomAccessReader(channel, metadata, null);
}
public static CompressedRandomAccessReader open(ICompressedFile file)
{
try
{
return new CompressedRandomAccessReader(file.channel(), file.getMetadata(), file);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
return new CompressedRandomAccessReader(file.channel(), file.getMetadata(), file);
}
private final TreeMap<Long, MappedByteBuffer> chunkSegments;
@ -76,34 +63,36 @@ public class CompressedRandomAccessReader extends RandomAccessReader
// raw checksum bytes
private ByteBuffer checksumBytes;
protected CompressedRandomAccessReader(ChannelProxy channel, CompressionMetadata metadata, ICompressedFile file) throws FileNotFoundException
protected CompressedRandomAccessReader(ChannelProxy channel, CompressionMetadata metadata, ICompressedFile file)
{
super(channel, metadata.chunkLength(), metadata.compressedFileLength, metadata.compressor().preferredBufferType(), file instanceof PoolingSegmentedFile ? (PoolingSegmentedFile) file : null);
super(channel, metadata.chunkLength(), metadata.compressedFileLength, metadata.compressor().preferredBufferType());
this.metadata = metadata;
checksum = new Adler32();
chunkSegments = file == null ? null : file.chunkSegments();
if (chunkSegments == null)
{
compressed = super.allocateBuffer(metadata.compressor().initialCompressedBufferLength(metadata.chunkLength()), metadata.compressor().preferredBufferType());
compressed = allocateBuffer(metadata.compressor().initialCompressedBufferLength(metadata.chunkLength()), metadata.compressor().preferredBufferType());
checksumBytes = ByteBuffer.wrap(new byte[4]);
}
}
@Override
protected ByteBuffer allocateBuffer(int bufferSize, BufferType bufferType)
protected int getBufferSize(int size)
{
assert Integer.bitCount(bufferSize) == 1;
return bufferType.allocate(bufferSize);
assert Integer.bitCount(size) == 1; //must be a power of two
return size;
}
@Override
public void deallocate()
public void close()
{
super.deallocate();
super.close();
if (compressed != null)
FileUtils.clean(compressed);
compressed = null;
{
BufferPool.put(compressed);
compressed = null;
}
}
private void reBufferStandard()

14
src/java/org/apache/cassandra/io/compress/CompressedThrottledReader.java
View File

@ -20,9 +20,6 @@ package org.apache.cassandra.io.compress;
*
*/
import java.io.FileNotFoundException;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.io.util.ChannelProxy;
@ -32,7 +29,7 @@ public class CompressedThrottledReader extends CompressedRandomAccessReader
{
private final RateLimiter limiter;
public CompressedThrottledReader(ChannelProxy channel, CompressionMetadata metadata, ICompressedFile file, RateLimiter limiter) throws FileNotFoundException
public CompressedThrottledReader(ChannelProxy channel, CompressionMetadata metadata, ICompressedFile file, RateLimiter limiter)
{
super(channel, metadata, file);
this.limiter = limiter;
@ -46,13 +43,6 @@ public class CompressedThrottledReader extends CompressedRandomAccessReader
public static CompressedThrottledReader open(ICompressedFile file, RateLimiter limiter)
{
try
{
return new CompressedThrottledReader(file.channel(), file.getMetadata(), file, limiter);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
return new CompressedThrottledReader(file.channel(), file.getMetadata(), file, limiter);
}
}

59
src/java/org/apache/cassandra/io/sstable/KeyIterator.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.io.sstable;
import java.io.DataInput;
import java.io.File;
import java.io.IOException;
@ -31,12 +32,59 @@ import org.apache.cassandra.utils.CloseableIterator;
public class KeyIterator extends AbstractIterator<DecoratedKey> implements CloseableIterator<DecoratedKey>
{
private final RandomAccessReader in;
private final static class In
{
private final File path;
private RandomAccessReader in;
public In(File path)
{
this.path = path;
}
private void maybeInit()
{
if (in == null)
in = RandomAccessReader.open(path);
}
public DataInput get()
{
maybeInit();
return in;
}
public boolean isEOF()
{
maybeInit();
return in.isEOF();
}
public void close()
{
if (in != null)
in.close();
}
public long getFilePointer()
{
maybeInit();
return in.getFilePointer();
}
public long length()
{
maybeInit();
return in.length();
}
}
private final In in;
public KeyIterator(Descriptor desc)
{
File path = new File(desc.filenameFor(Component.PRIMARY_INDEX));
in = RandomAccessReader.open(path);
in = new In(new File(desc.filenameFor(Component.PRIMARY_INDEX)));
}
protected DecoratedKey computeNext()
@ -45,8 +93,9 @@ public class KeyIterator extends AbstractIterator<DecoratedKey> implements Close
{
if (in.isEOF())
return endOfData();
DecoratedKey key = StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(in));
RowIndexEntry.Serializer.skip(in); // skip remainder of the entry
DecoratedKey key = StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(in.get()));
RowIndexEntry.Serializer.skip(in.get()); // skip remainder of the entry
return key;
}
catch (IOException e)

60
src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java
View File

@ -33,42 +33,55 @@ import org.apache.cassandra.utils.MergeIterator;
*/
public class ReducingKeyIterator implements CloseableIterator<DecoratedKey>
{
private final IMergeIterator<DecoratedKey,DecoratedKey> mi;
private final ArrayList<KeyIterator> iters;
private IMergeIterator<DecoratedKey,DecoratedKey> mi;
public ReducingKeyIterator(Collection<SSTableReader> sstables)
{
ArrayList<KeyIterator> iters = new ArrayList<KeyIterator>(sstables.size());
iters = new ArrayList<>(sstables.size());
for (SSTableReader sstable : sstables)
iters.add(new KeyIterator(sstable.descriptor));
mi = MergeIterator.get(iters, DecoratedKey.comparator, new MergeIterator.Reducer<DecoratedKey,DecoratedKey>()
}
private void maybeInit()
{
if (mi == null)
{
DecoratedKey reduced = null;
@Override
public boolean trivialReduceIsTrivial()
mi = MergeIterator.get(iters, DecoratedKey.comparator, new MergeIterator.Reducer<DecoratedKey,DecoratedKey>()
{
return true;
}
DecoratedKey reduced = null;
public void reduce(DecoratedKey current)
{
reduced = current;
}
@Override
public boolean trivialReduceIsTrivial()
{
return true;
}
protected DecoratedKey getReduced()
{
return reduced;
}
});
public void reduce(DecoratedKey current)
{
reduced = current;
}
protected DecoratedKey getReduced()
{
return reduced;
}
});
}
}
public void close() throws IOException
{
mi.close();
if (mi != null)
{
mi.close();
}
}
public long getTotalBytes()
{
maybeInit();
long m = 0;
for (Iterator<DecoratedKey> iter : mi.iterators())
{
@ -79,6 +92,8 @@ public class ReducingKeyIterator implements CloseableIterator<DecoratedKey>
public long getBytesRead()
{
maybeInit();
long m = 0;
for (Iterator<DecoratedKey> iter : mi.iterators())
{
@ -87,18 +102,15 @@ public class ReducingKeyIterator implements CloseableIterator<DecoratedKey>
return m;
}
public String getTaskType()
{
return "Secondary index build";
}
public boolean hasNext()
{
maybeInit();
return mi.hasNext();
}
public DecoratedKey next()
{
maybeInit();
return mi.next();
}

50
src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java
View File

@ -1,50 +0,0 @@
/*
* 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;
public class BufferedPoolingSegmentedFile extends PoolingSegmentedFile
{
public BufferedPoolingSegmentedFile(ChannelProxy channel, long length)
{
super(new Cleanup(channel), channel, length);
}
private BufferedPoolingSegmentedFile(BufferedPoolingSegmentedFile copy)
{
super(copy);
}
public BufferedPoolingSegmentedFile sharedCopy()
{
return new BufferedPoolingSegmentedFile(this);
}
public static class Builder extends SegmentedFile.Builder
{
public void addPotentialBoundary(long boundary)
{
// only one segment in a standard-io file
}
public SegmentedFile complete(ChannelProxy channel, long overrideLength)
{
long length = overrideLength > 0 ? overrideLength : channel.size();
return new BufferedPoolingSegmentedFile(channel, length);
}
}
}

136
src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
View File

@ -1,136 +0,0 @@
/*
* 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.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.TreeMap;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
import org.apache.cassandra.io.compress.CompressedSequentialWriter;
import org.apache.cassandra.io.compress.CompressedThrottledReader;
import org.apache.cassandra.io.compress.CompressionMetadata;
public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile implements ICompressedFile
{
public final CompressionMetadata metadata;
private final TreeMap<Long, MappedByteBuffer> chunkSegments;
public CompressedPoolingSegmentedFile(ChannelProxy channel, CompressionMetadata metadata)
{
this(channel, metadata, CompressedSegmentedFile.createMappedSegments(channel, metadata));
}
private CompressedPoolingSegmentedFile(ChannelProxy channel, CompressionMetadata metadata, TreeMap<Long, MappedByteBuffer> chunkSegments)
{
super(new Cleanup(channel, metadata, chunkSegments), channel, metadata.dataLength, metadata.compressedFileLength);
this.metadata = metadata;
this.chunkSegments = chunkSegments;
}
private CompressedPoolingSegmentedFile(CompressedPoolingSegmentedFile copy)
{
super(copy);
this.metadata = copy.metadata;
this.chunkSegments = copy.chunkSegments;
}
public ChannelProxy channel()
{
return channel;
}
public TreeMap<Long, MappedByteBuffer> chunkSegments()
{
return chunkSegments;
}
protected static final class Cleanup extends PoolingSegmentedFile.Cleanup
{
final CompressionMetadata metadata;
final TreeMap<Long, MappedByteBuffer> chunkSegments;
protected Cleanup(ChannelProxy channel, CompressionMetadata metadata, TreeMap<Long, MappedByteBuffer> chunkSegments)
{
super(channel);
this.metadata = metadata;
this.chunkSegments = chunkSegments;
}
public void tidy()
{
super.tidy();
metadata.close();
if (chunkSegments != null)
{
for (MappedByteBuffer segment : chunkSegments.values())
FileUtils.clean(segment);
}
}
}
public static class Builder extends CompressedSegmentedFile.Builder
{
public Builder(CompressedSequentialWriter writer)
{
super(writer);
}
public void addPotentialBoundary(long boundary)
{
// only one segment in a standard-io file
}
public SegmentedFile complete(ChannelProxy channel, long overrideLength)
{
return new CompressedPoolingSegmentedFile(channel, metadata(channel.filePath(), overrideLength));
}
}
public void dropPageCache(long before)
{
if (before >= metadata.dataLength)
super.dropPageCache(0);
super.dropPageCache(metadata.chunkFor(before).offset);
}
public RandomAccessReader createReader()
{
return CompressedRandomAccessReader.open(this);
}
public RandomAccessReader createThrottledReader(RateLimiter limiter)
{
return CompressedThrottledReader.open(this, limiter);
}
protected RandomAccessReader createPooledReader()
{
return CompressedRandomAccessReader.open(this);
}
public CompressionMetadata getMetadata()
{
return metadata;
}
public CompressedPoolingSegmentedFile sharedCopy()
{
return new CompressedPoolingSegmentedFile(this);
}
}

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

@ -284,7 +284,11 @@ public class FileUtils
public static void clean(ByteBuffer buffer)
{
if (isCleanerAvailable() && buffer.isDirect())
((DirectBuffer)buffer).cleaner().clean();
{
DirectBuffer db = (DirectBuffer) buffer;
if (db.cleaner() != null)
db.cleaner().clean();
}
}
public static void createDirectory(String directory)

78
src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java
View File

@ -1,78 +0,0 @@
/*
* 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 org.apache.cassandra.service.FileCacheService;
public abstract class PoolingSegmentedFile extends SegmentedFile
{
final FileCacheService.CacheKey cacheKey;
protected PoolingSegmentedFile(Cleanup cleanup, ChannelProxy channel, long length)
{
this(cleanup, channel, length, length);
}
protected PoolingSegmentedFile(Cleanup cleanup, ChannelProxy channel, long length, long onDiskLength)
{
super(cleanup, channel, length, onDiskLength);
cacheKey = cleanup.cacheKey;
}
public PoolingSegmentedFile(PoolingSegmentedFile copy)
{
super(copy);
cacheKey = copy.cacheKey;
}
protected static class Cleanup extends SegmentedFile.Cleanup
{
final FileCacheService.CacheKey cacheKey = new FileCacheService.CacheKey();
protected Cleanup(ChannelProxy channel)
{
super(channel);
}
public void tidy()
{
super.tidy();
FileCacheService.instance.invalidate(cacheKey, channel.filePath());
}
}
@SuppressWarnings("resource")
public FileDataInput getSegment(long position)
{
RandomAccessReader reader = FileCacheService.instance.get(cacheKey);
if (reader == null)
reader = createPooledReader();
reader.seek(position);
return reader;
}
protected RandomAccessReader createPooledReader()
{
return RandomAccessReader.open(channel, length, this);
}
public void recycle(RandomAccessReader reader)
{
FileCacheService.instance.put(cacheKey, reader);
}
}

121
src/java/org/apache/cassandra/io/util/RandomAccessReader.java
View File

@ -20,16 +20,19 @@ package org.apache.cassandra.io.util;
import java.io.*;
import java.nio.ByteBuffer;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.BufferPool;
public class RandomAccessReader extends AbstractDataInput implements FileDataInput
{
// default buffer size, 64Kb
public static final int DEFAULT_BUFFER_SIZE = 65536;
public static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
// the IO channel to the file, we do not own a reference to this due to
// performance reasons (CASSANDRA-9379) so it's up to the owner of the RAR to
// ensure that the channel stays open and that it is closed afterwards
protected final ChannelProxy channel;
// buffer which will cache file blocks
protected ByteBuffer buffer;
@ -38,47 +41,63 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
// `markedPointer` folds the offset of the last file mark
protected long bufferOffset, markedPointer;
protected final ChannelProxy channel;
// this can be overridden at construction to a value shorter than the true length of the file;
// if so, it acts as an imposed limit on reads, rather than a convenience property
private final long fileLength;
protected final PoolingSegmentedFile owner;
protected RandomAccessReader(ChannelProxy channel, int bufferSize, long overrideLength, BufferType bufferType, PoolingSegmentedFile owner)
protected RandomAccessReader(ChannelProxy channel, int bufferSize, long overrideLength, BufferType bufferType)
{
this.channel = channel.sharedCopy();
this.owner = owner;
this.channel = channel;
// allocating required size of the buffer
if (bufferSize <= 0)
throw new IllegalArgumentException("bufferSize must be positive");
// we can cache file length in read-only mode
fileLength = overrideLength <= 0 ? channel.size() : overrideLength;
buffer = allocateBuffer(bufferSize, bufferType);
buffer = allocateBuffer(getBufferSize(bufferSize), bufferType);
buffer.limit(0);
}
protected ByteBuffer allocateBuffer(int bufferSize, BufferType bufferType)
protected int getBufferSize(int size)
{
int size = (int) Math.min(fileLength, bufferSize);
return bufferType.allocate(size);
return (int)Math.min(fileLength, size);
}
public static RandomAccessReader open(ChannelProxy channel, long overrideSize, PoolingSegmentedFile owner)
protected ByteBuffer allocateBuffer(int size, BufferType bufferType)
{
return open(channel, DEFAULT_BUFFER_SIZE, overrideSize, owner);
return BufferPool.get(size, bufferType);
}
// A wrapper of the RandomAccessReader that closes the channel when done.
// For performance reasons RAR does not increase the reference count of
// a channel but assumes the owner will keep it open and close it,
// see CASSANDRA-9379, this thin class is just for those cases where we do
// not have a shared channel.
private static class RandomAccessReaderWithChannel extends RandomAccessReader
{
RandomAccessReaderWithChannel(File file)
{
super(new ChannelProxy(file), DEFAULT_BUFFER_SIZE, -1L, BufferType.OFF_HEAP);
}
@Override
public void close()
{
try
{
super.close();
}
finally
{
channel.close();
}
}
}
public static RandomAccessReader open(File file)
{
try (ChannelProxy channel = new ChannelProxy(file))
{
return open(channel);
}
return new RandomAccessReaderWithChannel(file);
}
public static RandomAccessReader open(ChannelProxy channel)
@ -88,27 +107,12 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
public static RandomAccessReader open(ChannelProxy channel, long overrideSize)
{
return open(channel, DEFAULT_BUFFER_SIZE, overrideSize, null);
return open(channel, DEFAULT_BUFFER_SIZE, overrideSize);
}
@VisibleForTesting
static RandomAccessReader open(ChannelProxy channel, int bufferSize, PoolingSegmentedFile owner)
public static RandomAccessReader open(ChannelProxy channel, int bufferSize, long overrideSize)
{
return open(channel, bufferSize, -1L, owner);
}
private static RandomAccessReader open(ChannelProxy channel, int bufferSize, long overrideSize, PoolingSegmentedFile owner)
{
return new RandomAccessReader(channel, bufferSize, overrideSize, BufferType.ON_HEAP, owner);
}
@VisibleForTesting
static RandomAccessReader open(SequentialWriter writer)
{
try (ChannelProxy channel = new ChannelProxy(writer.getPath()))
{
return open(channel, DEFAULT_BUFFER_SIZE, null);
}
return new RandomAccessReader(channel, bufferSize, overrideSize, BufferType.OFF_HEAP);
}
public ChannelProxy getChannel()
@ -212,32 +216,14 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
@Override
public void close()
{
if (owner == null || buffer == null)
{
// The buffer == null check is so that if the pool owner has deallocated us, calling close()
// will re-call deallocate rather than recycling a deallocated object.
// I'd be more comfortable if deallocate didn't have to handle being idempotent like that,
// but RandomAccessFile.close will call AbstractInterruptibleChannel.close which will
// re-call RAF.close -- in this case, [C]RAR.close since we are overriding that.
deallocate();
}
else
{
owner.recycle(this);
}
}
public void deallocate()
{
//make idempotent
//make idempotent
if (buffer == null)
return;
bufferOffset += buffer.position();
FileUtils.clean(buffer);
buffer = null; // makes sure we don't use this after it's ostensibly closed
channel.close();
bufferOffset += buffer.position();
BufferPool.put(buffer);
buffer = null;
}
@Override
@ -265,6 +251,9 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
if (newPosition < 0)
throw new IllegalArgumentException("new position should not be negative");
if (buffer == null)
throw new IllegalStateException("Attempted to seek in a closed RAR");
if (newPosition >= length()) // it is save to call length() in read-only mode
{
if (newPosition > length())
@ -315,7 +304,7 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
public int read(byte[] buff, int offset, int length)
{
if (buffer == null)
throw new AssertionError("Attempted to read from closed RAR");
throw new IllegalStateException("Attempted to read from closed RAR");
if (length == 0)
return 0;
@ -334,6 +323,10 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
public ByteBuffer readBytes(int length) throws EOFException
{
assert length >= 0 : "buffer length should not be negative: " + length;
if (buffer == null)
throw new IllegalStateException("Attempted to read from closed RAR");
try
{
ByteBuffer result = ByteBuffer.allocate(length);
@ -365,7 +358,7 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
public long getPosition()
{
return bufferOffset + buffer.position();
return bufferOffset + (buffer == null ? 0 : buffer.position());
}
public long getPositionLimit()

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

@ -135,14 +135,14 @@ public abstract class SegmentedFile extends SharedCloseableImpl
*/
public static Builder getBuilder(Config.DiskAccessMode mode, boolean compressed)
{
return compressed ? new CompressedPoolingSegmentedFile.Builder(null)
return compressed ? new CompressedSegmentedFile.Builder(null)
: mode == Config.DiskAccessMode.mmap ? new MmappedSegmentedFile.Builder()
: new BufferedPoolingSegmentedFile.Builder();
: new BufferedSegmentedFile.Builder();
}
public static Builder getCompressedBuilder(CompressedSequentialWriter writer)
{
return new CompressedPoolingSegmentedFile.Builder(writer);
return new CompressedSegmentedFile.Builder(writer);
}
/**

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

@ -44,6 +44,8 @@ import org.apache.cassandra.utils.SyncUtil;
*/
public class SequentialWriter extends OutputStream implements WritableByteChannel, Transactional
{
private static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
// isDirty - true if this.buffer contains any un-synced bytes
protected boolean isDirty = false, syncNeeded = false;
@ -151,12 +153,12 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
*/
public static SequentialWriter open(File file)
{
return new SequentialWriter(file, RandomAccessReader.DEFAULT_BUFFER_SIZE, BufferType.ON_HEAP);
return new SequentialWriter(file, DEFAULT_BUFFER_SIZE, BufferType.ON_HEAP);
}
public static ChecksummedSequentialWriter open(File file, File crcPath)
{
return new ChecksummedSequentialWriter(file, RandomAccessReader.DEFAULT_BUFFER_SIZE, crcPath);
return new ChecksummedSequentialWriter(file, DEFAULT_BUFFER_SIZE, crcPath);
}
public static CompressedSequentialWriter open(String dataFilePath,

15
src/java/org/apache/cassandra/io/util/ThrottledReader.java
View File

@ -21,8 +21,6 @@ package org.apache.cassandra.io.util;
*/
import java.io.FileNotFoundException;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.io.compress.BufferType;
@ -31,9 +29,9 @@ public class ThrottledReader extends RandomAccessReader
{
private final RateLimiter limiter;
protected ThrottledReader(ChannelProxy channel, long overrideLength, RateLimiter limiter) throws FileNotFoundException
protected ThrottledReader(ChannelProxy channel, long overrideLength, RateLimiter limiter)
{
super(channel, RandomAccessReader.DEFAULT_BUFFER_SIZE, overrideLength, BufferType.ON_HEAP, null);
super(channel, RandomAccessReader.DEFAULT_BUFFER_SIZE, overrideLength, BufferType.OFF_HEAP);
this.limiter = limiter;
}
@ -45,13 +43,6 @@ public class ThrottledReader extends RandomAccessReader
public static ThrottledReader open(ChannelProxy channel, long overrideLength, RateLimiter limiter)
{
try
{
return new ThrottledReader(channel, overrideLength, limiter);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
return new ThrottledReader(channel, overrideLength, limiter);
}
}

34
src/java/org/apache/cassandra/metrics/FileCacheMetrics.java → src/java/org/apache/cassandra/metrics/BufferPoolMetrics.java
View File

@ -20,41 +20,29 @@ package org.apache.cassandra.metrics;
import com.codahale.metrics.Gauge;
import com.codahale.metrics.Meter;
import com.codahale.metrics.RatioGauge;
import org.apache.cassandra.service.FileCacheService;
import org.apache.cassandra.utils.memory.BufferPool;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
public class FileCacheMetrics
public class BufferPoolMetrics
{
private static final MetricNameFactory factory = new DefaultNameFactory("FileCache");
private static final MetricNameFactory factory = new DefaultNameFactory("BufferPool");
/** Total number of hits */
public final Meter hits;
/** Total number of requests */
public final Meter requests;
/** hit rate */
public final Gauge<Double> hitRate;
/** Total size of file cache, in bytes */
/** Total number of misses */
public final Meter misses;
/** Total size of buffer pools, in bytes */
public final Gauge<Long> size;
public FileCacheMetrics()
public BufferPoolMetrics()
{
hits = Metrics.meter(factory.createMetricName("Hits"));
requests = Metrics.meter(factory.createMetricName("Requests"));
hitRate = Metrics.register(factory.createMetricName("HitRate"), new RatioGauge()
{
@Override
public Ratio getRatio()
{
return Ratio.of(hits.getCount(), requests.getCount());
}
});
misses = Metrics.meter(factory.createMetricName("Misses"));
size = Metrics.register(factory.createMetricName("Size"), new Gauge<Long>()
{
public Long getValue()
{
return FileCacheService.instance.sizeInBytes();
return BufferPool.sizeInBytes();
}
});
}

190
src/java/org/apache/cassandra/service/FileCacheService.java
View File

@ -1,190 +0,0 @@
/**
* 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.service;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.cache.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.metrics.FileCacheMetrics;
public class FileCacheService
{
private static final Logger logger = LoggerFactory.getLogger(FileCacheService.class);
private static final long MEMORY_USAGE_THRESHOLD = DatabaseDescriptor.getFileCacheSizeInMB() * 1024L * 1024L;
private static final int AFTER_ACCESS_EXPIRATION = 512; // in millis
public static FileCacheService instance = new FileCacheService();
private static final AtomicLong cacheKeyIdCounter = new AtomicLong();
public static final class CacheKey
{
final long id;
public CacheKey()
{
this.id = cacheKeyIdCounter.incrementAndGet();
}
public boolean equals(Object that)
{
return that instanceof CacheKey && ((CacheKey) that).id == this.id;
}
public int hashCode()
{
return (int) id;
}
}
private static final Callable<CacheBucket> cacheForPathCreator = new Callable<CacheBucket>()
{
@Override
public CacheBucket call()
{
return new CacheBucket();
}
};
private static final AtomicInteger memoryUsage = new AtomicInteger();
private final Cache<CacheKey, CacheBucket> cache;
private final FileCacheMetrics metrics = new FileCacheMetrics();
private static final class CacheBucket
{
final ConcurrentLinkedQueue<RandomAccessReader> queue = new ConcurrentLinkedQueue<>();
volatile boolean discarded = false;
}
protected FileCacheService()
{
RemovalListener<CacheKey, CacheBucket> onRemove = new RemovalListener<CacheKey, CacheBucket>()
{
@Override
public void onRemoval(RemovalNotification<CacheKey, CacheBucket> notification)
{
CacheBucket bucket = notification.getValue();
if (bucket == null)
return;
// set discarded before deallocating the readers, to ensure we don't leak any
bucket.discarded = true;
Queue<RandomAccessReader> q = bucket.queue;
boolean first = true;
for (RandomAccessReader reader = q.poll() ; reader != null ; reader = q.poll())
{
if (logger.isDebugEnabled() && first)
{
logger.debug("Evicting cold readers for {}", reader.getPath());
first = false;
}
memoryUsage.addAndGet(-1 * reader.getTotalBufferSize());
reader.deallocate();
}
}
};
cache = CacheBuilder.newBuilder()
.expireAfterAccess(AFTER_ACCESS_EXPIRATION, TimeUnit.MILLISECONDS)
.concurrencyLevel(DatabaseDescriptor.getConcurrentReaders())
.removalListener(onRemove)
.initialCapacity(16 << 10)
.build();
}
public RandomAccessReader get(CacheKey key)
{
metrics.requests.mark();
CacheBucket bucket = getCacheFor(key);
RandomAccessReader result = bucket.queue.poll();
if (result != null)
{
metrics.hits.mark();
memoryUsage.addAndGet(-result.getTotalBufferSize());
}
return result;
}
private CacheBucket getCacheFor(CacheKey key)
{
try
{
return cache.get(key, cacheForPathCreator);
}
catch (ExecutionException e)
{
throw new AssertionError(e);
}
}
@SuppressWarnings("resource")
public void put(CacheKey cacheKey, RandomAccessReader instance)
{
int memoryUsed = memoryUsage.get();
if (logger.isDebugEnabled())
logger.debug("Estimated memory usage is {} compared to actual usage {}", memoryUsed, sizeInBytes());
CacheBucket bucket = cache.getIfPresent(cacheKey);
if (memoryUsed >= MEMORY_USAGE_THRESHOLD || bucket == null)
{
instance.deallocate();
}
else
{
memoryUsage.addAndGet(instance.getTotalBufferSize());
bucket.queue.add(instance);
if (bucket.discarded)
{
RandomAccessReader reader = bucket.queue.poll();
if (reader != null)
{
memoryUsage.addAndGet(-1 * reader.getTotalBufferSize());
reader.deallocate();
}
}
}
}
public void invalidate(CacheKey cacheKey, String path)
{
if (logger.isDebugEnabled())
logger.debug("Invalidating cache for {}", path);
cache.invalidate(cacheKey);
}
// TODO: this method is unsafe, as it calls getTotalBufferSize() on items that can have been discarded
public long sizeInBytes()
{
long n = 0;
for (CacheBucket bucket : cache.asMap().values())
for (RandomAccessReader reader : bucket.queue)
n += reader.getTotalBufferSize();
return n;
}
}

2
src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java
View File

@ -55,7 +55,7 @@ public class CompressedStreamWriter extends StreamWriter
public void write(DataOutputStreamPlus out) throws IOException
{
long totalSize = totalSize();
try (RandomAccessReader file = sstable.openDataReader(); final ChannelProxy fc = file.getChannel())
try (RandomAccessReader file = sstable.openDataReader(); final ChannelProxy fc = file.getChannel().sharedCopy())
{
long progress = 0L;
// calculate chunks to transfer. we want to send continuous chunks altogether.

153
tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java → src/java/org/apache/cassandra/utils/DynamicList.java
View File

@ -15,16 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.stress.util;
package org.apache.cassandra.utils;
import java.util.Arrays;
import java.util.HashMap;
import java.util.TreeSet;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.cassandra.stress.generate.FasterRandom;
// simple thread-unsafe skiplist that permits indexing/removal by position, insertion at the end
// (though easily extended to insertion at any position, not necessary here)
@ -89,7 +85,6 @@ public class DynamicList<E>
}
}
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final int maxHeight;
private final Node<E> head;
private int size;
@ -115,112 +110,92 @@ public class DynamicList<E>
public Node<E> append(E value, int maxSize)
{
Node<E> newTail = new Node<>(randomLevel(), value);
if (size >= maxSize)
return null;
size++;
lock.writeLock().lock();
try
Node<E> tail = head;
for (int i = maxHeight - 1 ; i >= newTail.height() ; i--)
{
if (size >= maxSize)
return null;
size++;
Node<E> tail = head;
for (int i = maxHeight - 1 ; i >= newTail.height() ; i--)
{
Node<E> next;
while ((next = tail.next(i)) != null)
tail = next;
tail.size[i]++;
}
for (int i = newTail.height() - 1 ; i >= 0 ; i--)
{
Node<E> next;
while ((next = tail.next(i)) != null)
tail = next;
tail.setNext(i, newTail);
newTail.setPrev(i, tail);
}
return newTail;
Node<E> next;
while ((next = tail.next(i)) != null)
tail = next;
tail.size[i]++;
}
finally
for (int i = newTail.height() - 1 ; i >= 0 ; i--)
{
lock.writeLock().unlock();
Node<E> next;
while ((next = tail.next(i)) != null)
tail = next;
tail.setNext(i, newTail);
newTail.setPrev(i, tail);
}
return newTail;
}
// remove the provided node and its associated value from the list
public void remove(Node<E> node)
{
lock.writeLock().lock();
try
assert node.value != null;
node.value = null;
size--;
// go up through each level in the skip list, unlinking this node; this entails
// simply linking each neighbour to each other, and appending the size of the
// current level owned by this node's index to the preceding neighbour (since
// ownership is defined as any node that you must visit through the index,
// removal of ourselves from a level means the preceding index entry is the
// entry point to all of the removed node's descendants)
for (int i = 0 ; i < node.height() ; i++)
{
assert node.value != null;
node.value = null;
size--;
// go up through each level in the skip list, unlinking this node; this entails
// simply linking each neighbour to each other, and appending the size of the
// current level owned by this node's index to the preceding neighbour (since
// ownership is defined as any node that you must visit through the index,
// removal of ourselves from a level means the preceding index entry is the
// entry point to all of the removed node's descendants)
for (int i = 0 ; i < node.height() ; i++)
{
Node<E> prev = node.prev(i);
Node<E> next = node.next(i);
assert prev != null;
prev.setNext(i, next);
if (next != null)
next.setPrev(i, prev);
prev.size[i] += node.size[i] - 1;
}
// then go up the levels, removing 1 from the size at each height above ours
for (int i = node.height() ; i < maxHeight ; i++)
{
// if we're at our height limit, we backtrack at our top level until we
// hit a neighbour with a greater height
while (i == node.height())
node = node.prev(i - 1);
node.size[i]--;
}
Node<E> prev = node.prev(i);
Node<E> next = node.next(i);
assert prev != null;
prev.setNext(i, next);
if (next != null)
next.setPrev(i, prev);
prev.size[i] += node.size[i] - 1;
}
finally
// then go up the levels, removing 1 from the size at each height above ours
for (int i = node.height() ; i < maxHeight ; i++)
{
lock.writeLock().unlock();
// if we're at our height limit, we backtrack at our top level until we
// hit a neighbour with a greater height
while (i == node.height())
node = node.prev(i - 1);
node.size[i]--;
}
}
// retrieve the item at the provided index, or return null if the index is past the end of the list
public E get(int index)
{
lock.readLock().lock();
try
{
if (index >= size)
return null;
if (index >= size)
return null;
index++;
int c = 0;
Node<E> finger = head;
for (int i = maxHeight - 1 ; i >= 0 ; i--)
index++;
int c = 0;
Node<E> finger = head;
for (int i = maxHeight - 1 ; i >= 0 ; i--)
{
while (c + finger.size[i] <= index)
{
while (c + finger.size[i] <= index)
{
c += finger.size[i];
finger = finger.next(i);
}
c += finger.size[i];
finger = finger.next(i);
}
}
assert c == index;
return finger.value;
}
finally
{
lock.readLock().unlock();
}
assert c == index;
return finger.value;
}
public int size()
{
return size;
}
// some quick and dirty tests to confirm the skiplist works as intended

2
tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java → src/java/org/apache/cassandra/utils/FasterRandom.java
View File

@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.stress.generate;
package org.apache.cassandra.utils;
import java.util.Random;

91
src/java/org/apache/cassandra/utils/LockedDynamicList.java Normal file
View File

@ -0,0 +1,91 @@
/*
* 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;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
// simple thread-unsafe skiplist that permits indexing/removal by position, insertion at the end
// (though easily extended to insertion at any position, not necessary here)
// we use it for sampling items by position for visiting writes in the pool of pending writes
public class LockedDynamicList<E> extends DynamicList<E>
{
private final ReadWriteLock lock = new ReentrantReadWriteLock();
public LockedDynamicList(int maxExpectedSize)
{
super(maxExpectedSize);
}
// add the value to the end of the list, and return the associated Node that permits efficient removal
// regardless of its future position in the list from other modifications
public Node<E> append(E value, int maxSize)
{
lock.writeLock().lock();
try
{
return super.append(value, maxSize);
}
finally
{
lock.writeLock().unlock();
}
}
// remove the provided node and its associated value from the list
public void remove(Node<E> node)
{
lock.writeLock().lock();
try
{
super.remove(node);
}
finally
{
lock.writeLock().unlock();
}
}
// retrieve the item at the provided index, or return null if the index is past the end of the list
public E get(int index)
{
lock.readLock().lock();
try
{
return super.get(index);
}
finally
{
lock.readLock().unlock();
}
}
public int size()
{
lock.readLock().lock();
try
{
return super.size();
}
finally
{
lock.readLock().unlock();
}
}
}

7
src/java/org/apache/cassandra/utils/concurrent/Ref.java
View File

@ -282,7 +282,8 @@ public final class Ref<T> implements RefCounted<T>
globallyExtant.remove(this);
try
{
tidy.tidy();
if (tidy != null)
tidy.tidy();
}
catch (Throwable t)
{
@ -299,7 +300,9 @@ public final class Ref<T> implements RefCounted<T>
public String toString()
{
return tidy.getClass() + "@" + System.identityHashCode(tidy) + ":" + tidy.name();
if (tidy != null)
return tidy.getClass() + "@" + System.identityHashCode(tidy) + ":" + tidy.name();
return "@" + System.identityHashCode(this);
}
}

858
src/java/org/apache/cassandra/utils/memory/BufferPool.java Normal file
View File

@ -0,0 +1,858 @@
/*
* 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.ref.PhantomReference;
import java.lang.ref.ReferenceQueue;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.NoSpamLogger;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.metrics.BufferPoolMetrics;
import org.apache.cassandra.utils.concurrent.Ref;
/**
* A pool of ByteBuffers that can be recycled.
*/
public class BufferPool
{
/** The size of a page aligned buffer, 64kbit */
static final int CHUNK_SIZE = 64 << 10;
@VisibleForTesting
public static long MEMORY_USAGE_THRESHOLD = DatabaseDescriptor.getFileCacheSizeInMB() * 1024L * 1024L;
@VisibleForTesting
public static boolean ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = DatabaseDescriptor.getBufferPoolUseHeapIfExhausted();
@VisibleForTesting
public static boolean DISABLED = Boolean.parseBoolean(System.getProperty("cassandra.test.disable_buffer_pool", "false"));
@VisibleForTesting
public static boolean DEBUG = false;
private static final Logger logger = LoggerFactory.getLogger(BufferPool.class);
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 15L, TimeUnit.MINUTES);
private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocateDirect(0);
/** A global pool of chunks (page aligned buffers) */
private static final GlobalPool globalPool = new GlobalPool();
/** A thread local pool of chunks, where chunks come from the global pool */
private static final ThreadLocal<LocalPool> localPool = new ThreadLocal<LocalPool>() {
@Override
protected LocalPool initialValue()
{
return new LocalPool();
}
};
public static ByteBuffer get(int size)
{
if (DISABLED)
return allocate(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED);
else
return takeFromPool(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED);
}
public static ByteBuffer get(int size, BufferType bufferType)
{
boolean direct = bufferType == BufferType.OFF_HEAP;
if (DISABLED | !direct)
return allocate(size, !direct);
else
return takeFromPool(size, !direct);
}
/** Unlike the get methods, this will return null if the pool is exhausted */
public static ByteBuffer tryGet(int size)
{
if (DISABLED)
return allocate(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED);
else
return maybeTakeFromPool(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED);
}
private static ByteBuffer allocate(int size, boolean onHeap)
{
return onHeap
? ByteBuffer.allocate(size)
: ByteBuffer.allocateDirect(size);
}
private static ByteBuffer takeFromPool(int size, boolean allocateOnHeapWhenExhausted)
{
ByteBuffer ret = maybeTakeFromPool(size, allocateOnHeapWhenExhausted);
if (ret != null)
return ret;
if (logger.isTraceEnabled())
logger.trace("Requested buffer size {} has been allocated directly due to lack of capacity", size);
return localPool.get().allocate(size, allocateOnHeapWhenExhausted);
}
private static ByteBuffer maybeTakeFromPool(int size, boolean allocateOnHeapWhenExhausted)
{
if (size < 0)
throw new IllegalArgumentException("Size must be positive (" + size + ")");
if (size == 0)
return EMPTY_BUFFER;
if (size > CHUNK_SIZE)
{
if (logger.isTraceEnabled())
logger.trace("Requested buffer size {} is bigger than {}, allocating directly", size, CHUNK_SIZE);
return localPool.get().allocate(size, allocateOnHeapWhenExhausted);
}
return localPool.get().get(size);
}
public static void put(ByteBuffer buffer)
{
if (!(DISABLED | buffer.hasArray()))
localPool.get().put(buffer);
}
/** This is not thread safe and should only be used for unit testing. */
@VisibleForTesting
static void reset()
{
localPool.get().reset();
globalPool.reset();
}
@VisibleForTesting
static Chunk currentChunk()
{
return localPool.get().chunks[0];
}
@VisibleForTesting
static int numChunks()
{
int ret = 0;
for (Chunk chunk : localPool.get().chunks)
{
if (chunk != null)
ret++;
}
return ret;
}
@VisibleForTesting
static void assertAllRecycled()
{
globalPool.debug.check();
}
public static long sizeInBytes()
{
return globalPool.sizeInBytes();
}
static final class Debug
{
long recycleRound = 1;
final Queue<Chunk> allChunks = new ConcurrentLinkedQueue<>();
void register(Chunk chunk)
{
allChunks.add(chunk);
}
void recycle(Chunk chunk)
{
chunk.lastRecycled = recycleRound;
}
void check()
{
for (Chunk chunk : allChunks)
assert chunk.lastRecycled == recycleRound;
recycleRound++;
}
}
/**
* A queue of page aligned buffers, the chunks, which have been sliced from bigger chunks,
* the macro-chunks, also page aligned. Macro-chunks are allocated as long as we have not exceeded the
* memory maximum threshold, MEMORY_USAGE_THRESHOLD and are never released.
*
* This class is shared by multiple thread local pools and must be thread-safe.
*/
static final class GlobalPool
{
/** The size of a bigger chunk, 1-mbit, must be a multiple of CHUNK_SIZE */
static final int MACRO_CHUNK_SIZE = 1 << 20;
static
{
assert Integer.bitCount(CHUNK_SIZE) == 1; // must be a power of 2
assert Integer.bitCount(MACRO_CHUNK_SIZE) == 1; // must be a power of 2
assert MACRO_CHUNK_SIZE % CHUNK_SIZE == 0; // must be a multiple
if (DISABLED)
logger.info("Global buffer pool is disabled, allocating {}", ALLOCATE_ON_HEAP_WHEN_EXAHUSTED ? "on heap" : "off heap");
else
logger.info("Global buffer pool is enabled, when pool is exahusted (max is {} mb) it will allocate {}",
MEMORY_USAGE_THRESHOLD / (1024L * 1024L),
ALLOCATE_ON_HEAP_WHEN_EXAHUSTED ? "on heap" : "off heap");
}
private final Debug debug = new Debug();
private final Queue<Chunk> macroChunks = new ConcurrentLinkedQueue<>();
// TODO (future): it would be preferable to use a CLStack to improve cache occupancy; it would also be preferable to use "CoreLocal" storage
private final Queue<Chunk> chunks = new ConcurrentLinkedQueue<>();
private final AtomicLong memoryUsage = new AtomicLong();
/** Return a chunk, the caller will take owership of the parent chunk. */
public Chunk get()
{
while (true)
{
Chunk chunk = chunks.poll();
if (chunk != null)
return chunk;
if (!allocateMoreChunks())
// give it one last attempt, in case someone else allocated before us
return chunks.poll();
}
}
/**
* This method might be called by multiple threads and that's fine if we add more
* than one chunk at the same time as long as we don't exceed the MEMORY_USAGE_THRESHOLD.
*/
private boolean allocateMoreChunks()
{
while (true)
{
long cur = memoryUsage.get();
if (cur + MACRO_CHUNK_SIZE > MEMORY_USAGE_THRESHOLD)
{
noSpamLogger.info("Maximum memory usage reached ({} bytes), cannot allocate chunk of {} bytes",
MEMORY_USAGE_THRESHOLD, MACRO_CHUNK_SIZE);
return false;
}
if (memoryUsage.compareAndSet(cur, cur + MACRO_CHUNK_SIZE))
break;
}
// allocate a large chunk
Chunk chunk = new Chunk(allocateDirectAligned(MACRO_CHUNK_SIZE));
chunk.acquire(null);
macroChunks.add(chunk);
for (int i = 0 ; i < MACRO_CHUNK_SIZE ; i += CHUNK_SIZE)
{
Chunk add = new Chunk(chunk.get(CHUNK_SIZE));
chunks.add(add);
if (DEBUG)
debug.register(add);
}
return true;
}
public void recycle(Chunk chunk)
{
chunks.add(chunk);
}
public long sizeInBytes()
{
return memoryUsage.get();
}
/** This is not thread safe and should only be used for unit testing. */
@VisibleForTesting
void reset()
{
while (!chunks.isEmpty())
chunks.poll().reset();
while (!macroChunks.isEmpty())
macroChunks.poll().reset();
memoryUsage.set(0);
}
}
/**
* A thread local class that grabs chunks from the global pool for this thread allocations.
* Only one thread can do the allocations but multiple threads can release the allocations.
*/
static final class LocalPool
{
private final static BufferPoolMetrics metrics = new BufferPoolMetrics();
// a microqueue of Chunks:
// * if any are null, they are at the end;
// * new Chunks are added to the last null index
// * if no null indexes available, the smallest is swapped with the last index, and this replaced
// * this results in a queue that will typically be visited in ascending order of available space, so that
// small allocations preferentially slice from the Chunks with the smallest space available to furnish them
// WARNING: if we ever change the size of this, we must update removeFromLocalQueue, and addChunk
private final Chunk[] chunks = new Chunk[3];
private byte chunkCount = 0;
public LocalPool()
{
localPoolReferences.add(new LocalPoolRef(this, localPoolRefQueue));
}
private Chunk addChunkFromGlobalPool()
{
Chunk chunk = globalPool.get();
if (chunk == null)
return null;
addChunk(chunk);
return chunk;
}
private void addChunk(Chunk chunk)
{
chunk.acquire(this);
if (chunkCount < 3)
{
chunks[chunkCount++] = chunk;
return;
}
int smallestChunkIdx = 0;
if (chunks[1].free() < chunks[0].free())
smallestChunkIdx = 1;
if (chunks[2].free() < chunks[smallestChunkIdx].free())
smallestChunkIdx = 2;
chunks[smallestChunkIdx].release();
if (smallestChunkIdx != 2)
chunks[smallestChunkIdx] = chunks[2];
chunks[2] = chunk;
}
public ByteBuffer get(int size)
{
for (Chunk chunk : chunks)
{ // first see if our own chunks can serve this buffer
if (chunk == null)
break;
ByteBuffer buffer = chunk.get(size);
if (buffer != null)
return buffer;
}
// else ask the global pool
Chunk chunk = addChunkFromGlobalPool();
if (chunk != null)
return chunk.get(size);
return null;
}
private ByteBuffer allocate(int size, boolean onHeap)
{
metrics.misses.mark();
return BufferPool.allocate(size, onHeap);
}
public void put(ByteBuffer buffer)
{
Chunk chunk = Chunk.getParentChunk(buffer);
if (chunk == null)
{
FileUtils.clean(buffer);
return;
}
LocalPool owner = chunk.owner;
// ask the free method to take exclusive ownership of the act of recycling
// if we are either: already not owned by anyone, or owned by ourselves
long free = chunk.free(buffer, owner == null | owner == this);
if (free == 0L)
{
// 0L => we own recycling responsibility, so must recycle;
chunk.recycle();
// if we are also the owner, we must remove the Chunk from our local queue
if (owner == this)
removeFromLocalQueue(chunk);
}
else if (((free == -1L) & owner != this) && chunk.owner == null)
{
// although we try to take recycle ownership cheaply, it is not always possible to do so if the owner is racing to unset.
// we must also check after completely freeing if the owner has since been unset, and try to recycle
chunk.tryRecycle();
}
}
private void removeFromLocalQueue(Chunk chunk)
{
// since we only have three elements in the queue, it is clearer, easier and faster to just hard code the options
if (chunks[0] == chunk)
{ // remove first by shifting back second two
chunks[0] = chunks[1];
chunks[1] = chunks[2];
}
else if (chunks[1] == chunk)
{ // remove second by shifting back last
chunks[1] = chunks[2];
}
else assert chunks[2] == chunk;
// whatever we do, the last element myst be null
chunks[2] = null;
chunkCount--;
}
@VisibleForTesting
void reset()
{
chunkCount = 0;
for (int i = 0; i < chunks.length; i++)
{
if (chunks[i] != null)
{
chunks[i].owner = null;
chunks[i].freeSlots = 0L;
chunks[i].recycle();
chunks[i] = null;
}
}
}
}
private static final class LocalPoolRef extends PhantomReference<LocalPool>
{
private final Chunk[] chunks;
public LocalPoolRef(LocalPool localPool, ReferenceQueue<? super LocalPool> q)
{
super(localPool, q);
chunks = localPool.chunks;
}
public void release()
{
for (int i = 0 ; i < chunks.length ; i++)
{
if (chunks[i] != null)
{
chunks[i].release();
chunks[i] = null;
}
}
}
}
private static final ConcurrentLinkedQueue<LocalPoolRef> localPoolReferences = new ConcurrentLinkedQueue<>();
private static final ReferenceQueue<Object> localPoolRefQueue = new ReferenceQueue<>();
private static final ExecutorService EXEC = Executors.newFixedThreadPool(1, new NamedThreadFactory("LocalPool-Cleaner"));
static
{
EXEC.execute(new Runnable()
{
public void run()
{
try
{
while (true)
{
Object obj = localPoolRefQueue.remove();
if (obj instanceof LocalPoolRef)
{
((LocalPoolRef) obj).release();
localPoolReferences.remove(obj);
}
}
}
catch (InterruptedException e)
{
}
finally
{
EXEC.execute(this);
}
}
});
}
private static ByteBuffer allocateDirectAligned(int capacity)
{
int align = MemoryUtil.pageSize();
if (Integer.bitCount(align) != 1)
throw new IllegalArgumentException("Alignment must be a power of 2");
ByteBuffer buffer = ByteBuffer.allocateDirect(capacity + align);
long address = MemoryUtil.getAddress(buffer);
long offset = address & (align -1); // (address % align)
if (offset == 0)
{ // already aligned
buffer.limit(capacity);
}
else
{ // shift by offset
int pos = (int)(align - offset);
buffer.position(pos);
buffer.limit(pos + capacity);
}
return buffer.slice();
}
/**
* A memory chunk: it takes a buffer (the slab) and slices it
* into smaller buffers when requested.
*
* It divides the slab into 64 units and keeps a long mask, freeSlots,
* indicating if a unit is in use or not. Each bit in freeSlots corresponds
* to a unit, if the bit is set then the unit is free (available for allocation)
* whilst if it is not set then the unit is in use.
*
* When we receive a request of a given size we round up the size to the nearest
* multiple of allocation units required. Then we search for n consecutive free units,
* where n is the number of units required. We also align to page boundaries.
*
* When we reiceve a release request we work out the position by comparing the buffer
* address to our base address and we simply release the units.
*/
final static class Chunk
{
private final ByteBuffer slab;
private final long baseAddress;
private final int shift;
private volatile long freeSlots;
private static final AtomicLongFieldUpdater<Chunk> freeSlotsUpdater = AtomicLongFieldUpdater.newUpdater(Chunk.class, "freeSlots");
// the pool that is _currently allocating_ from this Chunk
// if this is set, it means the chunk may not be recycled because we may still allocate from it;
// if it has been unset the local pool has finished with it, and it may be recycled
private volatile LocalPool owner;
private long lastRecycled;
private final Chunk original;
Chunk(Chunk recycle)
{
assert recycle.freeSlots == 0L;
this.slab = recycle.slab;
this.baseAddress = recycle.baseAddress;
this.shift = recycle.shift;
this.freeSlots = -1L;
this.original = recycle.original;
if (DEBUG)
globalPool.debug.recycle(original);
}
Chunk(ByteBuffer slab)
{
assert !slab.hasArray();
this.slab = slab;
this.baseAddress = MemoryUtil.getAddress(slab);
// The number of bits by which we need to shift to obtain a unit
// "31 &" is because numberOfTrailingZeros returns 32 when the capacity is zero
this.shift = 31 & (Integer.numberOfTrailingZeros(slab.capacity() / 64));
// -1 means all free whilst 0 means all in use
this.freeSlots = slab.capacity() == 0 ? 0L : -1L;
this.original = DEBUG ? this : null;
}
/**
* Acquire the chunk for future allocations: set the owner and prep
* the free slots mask.
*/
void acquire(LocalPool owner)
{
assert this.owner == null;
this.owner = owner;
}
/**
* Set the owner to null and return the chunk to the global pool if the chunk is fully free.
* This method must be called by the LocalPool when it is certain that
* the local pool shall never try to allocate any more buffers from this chunk.
*/
void release()
{
this.owner = null;
tryRecycle();
}
void tryRecycle()
{
assert owner == null;
if (isFree() && freeSlotsUpdater.compareAndSet(this, -1L, 0L))
recycle();
}
void recycle()
{
assert freeSlots == 0L;
globalPool.recycle(new Chunk(this));
}
/**
* We stash the chunk in the attachment of a buffer
* that was returned by get(), this method simply
* retrives the chunk that sliced a buffer, if any.
*/
static Chunk getParentChunk(ByteBuffer buffer)
{
Object attachment = MemoryUtil.getAttachment(buffer);
if (attachment instanceof Chunk)
return (Chunk) attachment;
if (attachment instanceof Ref)
return ((Ref<Chunk>) attachment).get();
return null;
}
ByteBuffer setAttachment(ByteBuffer buffer)
{
if (Ref.DEBUG_ENABLED)
MemoryUtil.setAttachment(buffer, new Ref<>(this, null));
else
MemoryUtil.setAttachment(buffer, this);
return buffer;
}
boolean releaseAttachment(ByteBuffer buffer)
{
Object attachment = MemoryUtil.getAttachment(buffer);
if (attachment == null)
return false;
if (attachment instanceof Ref)
((Ref<Chunk>) attachment).release();
return true;
}
@VisibleForTesting
void reset()
{
Chunk parent = getParentChunk(slab);
if (parent != null)
parent.free(slab, false);
else
FileUtils.clean(slab);
}
@VisibleForTesting
long setFreeSlots(long val)
{
long ret = freeSlots;
freeSlots = val;
return ret;
}
int capacity()
{
return 64 << shift;
}
final int unit()
{
return 1 << shift;
}
final boolean isFree()
{
return freeSlots == -1L;
}
/** The total free size */
int free()
{
return Long.bitCount(freeSlots) * unit();
}
/**
* Return the next available slice of this size. If
* we have exceeded the capacity we return null.
*/
ByteBuffer get(int size)
{
// how many multiples of our units is the size?
// we add (unit - 1), so that when we divide by unit (>>> shift), we effectively round up
int slotCount = (size - 1 + unit()) >>> shift;
// if we require more than 64 slots, we cannot possibly accommodate the allocation
if (slotCount > 64)
return null;
// convert the slotCount into the bits needed in the bitmap, but at the bottom of the register
long slotBits = -1L >>> (64 - slotCount);
// in order that we always allocate page aligned results, we require that any allocation is "somewhat" aligned
// i.e. any single unit allocation can go anywhere; any 2 unit allocation must begin in one of the first 3 slots
// of a page; a 3 unit must go in the first two slots; and any four unit allocation must be fully page-aligned
// to achieve this, we construct a searchMask that constrains the bits we find to those we permit starting
// a match from. as we find bits, we remove them from the mask to continue our search.
// this has an odd property when it comes to concurrent alloc/free, as we can safely skip backwards if
// a new slot is freed up, but we always make forward progress (i.e. never check the same bits twice),
// so running time is bounded
long searchMask = 0x1111111111111111L;
searchMask *= 15L >>> ((slotCount - 1) & 3);
// i.e. switch (slotCount & 3)
// case 1: searchMask = 0xFFFFFFFFFFFFFFFFL
// case 2: searchMask = 0x7777777777777777L
// case 3: searchMask = 0x3333333333333333L
// case 0: searchMask = 0x1111111111111111L
// truncate the mask, removing bits that have too few slots proceeding them
searchMask &= -1L >>> (slotCount - 1);
// this loop is very unroll friendly, and would achieve high ILP, but not clear if the compiler will exploit this.
// right now, not worth manually exploiting, but worth noting for future
while (true)
{
long cur = freeSlots;
// find the index of the lowest set bit that also occurs in our mask (i.e. is permitted alignment, and not yet searched)
// we take the index, rather than finding the lowest bit, since we must obtain it anyway, and shifting is more efficient
// than multiplication
int index = Long.numberOfTrailingZeros(cur & searchMask);
// if no bit was actually found, we cannot serve this request, so return null.
// due to truncating the searchMask this immediately terminates any search when we run out of indexes
// that could accommodate the allocation, i.e. is equivalent to checking (64 - index) < slotCount
if (index == 64)
return null;
// remove this bit from our searchMask, so we don't return here next round
searchMask ^= 1L << index;
// if our bits occur starting at the index, remove ourselves from the bitmask and return
long candidate = slotBits << index;
if ((candidate & cur) == candidate)
{
// here we are sure we will manage to CAS successfully without changing candidate because
// there is only one thread allocating at the moment, the concurrency is with the release
// operations only
while (true)
{
// clear the candidate bits (freeSlots &= ~candidate)
if (freeSlotsUpdater.compareAndSet(this, cur, cur & ~candidate))
break;
cur = freeSlots;
// make sure no other thread has cleared the candidate bits
assert ((candidate & cur) == candidate);
}
return get(index << shift, size);
}
}
}
private ByteBuffer get(int offset, int size)
{
slab.limit(offset + size);
slab.position(offset);
return setAttachment(slab.slice());
}
/**
* Round the size to the next unit multiple.
*/
int roundUp(int v)
{
return BufferPool.roundUp(v, unit());
}
/**
* Release a buffer. Return:
* 0L if the buffer must be recycled after the call;
* -1L if it is free (and so we should tryRecycle if owner is now null)
* some other value otherwise
**/
long free(ByteBuffer buffer, boolean tryRelease)
{
if (!releaseAttachment(buffer))
return 1L;
long address = MemoryUtil.getAddress(buffer);
assert (address >= baseAddress) & (address <= baseAddress + capacity());
int position = (int)(address - baseAddress);
int size = roundUp(buffer.capacity());
position >>= shift;
int slotCount = size >> shift;
long slotBits = (1L << slotCount) - 1;
long shiftedSlotBits = (slotBits << position);
if (slotCount == 64)
{
assert size == capacity();
assert position == 0;
shiftedSlotBits = -1L;
}
long next;
while (true)
{
long cur = freeSlots;
next = cur | shiftedSlotBits;
assert next == (cur ^ shiftedSlotBits); // ensure no double free
if (tryRelease & (next == -1L))
next = 0L;
if (freeSlotsUpdater.compareAndSet(this, cur, next))
return next;
}
}
@Override
public String toString()
{
return String.format("[slab %s, slots bitmap %s, capacity %d, free %d]", slab, Long.toBinaryString(freeSlots), capacity(), free());
}
}
@VisibleForTesting
public static int roundUpNormal(int size)
{
return roundUp(size, CHUNK_SIZE / 64);
}
private static int roundUp(int size, int unit)
{
int mask = unit - 1;
return (size + mask) & ~mask;
}
}

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

@ -36,6 +36,7 @@ public abstract class MemoryUtil
private static final long DIRECT_BYTE_BUFFER_CAPACITY_OFFSET;
private static final long DIRECT_BYTE_BUFFER_LIMIT_OFFSET;
private static final long DIRECT_BYTE_BUFFER_POSITION_OFFSET;
private static final long DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET;
private static final Class<?> BYTE_BUFFER_CLASS;
private static final long BYTE_BUFFER_OFFSET_OFFSET;
private static final long BYTE_BUFFER_HB_OFFSET;
@ -62,6 +63,7 @@ public abstract class MemoryUtil
DIRECT_BYTE_BUFFER_CAPACITY_OFFSET = unsafe.objectFieldOffset(Buffer.class.getDeclaredField("capacity"));
DIRECT_BYTE_BUFFER_LIMIT_OFFSET = unsafe.objectFieldOffset(Buffer.class.getDeclaredField("limit"));
DIRECT_BYTE_BUFFER_POSITION_OFFSET = unsafe.objectFieldOffset(Buffer.class.getDeclaredField("position"));
DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET = unsafe.objectFieldOffset(clazz.getDeclaredField("att"));
DIRECT_BYTE_BUFFER_CLASS = clazz;
clazz = ByteBuffer.allocate(0).getClass();
@ -77,6 +79,17 @@ public abstract class MemoryUtil
}
}
public static int pageSize()
{
return unsafe.pageSize();
}
public static long getAddress(ByteBuffer buffer)
{
assert buffer.getClass() == DIRECT_BYTE_BUFFER_CLASS;
return unsafe.getLong(buffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET);
}
public static long allocate(long size)
{
return Native.malloc(size);
@ -177,9 +190,21 @@ public abstract class MemoryUtil
unsafe.putInt(instance, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, length);
}
public static Object getAttachment(ByteBuffer instance)
{
assert instance.getClass() == DIRECT_BYTE_BUFFER_CLASS;
return unsafe.getObject(instance, DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET);
}
public static void setAttachment(ByteBuffer instance, Object next)
{
assert instance.getClass() == DIRECT_BYTE_BUFFER_CLASS;
unsafe.putObject(instance, DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET, next);
}
public static ByteBuffer duplicateDirectByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer)
{
assert(source.isDirect());
assert source.getClass() == DIRECT_BYTE_BUFFER_CLASS;
unsafe.putLong(hollowBuffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, unsafe.getLong(source, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET));
unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_POSITION_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_POSITION_OFFSET));
unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_LIMIT_OFFSET));
@ -187,17 +212,6 @@ public abstract class MemoryUtil
return hollowBuffer;
}
public static ByteBuffer duplicateByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer)
{
assert(!source.isDirect());
unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_POSITION_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_POSITION_OFFSET));
unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_LIMIT_OFFSET));
unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET));
unsafe.putInt(hollowBuffer, BYTE_BUFFER_OFFSET_OFFSET, unsafe.getInt(source, BYTE_BUFFER_OFFSET_OFFSET));
unsafe.putObject(hollowBuffer, BYTE_BUFFER_HB_OFFSET, unsafe.getObject(source, BYTE_BUFFER_HB_OFFSET));
return hollowBuffer;
}
public static long getLongByByte(long address)
{
if (BIG_ENDIAN)

454
test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java Normal file
View File

@ -0,0 +1,454 @@
/*
* 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.nio.ByteBuffer;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.utils.DynamicList;
import static org.junit.Assert.*;
public class LongBufferPoolTest
{
private static final Logger logger = LoggerFactory.getLogger(LongBufferPoolTest.class);
@Test
public void testAllocate() throws InterruptedException, ExecutionException
{
testAllocate(Runtime.getRuntime().availableProcessors() * 2, TimeUnit.MINUTES.toNanos(2L), 16 << 20);
}
private static final class BufferCheck
{
final ByteBuffer buffer;
final long val;
DynamicList.Node<BufferCheck> listnode;
private BufferCheck(ByteBuffer buffer, long val)
{
this.buffer = buffer;
this.val = val;
}
void validate()
{
ByteBuffer read = buffer.duplicate();
while (read.remaining() > 8)
assert read.getLong() == val;
}
void init()
{
ByteBuffer write = buffer.duplicate();
while (write.remaining() > 8)
write.putLong(val);
}
}
public void testAllocate(int threadCount, long duration, int poolSize) throws InterruptedException, ExecutionException
{
final int avgBufferSize = 16 << 10;
final int stdevBufferSize = 10 << 10; // picked to ensure exceeding buffer size is rare, but occurs
final DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
System.out.println(String.format("%s - testing %d threads for %dm",
dateFormat.format(new Date()),
threadCount,
TimeUnit.NANOSECONDS.toMinutes(duration)));
final long until = System.nanoTime() + duration;
final CountDownLatch latch = new CountDownLatch(threadCount);
final SPSCQueue<BufferCheck>[] sharedRecycle = new SPSCQueue[threadCount];
final AtomicBoolean[] makingProgress = new AtomicBoolean[threadCount];
for (int i = 0 ; i < sharedRecycle.length ; i++)
{
sharedRecycle[i] = new SPSCQueue<>();
makingProgress[i] = new AtomicBoolean(true);
}
ExecutorService executorService = Executors.newFixedThreadPool(threadCount + 2);
List<Future<Boolean>> ret = new ArrayList<>(threadCount);
long prevPoolSize = BufferPool.MEMORY_USAGE_THRESHOLD;
BufferPool.MEMORY_USAGE_THRESHOLD = poolSize;
BufferPool.DEBUG = true;
// sum(1..n) = n/2 * (n + 1); we set zero to CHUNK_SIZE, so have n=threadCount-1
int targetSizeQuanta = ((threadCount) * (threadCount - 1)) / 2;
// fix targetSizeQuanta at 1/64th our poolSize, so that we only consciously exceed our pool size limit
targetSizeQuanta = (targetSizeQuanta * poolSize) / 64;
{
// setup some high churn allocate/deallocate, without any checking
final SPSCQueue<ByteBuffer> burn = new SPSCQueue<>();
final CountDownLatch doneAdd = new CountDownLatch(1);
executorService.submit(new TestUntil(until)
{
int count = 0;
void testOne() throws Exception
{
if (count * BufferPool.CHUNK_SIZE >= poolSize / 10)
{
if (burn.exhausted)
count = 0;
else
Thread.yield();
return;
}
ByteBuffer buffer = BufferPool.tryGet(BufferPool.CHUNK_SIZE);
if (buffer == null)
{
Thread.yield();
return;
}
BufferPool.put(buffer);
burn.add(buffer);
count++;
}
void cleanup()
{
doneAdd.countDown();
}
});
executorService.submit(new TestUntil(until)
{
void testOne() throws Exception
{
ByteBuffer buffer = burn.poll();
if (buffer == null)
{
Thread.yield();
return;
}
BufferPool.put(buffer);
}
void cleanup()
{
Uninterruptibles.awaitUninterruptibly(doneAdd);
}
});
}
for (int t = 0; t < threadCount; t++)
{
final int threadIdx = t;
final int targetSize = t == 0 ? BufferPool.CHUNK_SIZE : targetSizeQuanta * t;
ret.add(executorService.submit(new TestUntil(until)
{
final SPSCQueue<BufferCheck> shareFrom = sharedRecycle[threadIdx];
final DynamicList<BufferCheck> checks = new DynamicList<>((int) Math.max(1, targetSize / (1 << 10)));
final SPSCQueue<BufferCheck> shareTo = sharedRecycle[(threadIdx + 1) % threadCount];
final ThreadLocalRandom rand = ThreadLocalRandom.current();
int totalSize = 0;
int freeingSize = 0;
int size = 0;
void checkpoint()
{
if (!makingProgress[threadIdx].get())
makingProgress[threadIdx].set(true);
}
void testOne() throws Exception
{
long currentTargetSize = rand.nextInt(poolSize / 1024) == 0 ? 0 : targetSize;
int spinCount = 0;
while (totalSize > currentTargetSize - freeingSize)
{
// free buffers until we're below our target size
if (checks.size() == 0)
{
// if we're out of buffers to free, we're waiting on our neighbour to free them;
// first check if the consuming neighbour has caught up, and if so mark that free
if (shareTo.exhausted)
{
totalSize -= freeingSize;
freeingSize = 0;
}
else if (!recycleFromNeighbour())
{
if (++spinCount > 1000 && System.nanoTime() > until)
return;
// otherwise, free one of our other neighbour's buffers if can; and otherwise yield
Thread.yield();
}
continue;
}
// pick a random buffer, with preference going to earlier ones
BufferCheck check = sample();
checks.remove(check.listnode);
check.validate();
size = BufferPool.roundUpNormal(check.buffer.capacity());
if (size > BufferPool.CHUNK_SIZE)
size = 0;
// either share to free, or free immediately
if (rand.nextBoolean())
{
shareTo.add(check);
freeingSize += size;
// interleave this with potentially messing with the other neighbour's stuff
recycleFromNeighbour();
}
else
{
check.validate();
BufferPool.put(check.buffer);
totalSize -= size;
}
}
// allocate a new buffer
size = (int) Math.max(1, avgBufferSize + (stdevBufferSize * rand.nextGaussian()));
if (size <= BufferPool.CHUNK_SIZE)
{
totalSize += BufferPool.roundUpNormal(size);
allocate(size);
}
else if (rand.nextBoolean())
{
allocate(size);
}
else
{
// perform a burst allocation to exhaust all available memory
while (totalSize < poolSize)
{
size = (int) Math.max(1, avgBufferSize + (stdevBufferSize * rand.nextGaussian()));
if (size <= BufferPool.CHUNK_SIZE)
{
allocate(size);
totalSize += BufferPool.roundUpNormal(size);
}
}
}
// validate a random buffer we have stashed
checks.get(rand.nextInt(checks.size())).validate();
// free all of our neighbour's remaining shared buffers
while (recycleFromNeighbour());
}
void cleanup()
{
while (checks.size() > 0)
{
BufferCheck check = checks.get(0);
BufferPool.put(check.buffer);
checks.remove(check.listnode);
}
latch.countDown();
}
boolean recycleFromNeighbour()
{
BufferCheck check = shareFrom.poll();
if (check == null)
return false;
check.validate();
BufferPool.put(check.buffer);
return true;
}
BufferCheck allocate(int size)
{
ByteBuffer buffer = BufferPool.get(size);
assertNotNull(buffer);
BufferCheck check = new BufferCheck(buffer, rand.nextLong());
assertEquals(size, buffer.capacity());
assertEquals(0, buffer.position());
check.init();
check.listnode = checks.append(check);
return check;
}
BufferCheck sample()
{
// sample with preference to first elements:
// element at index n will be selected with likelihood (size - n) / sum1ToN(size)
int size = checks.size();
// pick a random number between 1 and sum1toN(size)
int sampleRange = sum1toN(size);
int sampleIndex = rand.nextInt(sampleRange);
// then binary search for the N, such that [sum1ToN(N), sum1ToN(N+1)) contains this random number
int moveBy = Math.max(size / 4, 1);
int index = size / 2;
while (true)
{
int baseSampleIndex = sum1toN(index);
int endOfSampleIndex = sum1toN(index + 1);
if (sampleIndex >= baseSampleIndex)
{
if (sampleIndex < endOfSampleIndex)
break;
index += moveBy;
}
else index -= moveBy;
moveBy = Math.max(moveBy / 2, 1);
}
// this gives us the inverse of our desired value, so just subtract it from the last index
index = size - (index + 1);
return checks.get(index);
}
private int sum1toN(int n)
{
return (n * (n + 1)) / 2;
}
}));
}
boolean first = true;
while (!latch.await(10L, TimeUnit.SECONDS))
{
if (!first)
BufferPool.assertAllRecycled();
first = false;
for (AtomicBoolean progress : makingProgress)
{
assert progress.get();
progress.set(false);
}
}
for (SPSCQueue<BufferCheck> queue : sharedRecycle)
{
BufferCheck check;
while ( null != (check = queue.poll()) )
{
check.validate();
BufferPool.put(check.buffer);
}
}
assertEquals(0, executorService.shutdownNow().size());
BufferPool.MEMORY_USAGE_THRESHOLD = prevPoolSize;
for (Future<Boolean> r : ret)
assertTrue(r.get());
System.out.println(String.format("%s - finished.",
dateFormat.format(new Date())));
}
static abstract class TestUntil implements Callable<Boolean>
{
final long until;
protected TestUntil(long until)
{
this.until = until;
}
abstract void testOne() throws Exception;
void checkpoint() {}
void cleanup() {}
public Boolean call() throws Exception
{
try
{
while (System.nanoTime() < until)
{
checkpoint();
for (int i = 0 ; i < 100 ; i++)
testOne();
}
}
catch (Exception ex)
{
logger.error("Got exception {}, current chunk {}",
ex.getMessage(),
BufferPool.currentChunk());
ex.printStackTrace();
return false;
}
finally
{
cleanup();
}
return true;
}
}
public static void main(String[] args) throws InterruptedException, ExecutionException
{
new LongBufferPoolTest().testAllocate(Runtime.getRuntime().availableProcessors(), TimeUnit.HOURS.toNanos(2L), 16 << 20);
}
/**
* A single producer, single consumer queue.
*/
private static final class SPSCQueue<V>
{
static final class Node<V>
{
volatile Node<V> next;
final V value;
Node(V value)
{
this.value = value;
}
}
private volatile boolean exhausted = true;
Node<V> head = new Node<>(null);
Node<V> tail = head;
void add(V value)
{
exhausted = false;
tail = tail.next = new Node<>(value);
}
V poll()
{
Node<V> next = head.next;
if (next == null)
{
// this is racey, but good enough for our purposes
exhausted = true;
return null;
}
head = next;
return next.value;
}
}
}

1
test/unit/org/apache/cassandra/io/compress/CompressorTest.java
View File

@ -33,6 +33,7 @@ import org.junit.Test;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.BufferPool;
public class CompressorTest
{

120
test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java
View File

@ -18,8 +18,6 @@
*
*/
package org.apache.cassandra.io.util;
import org.apache.cassandra.service.FileCacheService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.SyncUtil;
@ -30,10 +28,6 @@ import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.util.Arrays;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.cassandra.Util.expectEOF;
import static org.apache.cassandra.Util.expectException;
@ -48,6 +42,7 @@ public class BufferedRandomAccessFileTest
public void testReadAndWrite() throws Exception
{
SequentialWriter w = createTempFile("braf");
ChannelProxy channel = new ChannelProxy(w.getPath());
// writting string of data to the file
byte[] data = "Hello".getBytes();
@ -58,7 +53,7 @@ public class BufferedRandomAccessFileTest
w.sync();
// reading small amount of data from file, this is handled by initial buffer
RandomAccessReader r = RandomAccessReader.open(w);
RandomAccessReader r = RandomAccessReader.open(channel);
byte[] buffer = new byte[data.length];
assertEquals(data.length, r.read(buffer));
@ -81,7 +76,7 @@ public class BufferedRandomAccessFileTest
w.sync();
r = RandomAccessReader.open(w); // re-opening file in read-only mode
r = RandomAccessReader.open(channel); // re-opening file in read-only mode
// reading written buffer
r.seek(initialPosition); // back to initial (before write) position
@ -130,6 +125,7 @@ public class BufferedRandomAccessFileTest
w.finish();
r.close();
channel.close();
}
@Test
@ -142,7 +138,8 @@ public class BufferedRandomAccessFileTest
byte[] in = generateByteArray(RandomAccessReader.DEFAULT_BUFFER_SIZE);
w.write(in);
RandomAccessReader r = RandomAccessReader.open(w);
ChannelProxy channel = new ChannelProxy(w.getPath());
RandomAccessReader r = RandomAccessReader.open(channel);
// Read it into a same size array.
byte[] out = new byte[RandomAccessReader.DEFAULT_BUFFER_SIZE];
@ -154,6 +151,7 @@ public class BufferedRandomAccessFileTest
r.close();
w.finish();
channel.close();
}
@Test
@ -181,9 +179,11 @@ public class BufferedRandomAccessFileTest
w.finish();
// will use cachedlength
RandomAccessReader r = RandomAccessReader.open(tmpFile);
assertEquals(lessThenBuffer.length + biggerThenBuffer.length, r.length());
r.close();
try (ChannelProxy channel = new ChannelProxy(tmpFile);
RandomAccessReader r = RandomAccessReader.open(channel))
{
assertEquals(lessThenBuffer.length + biggerThenBuffer.length, r.length());
}
}
@Test
@ -201,7 +201,8 @@ public class BufferedRandomAccessFileTest
w.write(data);
w.sync();
final RandomAccessReader r = RandomAccessReader.open(w);
final ChannelProxy channel = new ChannelProxy(w.getPath());
final RandomAccessReader r = RandomAccessReader.open(channel);
ByteBuffer content = r.readBytes((int) r.length());
@ -225,6 +226,7 @@ public class BufferedRandomAccessFileTest
w.finish();
r.close();
channel.close();
}
@Test
@ -235,7 +237,8 @@ public class BufferedRandomAccessFileTest
w.write(data);
w.finish();
final RandomAccessReader file = RandomAccessReader.open(w);
final ChannelProxy channel = new ChannelProxy(w.getPath());
final RandomAccessReader file = RandomAccessReader.open(channel);
file.seek(0);
assertEquals(file.getFilePointer(), 0);
@ -265,6 +268,7 @@ public class BufferedRandomAccessFileTest
}, IllegalArgumentException.class); // throws IllegalArgumentException
file.close();
channel.close();
}
@Test
@ -274,7 +278,8 @@ public class BufferedRandomAccessFileTest
w.write(generateByteArray(RandomAccessReader.DEFAULT_BUFFER_SIZE * 2));
w.finish();
RandomAccessReader file = RandomAccessReader.open(w);
ChannelProxy channel = new ChannelProxy(w.getPath());
RandomAccessReader file = RandomAccessReader.open(channel);
file.seek(0); // back to the beginning of the file
assertEquals(file.skipBytes(10), 10);
@ -294,6 +299,7 @@ public class BufferedRandomAccessFileTest
assertEquals(file.bytesRemaining(), file.length());
file.close();
channel.close();
}
@Test
@ -308,7 +314,8 @@ public class BufferedRandomAccessFileTest
w.sync();
RandomAccessReader r = RandomAccessReader.open(w);
ChannelProxy channel = new ChannelProxy(w.getPath());
RandomAccessReader r = RandomAccessReader.open(channel);
// position should change after skip bytes
r.seek(0);
@ -322,6 +329,7 @@ public class BufferedRandomAccessFileTest
w.finish();
r.close();
channel.close();
}
@Test
@ -344,7 +352,7 @@ public class BufferedRandomAccessFileTest
{
File file1 = writeTemporaryFile(new byte[16]);
try (final ChannelProxy channel = new ChannelProxy(file1);
final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, null))
final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, -1L))
{
expectEOF(new Callable<Object>()
{
@ -362,7 +370,7 @@ public class BufferedRandomAccessFileTest
{
File file1 = writeTemporaryFile(new byte[16]);
try (final ChannelProxy channel = new ChannelProxy(file1);
final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, null))
final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, -1L))
{
expectEOF(new Callable<Object>()
{
@ -397,7 +405,8 @@ public class BufferedRandomAccessFileTest
w.sync();
RandomAccessReader r = RandomAccessReader.open(w);
ChannelProxy channel = new ChannelProxy(w.getPath());
RandomAccessReader r = RandomAccessReader.open(channel);
assertEquals(r.bytesRemaining(), toWrite);
@ -413,6 +422,7 @@ public class BufferedRandomAccessFileTest
w.finish();
r.close();
channel.close();
}
@Test
@ -483,7 +493,8 @@ public class BufferedRandomAccessFileTest
w.finish();
RandomAccessReader file = RandomAccessReader.open(w);
ChannelProxy channel = new ChannelProxy(w.getPath());
RandomAccessReader file = RandomAccessReader.open(channel);
file.seek(10);
FileMark mark = file.mark();
@ -508,6 +519,7 @@ public class BufferedRandomAccessFileTest
assertEquals(file.bytesPastMark(), 0);
file.close();
channel.close();
}
@Test (expected = AssertionError.class)
@ -518,7 +530,8 @@ public class BufferedRandomAccessFileTest
w.write(new byte[30]);
w.flush();
try (RandomAccessReader r = RandomAccessReader.open(w))
try (ChannelProxy channel = new ChannelProxy(w.getPath());
RandomAccessReader r = RandomAccessReader.open(channel))
{
r.seek(10);
r.mark();
@ -529,71 +542,6 @@ public class BufferedRandomAccessFileTest
}
}
@Test
public void testFileCacheService() throws IOException, InterruptedException
{
//see https://issues.apache.org/jira/browse/CASSANDRA-7756
final FileCacheService.CacheKey cacheKey = new FileCacheService.CacheKey();
final int THREAD_COUNT = 40;
ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
SequentialWriter w1 = createTempFile("fscache1");
SequentialWriter w2 = createTempFile("fscache2");
w1.write(new byte[30]);
w1.finish();
w2.write(new byte[30]);
w2.finish();
for (int i = 0; i < 20; i++)
{
RandomAccessReader r1 = RandomAccessReader.open(w1);
RandomAccessReader r2 = RandomAccessReader.open(w2);
FileCacheService.instance.put(cacheKey, r1);
FileCacheService.instance.put(cacheKey, r2);
final CountDownLatch finished = new CountDownLatch(THREAD_COUNT);
final AtomicBoolean hadError = new AtomicBoolean(false);
for (int k = 0; k < THREAD_COUNT; k++)
{
executorService.execute( new Runnable()
{
@Override
public void run()
{
try
{
long size = FileCacheService.instance.sizeInBytes();
while (size > 0)
size = FileCacheService.instance.sizeInBytes();
}
catch (Throwable t)
{
t.printStackTrace();
hadError.set(true);
}
finally
{
finished.countDown();
}
}
});
}
finished.await();
assert !hadError.get();
}
}
@Test
public void testReadOnly() throws IOException
{

852
test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java Normal file
View File

@ -0,0 +1,852 @@
/**
* 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.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.RandomAccessReader;
import static org.junit.Assert.*;
public class BufferPoolTest
{
@Before
public void setUp()
{
BufferPool.MEMORY_USAGE_THRESHOLD = 8 * 1024L * 1024L;
BufferPool.DISABLED = false;
}
@After
public void cleanUp()
{
BufferPool.reset();
}
@Test
public void testGetPut() throws InterruptedException
{
final int size = RandomAccessReader.DEFAULT_BUFFER_SIZE;
ByteBuffer buffer = BufferPool.get(size);
assertNotNull(buffer);
assertEquals(size, buffer.capacity());
assertEquals(true, buffer.isDirect());
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, BufferPool.sizeInBytes());
BufferPool.put(buffer);
assertEquals(null, BufferPool.currentChunk());
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, BufferPool.sizeInBytes());
}
@Test
public void testPageAligned()
{
final int size = 1024;
for (int i = size;
i <= BufferPool.CHUNK_SIZE;
i += size)
{
checkPageAligned(i);
}
}
private void checkPageAligned(int size)
{
ByteBuffer buffer = BufferPool.get(size);
assertNotNull(buffer);
assertEquals(size, buffer.capacity());
assertTrue(buffer.isDirect());
long address = MemoryUtil.getAddress(buffer);
assertTrue((address % MemoryUtil.pageSize()) == 0);
BufferPool.put(buffer);
}
@Test
public void testDifferentSizes() throws InterruptedException
{
final int size1 = 1024;
final int size2 = 2048;
ByteBuffer buffer1 = BufferPool.get(size1);
assertNotNull(buffer1);
assertEquals(size1, buffer1.capacity());
ByteBuffer buffer2 = BufferPool.get(size2);
assertNotNull(buffer2);
assertEquals(size2, buffer2.capacity());
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, BufferPool.sizeInBytes());
BufferPool.put(buffer1);
BufferPool.put(buffer2);
assertEquals(null, BufferPool.currentChunk());
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, BufferPool.sizeInBytes());
}
@Test
public void testMaxMemoryExceededDirect()
{
boolean cur = BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED;
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = false;
requestDoubleMaxMemory();
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = cur;
}
@Test
public void testMaxMemoryExceededHeap()
{
boolean cur = BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED;
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = true;
requestDoubleMaxMemory();
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = cur;
}
@Test
public void testMaxMemoryExceeded_SameAsChunkSize()
{
BufferPool.MEMORY_USAGE_THRESHOLD = BufferPool.GlobalPool.MACRO_CHUNK_SIZE;
requestDoubleMaxMemory();
}
@Test
public void testMaxMemoryExceeded_SmallerThanChunkSize()
{
BufferPool.MEMORY_USAGE_THRESHOLD = BufferPool.GlobalPool.MACRO_CHUNK_SIZE / 2;
requestDoubleMaxMemory();
}
@Test
public void testRecycle()
{
requestUpToSize(RandomAccessReader.DEFAULT_BUFFER_SIZE, 3 * BufferPool.CHUNK_SIZE);
}
private void requestDoubleMaxMemory()
{
requestUpToSize(RandomAccessReader.DEFAULT_BUFFER_SIZE, (int)(2 * BufferPool.MEMORY_USAGE_THRESHOLD));
}
private void requestUpToSize(int bufferSize, int totalSize)
{
final int numBuffers = totalSize / bufferSize;
List<ByteBuffer> buffers = new ArrayList<>(numBuffers);
for (int i = 0; i < numBuffers; i++)
{
ByteBuffer buffer = BufferPool.get(bufferSize);
assertNotNull(buffer);
assertEquals(bufferSize, buffer.capacity());
if (BufferPool.sizeInBytes() > BufferPool.MEMORY_USAGE_THRESHOLD)
assertEquals(BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED, !buffer.isDirect());
buffers.add(buffer);
}
for (ByteBuffer buffer : buffers)
BufferPool.put(buffer);
}
@Test
public void testBigRequest()
{
final int size = BufferPool.CHUNK_SIZE + 1;
ByteBuffer buffer = BufferPool.get(size);
assertNotNull(buffer);
assertEquals(size, buffer.capacity());
BufferPool.put(buffer);
}
@Test
public void testFillUpChunks()
{
final int size = RandomAccessReader.DEFAULT_BUFFER_SIZE;
final int numBuffers = BufferPool.CHUNK_SIZE / size;
List<ByteBuffer> buffers1 = new ArrayList<>(numBuffers);
List<ByteBuffer> buffers2 = new ArrayList<>(numBuffers);
for (int i = 0; i < numBuffers; i++)
buffers1.add(BufferPool.get(size));
BufferPool.Chunk chunk1 = BufferPool.currentChunk();
assertNotNull(chunk1);
for (int i = 0; i < numBuffers; i++)
buffers2.add(BufferPool.get(size));
assertEquals(2, BufferPool.numChunks());
for (ByteBuffer buffer : buffers1)
BufferPool.put(buffer);
assertEquals(1, BufferPool.numChunks());
for (ByteBuffer buffer : buffers2)
BufferPool.put(buffer);
assertEquals(0, BufferPool.numChunks());
buffers2.clear();
}
@Test
public void testOutOfOrderFrees()
{
final int size = 4096;
final int maxFreeSlots = BufferPool.CHUNK_SIZE / size;
final int[] idxs = new int[maxFreeSlots];
for (int i = 0; i < maxFreeSlots; i++)
idxs[i] = i;
doTestFrees(size, maxFreeSlots, idxs);
}
@Test
public void testInOrderFrees()
{
final int size = 4096;
final int maxFreeSlots = BufferPool.CHUNK_SIZE / size;
final int[] idxs = new int[maxFreeSlots];
for (int i = 0; i < maxFreeSlots; i++)
idxs[i] = maxFreeSlots - 1 - i;
doTestFrees(size, maxFreeSlots, idxs);
}
@Test
public void testRandomFrees()
{
doTestRandomFrees(12345567878L);
BufferPool.reset();
doTestRandomFrees(20452249587L);
BufferPool.reset();
doTestRandomFrees(82457252948L);
BufferPool.reset();
doTestRandomFrees(98759284579L);
BufferPool.reset();
doTestRandomFrees(19475257244L);
}
private void doTestRandomFrees(long seed)
{
final int size = 4096;
final int maxFreeSlots = BufferPool.CHUNK_SIZE / size;
final int[] idxs = new int[maxFreeSlots];
for (int i = 0; i < maxFreeSlots; i++)
idxs[i] = maxFreeSlots - 1 - i;
Random rnd = new Random();
rnd.setSeed(seed);
for (int i = idxs.length - 1; i > 0; i--)
{
int idx = rnd.nextInt(i+1);
int v = idxs[idx];
idxs[idx] = idxs[i];
idxs[i] = v;
}
doTestFrees(size, maxFreeSlots, idxs);
}
private void doTestFrees(final int size, final int maxFreeSlots, final int[] toReleaseIdxs)
{
List<ByteBuffer> buffers = new ArrayList<>(maxFreeSlots);
for (int i = 0; i < maxFreeSlots; i++)
{
buffers.add(BufferPool.get(size));
}
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertFalse(chunk.isFree());
int freeSize = BufferPool.CHUNK_SIZE - maxFreeSlots * size;
assertEquals(freeSize, chunk.free());
for (int i : toReleaseIdxs)
{
ByteBuffer buffer = buffers.get(i);
assertNotNull(buffer);
assertEquals(size, buffer.capacity());
BufferPool.put(buffer);
freeSize += size;
if (freeSize == chunk.capacity())
assertEquals(0, chunk.free());
else
assertEquals(freeSize, chunk.free());
}
assertFalse(chunk.isFree());
}
@Test
public void testDifferentSizeBuffersOnOneChunk()
{
int[] sizes = new int[] {
5, 1024, 4096, 8, 16000, 78, 512, 256, 63, 55, 89, 90, 255, 32, 2048, 128
};
int sum = 0;
List<ByteBuffer> buffers = new ArrayList<>(sizes.length);
for (int i = 0; i < sizes.length; i++)
{
ByteBuffer buffer = BufferPool.get(sizes[i]);
assertNotNull(buffer);
assertTrue(buffer.capacity() >= sizes[i]);
buffers.add(buffer);
sum += BufferPool.currentChunk().roundUp(buffer.capacity());
}
// else the test will fail, adjust sizes as required
assertTrue(sum <= BufferPool.GlobalPool.MACRO_CHUNK_SIZE);
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
Random rnd = new Random();
rnd.setSeed(298347529L);
while (!buffers.isEmpty())
{
int index = rnd.nextInt(buffers.size());
ByteBuffer buffer = buffers.remove(index);
BufferPool.put(buffer);
}
assertEquals(null, BufferPool.currentChunk());
assertEquals(0, chunk.free());
}
@Test
public void testChunkExhausted()
{
final int size = BufferPool.CHUNK_SIZE / 64; // 1kbit
int[] sizes = new int[128];
Arrays.fill(sizes, size);
int sum = 0;
List<ByteBuffer> buffers = new ArrayList<>(sizes.length);
for (int i = 0; i < sizes.length; i++)
{
ByteBuffer buffer = BufferPool.get(sizes[i]);
assertNotNull(buffer);
assertTrue(buffer.capacity() >= sizes[i]);
buffers.add(buffer);
sum += buffer.capacity();
}
// else the test will fail, adjust sizes as required
assertTrue(sum <= BufferPool.GlobalPool.MACRO_CHUNK_SIZE);
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
for (int i = 0; i < sizes.length; i++)
{
BufferPool.put(buffers.get(i));
}
assertEquals(null, BufferPool.currentChunk());
assertEquals(0, chunk.free());
}
@Test
public void testCompactIfOutOfCapacity()
{
final int size = 4096;
final int numBuffersInChunk = BufferPool.GlobalPool.MACRO_CHUNK_SIZE / size;
List<ByteBuffer> buffers = new ArrayList<>(numBuffersInChunk);
Set<Long> addresses = new HashSet<>(numBuffersInChunk);
for (int i = 0; i < numBuffersInChunk; i++)
{
ByteBuffer buffer = BufferPool.get(size);
buffers.add(buffer);
addresses.add(MemoryUtil.getAddress(buffer));
}
for (int i = numBuffersInChunk - 1; i >= 0; i--)
BufferPool.put(buffers.get(i));
buffers.clear();
for (int i = 0; i < numBuffersInChunk; i++)
{
ByteBuffer buffer = BufferPool.get(size);
assertNotNull(buffer);
assertEquals(size, buffer.capacity());
addresses.remove(MemoryUtil.getAddress(buffer));
buffers.add(buffer);
}
assertTrue(addresses.isEmpty()); // all 5 released buffers were used
for (ByteBuffer buffer : buffers)
BufferPool.put(buffer);
}
@Test
public void testHeapBuffer()
{
ByteBuffer buffer = BufferPool.get(1024, BufferType.ON_HEAP);
assertNotNull(buffer);
assertEquals(1024, buffer.capacity());
assertFalse(buffer.isDirect());
assertNotNull(buffer.array());
BufferPool.put(buffer);
}
@Test
public void testSingleBufferOneChunk()
{
checkBuffer(0);
checkBuffer(1);
checkBuffer(2);
checkBuffer(4);
checkBuffer(5);
checkBuffer(8);
checkBuffer(16);
checkBuffer(32);
checkBuffer(64);
checkBuffer(65);
checkBuffer(127);
checkBuffer(128);
checkBuffer(129);
checkBuffer(255);
checkBuffer(256);
checkBuffer(512);
checkBuffer(1024);
checkBuffer(2048);
checkBuffer(4096);
checkBuffer(8192);
checkBuffer(16384);
checkBuffer(16385);
checkBuffer(32767);
checkBuffer(32768);
checkBuffer(32769);
checkBuffer(33172);
checkBuffer(33553);
checkBuffer(36000);
checkBuffer(65535);
checkBuffer(65536);
checkBuffer(65537);
}
private void checkBuffer(int size)
{
ByteBuffer buffer = BufferPool.get(size);
assertEquals(size, buffer.capacity());
if (size > 0 && size < BufferPool.CHUNK_SIZE)
{
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
assertEquals(chunk.capacity(), chunk.free() + chunk.roundUp(size));
}
BufferPool.put(buffer);
}
@Test
public void testMultipleBuffersOneChunk()
{
checkBuffers(32768, 33553);
checkBuffers(32768, 32768);
checkBuffers(48450, 33172);
checkBuffers(32768, 15682, 33172);
}
private void checkBuffers(int ... sizes)
{
List<ByteBuffer> buffers = new ArrayList<>(sizes.length);
for (int size : sizes)
{
ByteBuffer buffer = BufferPool.get(size);
assertEquals(size, buffer.capacity());
buffers.add(buffer);
}
for (ByteBuffer buffer : buffers)
BufferPool.put(buffer);
}
@Test
public void testBuffersWithGivenSlots()
{
checkBufferWithGivenSlots(21241, (-1L << 27) ^ (1L << 40));
}
private void checkBufferWithGivenSlots(int size, long freeSlots)
{
//first allocate to make sure there is a chunk
ByteBuffer buffer = BufferPool.get(size);
// now get the current chunk and override the free slots mask
BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
long oldFreeSlots = chunk.setFreeSlots(freeSlots);
// now check we can still get the buffer with the free slots mask changed
ByteBuffer buffer2 = BufferPool.get(size);
assertEquals(size, buffer.capacity());
BufferPool.put(buffer2);
// reset the free slots
chunk.setFreeSlots(oldFreeSlots);
BufferPool.put(buffer);
}
@Test
public void testZeroSizeRequest()
{
ByteBuffer buffer = BufferPool.get(0);
assertNotNull(buffer);
assertEquals(0, buffer.capacity());
BufferPool.put(buffer);
}
@Test(expected = IllegalArgumentException.class)
public void testNegativeSizeRequest()
{
BufferPool.get(-1);
}
@Test
public void testBufferPoolDisabled()
{
BufferPool.DISABLED = true;
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = true;
ByteBuffer buffer = BufferPool.get(1024);
assertEquals(0, BufferPool.numChunks());
assertNotNull(buffer);
assertEquals(1024, buffer.capacity());
assertFalse(buffer.isDirect());
assertNotNull(buffer.array());
BufferPool.put(buffer);
assertEquals(0, BufferPool.numChunks());
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = false;
buffer = BufferPool.get(1024);
assertEquals(0, BufferPool.numChunks());
assertNotNull(buffer);
assertEquals(1024, buffer.capacity());
assertTrue(buffer.isDirect());
BufferPool.put(buffer);
assertEquals(0, BufferPool.numChunks());
// clean-up
BufferPool.DISABLED = false;
BufferPool.ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = true;
}
@Test
public void testMT_SameSizeImmediateReturn() throws InterruptedException
{
checkMultipleThreads(40, 1, true, RandomAccessReader.DEFAULT_BUFFER_SIZE);
}
@Test
public void testMT_SameSizePostponedReturn() throws InterruptedException
{
checkMultipleThreads(40, 1, false, RandomAccessReader.DEFAULT_BUFFER_SIZE);
}
@Test
public void testMT_TwoSizesOneBufferImmediateReturn() throws InterruptedException
{
checkMultipleThreads(40, 1, true, 1024, 2048);
}
@Test
public void testMT_TwoSizesOneBufferPostponedReturn() throws InterruptedException
{
checkMultipleThreads(40, 1, false, 1024, 2048);
}
@Test
public void testMT_TwoSizesTwoBuffersImmediateReturn() throws InterruptedException
{
checkMultipleThreads(40, 2, true, 1024, 2048);
}
@Test
public void testMT_TwoSizesTwoBuffersPostponedReturn() throws InterruptedException
{
checkMultipleThreads(40, 2, false, 1024, 2048);
}
@Test
public void testMT_MultipleSizesOneBufferImmediateReturn() throws InterruptedException
{
checkMultipleThreads(40,
1,
true,
1024,
2048,
3072,
4096,
5120);
}
@Test
public void testMT_MultipleSizesOneBufferPostponedReturn() throws InterruptedException
{
checkMultipleThreads(40,
1,
false,
1024,
2048,
3072,
4096,
5120);
}
@Test
public void testMT_MultipleSizesMultipleBuffersImmediateReturn() throws InterruptedException
{
checkMultipleThreads(40,
4,
true,
1024,
2048,
3072,
4096,
5120);
}
@Test
public void testMT_MultipleSizesMultipleBuffersPostponedReturn() throws InterruptedException
{
checkMultipleThreads(40,
3,
false,
1024,
2048,
3072,
4096,
5120);
}
private void checkMultipleThreads(int threadCount, int numBuffersPerThread, final boolean returnImmediately, final int ... sizes) throws InterruptedException
{
ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
final CountDownLatch finished = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++)
{
final int[] threadSizes = new int[numBuffersPerThread];
for (int j = 0; j < threadSizes.length; j++)
threadSizes[j] = sizes[(i * numBuffersPerThread + j) % sizes.length];
final Random rand = new Random();
executorService.submit(new Runnable()
{
@Override
public void run()
{
try
{
Thread.sleep(rand.nextInt(3));
List<ByteBuffer> toBeReturned = new ArrayList<ByteBuffer>(threadSizes.length);
for (int j = 0; j < threadSizes.length; j++)
{
ByteBuffer buffer = BufferPool.get(threadSizes[j]);
assertNotNull(buffer);
assertEquals(threadSizes[j], buffer.capacity());
for (int i = 0; i < 10; i++)
buffer.putInt(i);
buffer.rewind();
Thread.sleep(rand.nextInt(3));
for (int i = 0; i < 10; i++)
assertEquals(i, buffer.getInt());
if (returnImmediately)
BufferPool.put(buffer);
else
toBeReturned.add(buffer);
assertTrue(BufferPool.sizeInBytes() > 0);
}
Thread.sleep(rand.nextInt(3));
for (ByteBuffer buffer : toBeReturned)
BufferPool.put(buffer);
}
catch (Exception ex)
{
ex.printStackTrace();
fail(ex.getMessage());
}
finally
{
finished.countDown();
}
}
});
}
finished.await();
assertEquals(0, executorService.shutdownNow().size());
// Make sure thread local storage gets GC-ed
for (int i = 0; i < 5; i++)
{
System.gc();
Thread.sleep(100);
}
}
@Ignore
public void testMultipleThreadsReleaseSameBuffer() throws InterruptedException
{
doMultipleThreadsReleaseBuffers(45, 4096);
}
@Ignore
public void testMultipleThreadsReleaseDifferentBuffer() throws InterruptedException
{
doMultipleThreadsReleaseBuffers(45, 4096, 8192);
}
private void doMultipleThreadsReleaseBuffers(final int threadCount, final int ... sizes) throws InterruptedException
{
final ByteBuffer[] buffers = new ByteBuffer[sizes.length];
int sum = 0;
for (int i = 0; i < sizes.length; i++)
{
buffers[i] = BufferPool.get(sizes[i]);
assertNotNull(buffers[i]);
assertEquals(sizes[i], buffers[i].capacity());
sum += BufferPool.currentChunk().roundUp(buffers[i].capacity());
}
final BufferPool.Chunk chunk = BufferPool.currentChunk();
assertNotNull(chunk);
assertFalse(chunk.isFree());
// if we use multiple chunks the test will fail, adjust sizes accordingly
assertTrue(sum < BufferPool.GlobalPool.MACRO_CHUNK_SIZE);
ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
final CountDownLatch finished = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++)
{
final int idx = i % sizes.length;
final ByteBuffer buffer = buffers[idx];
executorService.submit(new Runnable()
{
@Override
public void run()
{
try
{
assertNotSame(chunk, BufferPool.currentChunk());
BufferPool.put(buffer);
}
catch (AssertionError ex)
{ //this is expected if we release a buffer more than once
ex.printStackTrace();
}
catch (Throwable t)
{
t.printStackTrace();
fail(t.getMessage());
}
finally
{
finished.countDown();
}
}
});
}
finished.await();
assertEquals(0, executorService.shutdownNow().size());
executorService = null;
// Make sure thread local storage gets GC-ed
System.gc();
System.gc();
System.gc();
assertTrue(BufferPool.currentChunk().isFree());
//make sure the main thread can still allocate buffers
ByteBuffer buffer = BufferPool.get(sizes[0]);
assertNotNull(buffer);
assertEquals(sizes[0], buffer.capacity());
BufferPool.put(buffer);
}
}

2
tools/stress/src/org/apache/cassandra/stress/generate/Seed.java
View File

@ -20,7 +20,7 @@ package org.apache.cassandra.stress.generate;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.cassandra.stress.util.DynamicList;
import org.apache.cassandra.utils.DynamicList;
public class Seed implements Comparable<Seed>
{

6
tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java
View File

@ -25,7 +25,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.DynamicList;
import org.apache.cassandra.utils.LockedDynamicList;
public class SeedManager
{
@ -34,7 +34,7 @@ public class SeedManager
final Generator writes;
final Generator reads;
final ConcurrentHashMap<Long, Seed> managing = new ConcurrentHashMap<>();
final DynamicList<Seed> sampleFrom;
final LockedDynamicList<Seed> sampleFrom;
final Distribution sample;
final long sampleOffset;
final int sampleSize;
@ -69,7 +69,7 @@ public class SeedManager
long sampleSize = 1 + Math.max(sample.minValue(), sample.maxValue()) - sampleOffset;
if (sampleOffset < 0 || sampleSize > Integer.MAX_VALUE)
throw new IllegalArgumentException("sample range is invalid");
this.sampleFrom = new DynamicList<>((int) sampleSize);
this.sampleFrom = new LockedDynamicList<>((int) sampleSize);
this.sample = DistributionInverted.invert(sample);
this.sampleSize = (int) sampleSize;
this.updateSampleImmediately = visits.average() > 1;

2
tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java
View File

@ -24,7 +24,7 @@ import java.nio.ByteBuffer;
import java.util.Arrays;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.stress.generate.FasterRandom;
import org.apache.cassandra.utils.FasterRandom;
public class Bytes extends Generator<ByteBuffer>
{

2
tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java
View File

@ -21,7 +21,7 @@
package org.apache.cassandra.stress.generate.values;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.stress.generate.FasterRandom;
import org.apache.cassandra.utils.FasterRandom;
public class Strings extends Generator<String>
{