diff --git a/CHANGES.txt b/CHANGES.txt index 061bd5ffb3..144252d73c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index ea22e01c4e..4593a7e757 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -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 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 269f5777c6..e5891ab0d7 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 9e47da79d4..b5ee921c0a 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; diff --git a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java index e6ac60abda..ac120b4c87 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java @@ -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 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() diff --git a/src/java/org/apache/cassandra/io/compress/CompressedThrottledReader.java b/src/java/org/apache/cassandra/io/compress/CompressedThrottledReader.java index a29129c2bd..ea5edafb44 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedThrottledReader.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedThrottledReader.java @@ -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); } } diff --git a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java index 4d1c6630c3..5de2452fb1 100644 --- a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java @@ -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 implements CloseableIterator { - 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 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) diff --git a/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java b/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java index a1fda57fac..e5f61b2055 100644 --- a/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/ReducingKeyIterator.java @@ -33,42 +33,55 @@ import org.apache.cassandra.utils.MergeIterator; */ public class ReducingKeyIterator implements CloseableIterator { - private final IMergeIterator mi; + private final ArrayList iters; + private IMergeIterator mi; public ReducingKeyIterator(Collection sstables) { - ArrayList iters = new ArrayList(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() + } + + private void maybeInit() + { + if (mi == null) { - DecoratedKey reduced = null; - - @Override - public boolean trivialReduceIsTrivial() + mi = MergeIterator.get(iters, DecoratedKey.comparator, new MergeIterator.Reducer() { - 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 iter : mi.iterators()) { @@ -79,6 +92,8 @@ public class ReducingKeyIterator implements CloseableIterator public long getBytesRead() { + maybeInit(); + long m = 0; for (Iterator iter : mi.iterators()) { @@ -87,18 +102,15 @@ public class ReducingKeyIterator implements CloseableIterator return m; } - public String getTaskType() - { - return "Secondary index build"; - } - public boolean hasNext() { + maybeInit(); return mi.hasNext(); } public DecoratedKey next() { + maybeInit(); return mi.next(); } diff --git a/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java deleted file mode 100644 index b623e54daf..0000000000 --- a/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java +++ /dev/null @@ -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); - } - } -} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java deleted file mode 100644 index fdc4f61ee3..0000000000 --- a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java +++ /dev/null @@ -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 chunkSegments; - - public CompressedPoolingSegmentedFile(ChannelProxy channel, CompressionMetadata metadata) - { - this(channel, metadata, CompressedSegmentedFile.createMappedSegments(channel, metadata)); - } - - private CompressedPoolingSegmentedFile(ChannelProxy channel, CompressionMetadata metadata, TreeMap 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 chunkSegments() - { - return chunkSegments; - } - - protected static final class Cleanup extends PoolingSegmentedFile.Cleanup - { - final CompressionMetadata metadata; - final TreeMap chunkSegments; - protected Cleanup(ChannelProxy channel, CompressionMetadata metadata, TreeMap 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); - } -} diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java index 25669528e1..35e14199b7 100644 --- a/src/java/org/apache/cassandra/io/util/FileUtils.java +++ b/src/java/org/apache/cassandra/io/util/FileUtils.java @@ -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) diff --git a/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java deleted file mode 100644 index a5fa20bebe..0000000000 --- a/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java +++ /dev/null @@ -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); - } -} diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index 278f55cb5a..fef206e328 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -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() diff --git a/src/java/org/apache/cassandra/io/util/SegmentedFile.java b/src/java/org/apache/cassandra/io/util/SegmentedFile.java index edbd742599..fbaa77c112 100644 --- a/src/java/org/apache/cassandra/io/util/SegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/SegmentedFile.java @@ -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); } /** diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java index f3268a2d6a..e76add54a7 100644 --- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java @@ -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, diff --git a/src/java/org/apache/cassandra/io/util/ThrottledReader.java b/src/java/org/apache/cassandra/io/util/ThrottledReader.java index f72598464d..ea21355f0d 100644 --- a/src/java/org/apache/cassandra/io/util/ThrottledReader.java +++ b/src/java/org/apache/cassandra/io/util/ThrottledReader.java @@ -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); } } diff --git a/src/java/org/apache/cassandra/metrics/FileCacheMetrics.java b/src/java/org/apache/cassandra/metrics/BufferPoolMetrics.java similarity index 61% rename from src/java/org/apache/cassandra/metrics/FileCacheMetrics.java rename to src/java/org/apache/cassandra/metrics/BufferPoolMetrics.java index c240c0389b..107717dfaa 100644 --- a/src/java/org/apache/cassandra/metrics/FileCacheMetrics.java +++ b/src/java/org/apache/cassandra/metrics/BufferPoolMetrics.java @@ -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 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 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() { public Long getValue() { - return FileCacheService.instance.sizeInBytes(); + return BufferPool.sizeInBytes(); } }); } diff --git a/src/java/org/apache/cassandra/service/FileCacheService.java b/src/java/org/apache/cassandra/service/FileCacheService.java deleted file mode 100644 index 1e123101bf..0000000000 --- a/src/java/org/apache/cassandra/service/FileCacheService.java +++ /dev/null @@ -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 cacheForPathCreator = new Callable() - { - @Override - public CacheBucket call() - { - return new CacheBucket(); - } - }; - - private static final AtomicInteger memoryUsage = new AtomicInteger(); - - private final Cache cache; - private final FileCacheMetrics metrics = new FileCacheMetrics(); - - private static final class CacheBucket - { - final ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>(); - volatile boolean discarded = false; - } - - protected FileCacheService() - { - RemovalListener onRemove = new RemovalListener() - { - @Override - public void onRemoval(RemovalNotification 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 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; - } -} diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java index 144980cecb..6b247073be 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java @@ -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. diff --git a/tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java b/src/java/org/apache/cassandra/utils/DynamicList.java similarity index 65% rename from tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java rename to src/java/org/apache/cassandra/utils/DynamicList.java index ee040635f9..fc3d523d57 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java +++ b/src/java/org/apache/cassandra/utils/DynamicList.java @@ -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 } } - private final ReadWriteLock lock = new ReentrantReadWriteLock(); private final int maxHeight; private final Node head; private int size; @@ -115,112 +110,92 @@ public class DynamicList public Node append(E value, int maxSize) { Node newTail = new Node<>(randomLevel(), value); + if (size >= maxSize) + return null; + size++; - lock.writeLock().lock(); - try + Node tail = head; + for (int i = maxHeight - 1 ; i >= newTail.height() ; i--) { - if (size >= maxSize) - return null; - size++; - - Node tail = head; - for (int i = maxHeight - 1 ; i >= newTail.height() ; i--) - { - Node next; - while ((next = tail.next(i)) != null) - tail = next; - tail.size[i]++; - } - - for (int i = newTail.height() - 1 ; i >= 0 ; i--) - { - Node next; - while ((next = tail.next(i)) != null) - tail = next; - tail.setNext(i, newTail); - newTail.setPrev(i, tail); - } - - return newTail; + Node 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 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 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 prev = node.prev(i); - Node 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 prev = node.prev(i); + Node 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 finger = head; - for (int i = maxHeight - 1 ; i >= 0 ; i--) + index++; + int c = 0; + Node 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 diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java b/src/java/org/apache/cassandra/utils/FasterRandom.java similarity index 98% rename from tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java rename to src/java/org/apache/cassandra/utils/FasterRandom.java index 455fec480c..8235e3b87b 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java +++ b/src/java/org/apache/cassandra/utils/FasterRandom.java @@ -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; diff --git a/src/java/org/apache/cassandra/utils/LockedDynamicList.java b/src/java/org/apache/cassandra/utils/LockedDynamicList.java new file mode 100644 index 0000000000..cbda7e573d --- /dev/null +++ b/src/java/org/apache/cassandra/utils/LockedDynamicList.java @@ -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 extends DynamicList +{ + + 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 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 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(); + } + } +} diff --git a/src/java/org/apache/cassandra/utils/concurrent/Ref.java b/src/java/org/apache/cassandra/utils/concurrent/Ref.java index 3df3ea452b..6825cd9a3d 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Ref.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Ref.java @@ -282,7 +282,8 @@ public final class Ref implements RefCounted globallyExtant.remove(this); try { - tidy.tidy(); + if (tidy != null) + tidy.tidy(); } catch (Throwable t) { @@ -299,7 +300,9 @@ public final class Ref implements RefCounted 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); } } diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java new file mode 100644 index 0000000000..0731301961 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -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 = new ThreadLocal() { + @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 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 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 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 + { + private final Chunk[] chunks; + public LocalPoolRef(LocalPool localPool, ReferenceQueue 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 localPoolReferences = new ConcurrentLinkedQueue<>(); + + private static final ReferenceQueue 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 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) 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) 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; + } +} diff --git a/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java b/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java index d2b2879249..5671fa95df 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java +++ b/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java @@ -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) diff --git a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java new file mode 100644 index 0000000000..17ac5693e6 --- /dev/null +++ b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java @@ -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 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[] 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> 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 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 shareFrom = sharedRecycle[threadIdx]; + final DynamicList checks = new DynamicList<>((int) Math.max(1, targetSize / (1 << 10))); + final SPSCQueue 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 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 r : ret) + assertTrue(r.get()); + + System.out.println(String.format("%s - finished.", + dateFormat.format(new Date()))); + } + + static abstract class TestUntil implements Callable + { + 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 + { + static final class Node + { + volatile Node next; + final V value; + Node(V value) + { + this.value = value; + } + } + + private volatile boolean exhausted = true; + Node head = new Node<>(null); + Node tail = head; + + void add(V value) + { + exhausted = false; + tail = tail.next = new Node<>(value); + } + + V poll() + { + Node next = head.next; + if (next == null) + { + // this is racey, but good enough for our purposes + exhausted = true; + return null; + } + head = next; + return next.value; + } + } + +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/io/compress/CompressorTest.java b/test/unit/org/apache/cassandra/io/compress/CompressorTest.java index 6018cc7e4b..8ff0074304 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressorTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressorTest.java @@ -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 { diff --git a/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java b/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java index 0c1583df16..3c9aa0e2af 100644 --- a/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java +++ b/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java @@ -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() { @@ -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() { @@ -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 { diff --git a/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java b/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java new file mode 100644 index 0000000000..208cd32d6a --- /dev/null +++ b/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java @@ -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 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 buffers1 = new ArrayList<>(numBuffers); + List 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 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 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 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 buffers = new ArrayList<>(numBuffersInChunk); + Set 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 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 toBeReturned = new ArrayList(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); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java index 9e2e65bd57..243ac30f5b 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java @@ -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 { diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java b/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java index 071d888b9d..5020b45d89 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java @@ -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 managing = new ConcurrentHashMap<>(); - final DynamicList sampleFrom; + final LockedDynamicList 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; diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java index 3c15c8790d..c247e480fc 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java @@ -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 { diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java index b58fee28c4..db78eb7035 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java @@ -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 {